consumer: sort fallback DML before flush (#5824)

[ti-chi-bot]

This is an automated cherry-pick of #5824

What problem does this PR solve?

Issue Number: close #5713

What is changed and how it works?

The consumer no longer drops fallback DML during append if the current message commit-ts is greater than global watermark. It keeps the DML in EventsGroup, then sorts and flushes it when a DDL or watermark arrives.

In the BAB case, if the consumer receives B first and later receives replayed A B, A is no longer ignored because its commitTs is smaller than the high watermark. It is buffered, then sorted into A B before writing.

Check List

Tests

• Unit test
• Integration test

Questions

Will it cause performance regression or break compatibility?

Do you need to update user documentation, design documentation or monitoring documentation?

Release note

Please refer to [Release Notes Language Style Guide](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/release-notes-style-guide.html) to write a quality release note.

If you don't think this PR needs a release note then fill it with `None`.

Summary by CodeRabbit

Summary by CodeRabbit

• Bug Fixes

  • Improved event processing to reliably order out-of-sequence data changes by commit timestamp before flushing.
  • Ensured messages are retained and processed consistently across Kafka, Pulsar, and storage consumers.
  • Improved cleanup of processed events to prevent stale data from remaining in memory.

• Tests

  • Added coverage for out-of-order messages, stable ordering of equal timestamps, and complete event retrieval.

[ti-chi-bot]

@wk989898 This PR has conflicts, I have hold it.
Please resolve them or ask others to resolve them, then comment /unhold to remove the hold label.

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on base/target branches other than the default branch.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: e58aa6cd-8f1a-4f05-8f3a-930838d9c624

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[ti-chi-bot]

@ti-chi-bot: ## If you want to know how to resolve it, please read the guide in TiDB Dev Guide.

Details

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the ti-community-infra/tichi repository.

[ti-chi-bot]

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by:
Once this PR has been reviewed and has the lgtm label, please ask for approval from wk989898. For more information see the Code Review Process.
Please ensure that each of them provides their approval before proceeding.

The full list of commands accepted by this bot can be found here.

Details

Needs approval from an approver in each of these files:

• OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[ti-chi-bot]

Cherry-pick conflicts appear resolved; removing the do-not-merge/hold label.

[wk989898]

/test next-gen