kafka: decouple batch size from Kafka message size limit (#5420)

[ti-chi-bot]

This is an automated cherry-pick of #5420

What problem does this PR solve?

Issue Number: close #1405

Kafka sink currently uses the changefeed max-message-bytes setting for two different purposes:

• deciding when TiCDC should split a batch of row events;
• limiting the final Kafka message and deciding when to invoke large-message-handle.

This means TiCDC can reject a message even when Kafka can accept it. For example, if the changefeed has max-message-bytes=10 MiB, the Kafka topic allows 20 MiB, and one encoded row event is 12 MiB, TiCDC still returns ErrMessageTooLarge or invokes large-message-handle. Operators must then pause the changefeed and update its configuration in addition to updating Kafka.

Whether a message is too large should be determined by Kafka's message-size limit. The changefeed setting should only control TiCDC's batching behavior.

What is changed and how it works?

This PR separates the Kafka message limit from the TiCDC batch threshold. Define:

• T: the changefeed max-message-bytes setting;
• K: the Kafka topic max.message.bytes for an existing topic, or the broker message.max.bytes for a topic that does not exist yet;
• producer message limit: K;
• batch threshold: min(T, K).

The implementation:

• preserves T as MaxBatchedBytes before reading Kafka metadata;
• updates MaxMessageBytes to K and uses it for both the producer and the codec's final message-size check;
• passes both limits to codec configuration;
• uses MaxBatchedBytes only when deciding whether to add another row to the current batch;
• invokes large-message-handle only when the original encoded message exceeds MaxMessageBytes (K), and performs a final size check after handling;
• falls back to the configured T as the producer limit if the Kafka topic or broker size configuration cannot be read.

As a result:

• a single message larger than T but not larger than K is sent directly;
• multiple rows are still split according to min(T, K);
• a message larger than K invokes the configured large-message-handle, or returns ErrMessageTooLarge when it still cannot fit;
• after the Kafka size limit is increased, a restarted Kafka sink reads the new limit and can recover without changing the changefeed max-message-bytes setting.

This PR does not introduce a separate model for Sarama's request-size limit or cap K using sarama.MaxRequestSize. Sarama request sizing remains outside the scope of this change.

Non-Kafka sinks pass the same value for the final message limit and batch threshold, preserving their existing behavior.

Check List

Tests

• Unit test
  • Kafka topic and broker limits are propagated to the producer and codec configuration.
  • The batch threshold remains min(T, K).
  • Open Protocol splits batches at the batch threshold while allowing one row larger than that threshold when it is within the final message limit.
  • Open Protocol, Canal JSON, and Simple Protocol enforce the final message limit and large-message handling boundary.
• Integration test
  • kafka_big_messages covers Canal JSON, Open Protocol, Simple JSON, Simple Avro, and Avro.
  • Each case first sends a message larger than Kafka's current topic limit and verifies ErrMessageTooLarge.
  • The test then only increases the topic max.message.bytes, without updating, pausing, or manually resuming the changefeed, and verifies that synchronization automatically recovers and passes sync-diff.
  • Existing claim-check and handle-key-only cases explicitly configure the Kafka topic limit so they continue to exercise the large-message handler.

Questions

Will it cause performance regression or break compatibility?

No performance regression or compatibility break is expected.

The batching behavior of max-message-bytes is preserved. This PR fixes the previous conflation of the batch threshold and Kafka's actual message-size limit: when T < message size <= K, Kafka can accept the message, so TiCDC sends it instead of incorrectly returning ErrMessageTooLarge or invoking large-message-handle. If K < T, the batch threshold is reduced to K so TiCDC does not intentionally build an unsendable batch.

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

Yes. User documentation should clarify that max-message-bytes controls Kafka batching, while Kafka topic or broker configuration controls the final message limit. No monitoring update is required.

Release note

Decouple the Kafka sink batch threshold from Kafka's message-size limit, so increasing the Kafka topic or broker limit can unblock a large row event without changing the TiCDC changefeed configuration.

[ti-chi-bot]

@3AceShowHand 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.

[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.

[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: f14b9a60-5a06-46dd-8625-6ffa84af38a9

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

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

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]

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

[3AceShowHand]

/test all

[3AceShowHand]

/test all

[3AceShowHand]

/test all

[3AceShowHand]

/test all

[ti-chi-bot]

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: 3AceShowHand, nongfushanquan

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

The pull request process is described here

Details

Needs approval from an approver in each of these files:

• OWNERS [3AceShowHand,nongfushanquan]
• pkg/config/OWNERS [nongfushanquan]

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