kafka: decouple batch size from Kafka message size limit - #5420
Conversation
ti-chi-bot
Bot
added
the
release-note
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughThis change separates per-message and batch message limits in codec configuration, Kafka options, sink encoder wiring, and protocol encoders. Kafka derives limits from topic or broker settings, while Open, Simple, and Canal encoders apply distinct batching and oversized-message checks. ChangesSeparate message and batch limits
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant KafkaOptions
participant SinkHelper
participant CodecConfig
participant Encoder
KafkaOptions->>KafkaOptions: derive message and batch limits
KafkaOptions->>SinkHelper: pass both limits
SinkHelper->>CodecConfig: configure MaxMessageBytes and MaxBatchedBytes
CodecConfig->>Encoder: provide effective limits
Encoder->>Encoder: batch by batch limit and enforce message limit
Possibly related PRs
Suggested labels: Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
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. Comment |
ti-chi-bot
Bot
added
the
size/XL
|
/test all |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request decouples the final encoded message size limit (MaxMessageBytes) from the batch splitting and large-message threshold (MaxBatchMessageBytes) across various TiCDC sinks and codecs (including Kafka, Pulsar, Cloud Storage, Avro, Canal JSON, Open Protocol, and Simple). This allows for more granular control over message batching and limits. The feedback suggests simplifying duplicate validation logic in the simple encoder and adding configuration validation to ensure MaxBatchMessageBytes does not exceed MaxMessageBytes.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@pkg/sink/codec/simple/encoder_test.go`:
- Around line 1595-1611: The TestDMLLargerThanBatchLimit test sets
MaxBatchMessageBytes to 50 but never verifies that the actual encoded message
payload exceeds this threshold, making the test non-deterministic and unable to
guarantee it exercises the intended code path. After calling enc.Build() to
obtain the messages, add an assertion that explicitly checks the payload size of
messages[0] is greater than the MaxBatchMessageBytes limit to ensure the test
fixture is sufficiently large and the behavior remains deterministic.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 5d97615e-8665-4e5f-9988-6942075570a7
📒 Files selected for processing (18)
downstreamadapter/sink/cloudstorage/encoder_group_test.godownstreamadapter/sink/cloudstorage/sink.godownstreamadapter/sink/helper/helper.godownstreamadapter/sink/kafka/helper.godownstreamadapter/sink/kafka/sink_test.godownstreamadapter/sink/pulsar/helper.gopkg/sink/codec/avro/encoder.gopkg/sink/codec/canal/canal_json_encoder.gopkg/sink/codec/canal/canal_json_encoder_test.gopkg/sink/codec/canal/canal_json_txn_encoder.gopkg/sink/codec/common/config.gopkg/sink/codec/open/encoder.gopkg/sink/codec/open/encoder_test.gopkg/sink/codec/simple/encoder.gopkg/sink/codec/simple/encoder_test.gopkg/sink/kafka/options.gopkg/sink/kafka/options_test.gopkg/sink/kafka/sarama_config.go
ti-chi-bot
Bot
added
size/XXL
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
downstreamadapter/sink/kafka/sink.go (1)
90-110: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winReorder
GetEncoderConfigafterNewSaramaFactoryto ensure accurate validation.In
newKafkaSinkComponent,GetEncoderConfigis instantiated afterkafka.NewSaramaFactory. InVerify, it is instantiated beforehand. Since the PR introduces logic that derives producer limits based on Kafka broker/topic configurations, any mutation tooptions.MaxMessageBytesduringNewSaramaFactorywill not be reflected inVerify'sencoderConfig. This could causeVerifyto bypass limit validation checks (e.g., if the user-configured batch limit exceeds the derived producer limit) and return false positives.Aligning the order with
newKafkaSinkComponentresolves this.💡 Proposed reordering
- encoderConfig, err := helper.GetEncoderConfig( - changefeedID, uri, protocol, sinkConfig, - options.MaxMessageBytes, options.MaxBatchedBytes, - ) - if err != nil { - return errors.Trace(err) - } - isAvroLike := protocol == config.ProtocolAvro || protocol == config.ProtocolDebeziumAvro if _, err = eventrouter.NewEventRouter(sinkConfig, topic, false, isAvroLike); err != nil { return errors.Trace(err) } if _, err = columnselector.New(sinkConfig); err != nil { return errors.Trace(err) } factory, err := kafka.NewSaramaFactory(ctx, options, changefeedID) if err != nil { return errors.WrapError(errors.ErrKafkaNewProducer, err) } + + encoderConfig, err := helper.GetEncoderConfig( + changefeedID, uri, protocol, sinkConfig, + options.MaxMessageBytes, options.MaxBatchedBytes, + ) + if err != nil { + return errors.Trace(err) + }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@downstreamadapter/sink/kafka/sink.go` around lines 90 - 110, In Verify, move the GetEncoderConfig call to execute after kafka.NewSaramaFactory so it uses any producer-limit mutations applied during factory creation. Keep the existing encoder-config error handling and subsequent validation unchanged, ensuring derived Kafka limits are reflected before validation.
🧹 Nitpick comments (1)
pkg/sink/codec/simple/encoder.go (1)
153-162: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRemove DDL message size check for consistency.
The size check against
MaxMessageBytesfor DDL and Checkpoint events was intentionally removed in the Open and Canal JSON encoders to let Kafka enforce the producer limit. It was also removed from this file'sEncodeCheckpointEventbut was left behind here inEncodeDDLEvent.Please consider removing it to keep the encoder behaviors consistent.
♻️ Proposed fix
result := common.NewMsg(nil, value) - if result.Length() > e.config.MaxMessageBytes { - log.Error("DDL message is too large for simple", - zap.Int("maxMessageBytes", e.config.MaxMessageBytes), - zap.Int("length", result.Length()), - zap.String("table", event.GetTargetTableName())) - return nil, errors.ErrMessageTooLarge.GenWithStackByArgs(event.GetTargetTableName(), result.Length(), e.config.MaxMessageBytes) - } return result, nil }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/sink/codec/simple/encoder.go` around lines 153 - 162, Remove the MaxMessageBytes length validation, oversized-message logging, and ErrMessageTooLarge return from EncodeDDLEvent so DDL messages are returned directly like checkpoint events and the other encoders.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@downstreamadapter/sink/kafka/sink.go`:
- Around line 90-110: In Verify, move the GetEncoderConfig call to execute after
kafka.NewSaramaFactory so it uses any producer-limit mutations applied during
factory creation. Keep the existing encoder-config error handling and subsequent
validation unchanged, ensuring derived Kafka limits are reflected before
validation.
---
Nitpick comments:
In `@pkg/sink/codec/simple/encoder.go`:
- Around line 153-162: Remove the MaxMessageBytes length validation,
oversized-message logging, and ErrMessageTooLarge return from EncodeDDLEvent so
DDL messages are returned directly like checkpoint events and the other
encoders.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: ac9d7ebc-bff8-4135-9171-80ff09c6b3a1
📒 Files selected for processing (9)
downstreamadapter/sink/cloudstorage/sink.godownstreamadapter/sink/kafka/helper.godownstreamadapter/sink/kafka/sink.godownstreamadapter/sink/kafka/sink_test.gopkg/sink/codec/canal/canal_json_encoder.gopkg/sink/codec/open/encoder.gopkg/sink/codec/simple/encoder.gopkg/sink/kafka/options.gopkg/sink/kafka/options_test.go
🚧 Files skipped from review as they are similar to previous changes (4)
- downstreamadapter/sink/kafka/sink_test.go
- downstreamadapter/sink/cloudstorage/sink.go
- pkg/sink/kafka/options_test.go
- pkg/sink/kafka/options.go
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
pkg/sink/kafka/options_test.go (1)
411-412: 🎯 Functional Correctness | 🔴 Critical | ⚡ Quick winInitialize
MaxBatchedBytesto prevent test failure.This subtest manually sets
options.MaxMessageBytesbut forgets to setoptions.MaxBatchedBytes. As a result,options.MaxBatchedBytesretains its default value of 10MB. Consequently,adjustTopicOptionssets it tomin(10MB, expectedProducerLimit), which breaks the downstream assertion at line 428 expectingmin(configuredMaxMessageBytes, expectedProducerLimit).To fix the test and accurately simulate the
options.Applystep, initializeMaxBatchedByteshere as well.🐛 Proposed fix for test setup
configuredMaxMessageBytes := test.configuredMaxMessageBytes(adminFixture) options.MaxMessageBytes = configuredMaxMessageBytes + options.MaxBatchedBytes = configuredMaxMessageBytes🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/sink/kafka/options_test.go` around lines 411 - 412, Update the test setup around configuredMaxMessageBytes in the relevant options subtest to initialize options.MaxBatchedBytes to the same configured value as options.MaxMessageBytes, matching the options.Apply behavior and preserving the downstream expected limit assertion.pkg/sink/codec/common/config.go (1)
219-221: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winSynchronize
MaxBatchedByteswhenMaxMessageBytesis parsed from the URI.If
MaxMessageBytesis configured via the URL, it must also be copied toMaxBatchedBytes(just as it is inoptions.go'sApply). Without this synchronization, any downstream usage that callsApplyand setsMaxMessageBytesto a value smaller than the default 10MB will failValidate()becauseMaxBatchedByteswould remain 10MB.🐛 Proposed fix to sync batch limit
if urlParameter.MaxMessageBytes != nil { c.MaxMessageBytes = *urlParameter.MaxMessageBytes + c.MaxBatchedBytes = *urlParameter.MaxMessageBytes }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/sink/codec/common/config.go` around lines 219 - 221, Update the URI parsing block handling urlParameter.MaxMessageBytes in the configuration code to also assign the parsed value to c.MaxBatchedBytes. Keep both limits synchronized, matching the existing behavior in options.go’s Apply method, while preserving the current assignment to c.MaxMessageBytes.
🧹 Nitpick comments (1)
pkg/sink/codec/common/config.go (1)
474-478: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider enforcing
MaxBatchedBytes <= 0.For consistency with the
MaxMessageBytes <= 0validation on line 469 and to prevent a0value which would disable batch limits entirely (or cause infinite splitting depending on codec logic), consider using<= 0here as well.♻️ Proposed refactor
- if c.MaxBatchedBytes < 0 { + if c.MaxBatchedBytes <= 0 { return errors.ErrCodecInvalidConfig.Wrap( errors.Errorf("invalid max-batch-message-bytes %d", c.MaxBatchedBytes), ) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/sink/codec/common/config.go` around lines 474 - 478, Update the MaxBatchedBytes validation in the codec configuration validation flow to reject zero as well as negative values by changing the boundary check to <= 0. Preserve the existing ErrCodecInvalidConfig wrapping and error message behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@pkg/sink/codec/open/encoder.go`:
- Line 176: Update the batching condition in the encoder flow to access
d.config.MaxBatchedBytes as a field rather than invoking it as a method.
Preserve the existing message-length and row-count checks unchanged.
In `@pkg/sink/kafka/options_test.go`:
- Around line 736-740: Update the MaxBatchedBytes references in the test
assertion and the encoder logic to access it as an int field rather than
invoking it as a method. Remove the parentheses from both usages in
options_test.go and the relevant code in the encoder implementation.
---
Outside diff comments:
In `@pkg/sink/codec/common/config.go`:
- Around line 219-221: Update the URI parsing block handling
urlParameter.MaxMessageBytes in the configuration code to also assign the parsed
value to c.MaxBatchedBytes. Keep both limits synchronized, matching the existing
behavior in options.go’s Apply method, while preserving the current assignment
to c.MaxMessageBytes.
In `@pkg/sink/kafka/options_test.go`:
- Around line 411-412: Update the test setup around configuredMaxMessageBytes in
the relevant options subtest to initialize options.MaxBatchedBytes to the same
configured value as options.MaxMessageBytes, matching the options.Apply behavior
and preserving the downstream expected limit assertion.
---
Nitpick comments:
In `@pkg/sink/codec/common/config.go`:
- Around line 474-478: Update the MaxBatchedBytes validation in the codec
configuration validation flow to reject zero as well as negative values by
changing the boundary check to <= 0. Preserve the existing ErrCodecInvalidConfig
wrapping and error message behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 27d9b88f-2c1a-4d21-aaf4-57934a15a725
📒 Files selected for processing (11)
downstreamadapter/sink/helper/helper.godownstreamadapter/sink/kafka/sink.gopkg/sink/codec/canal/canal_json_encoder_test.gopkg/sink/codec/common/config.gopkg/sink/codec/common/config_test.gopkg/sink/codec/open/encoder.gopkg/sink/codec/open/encoder_test.gopkg/sink/codec/simple/encoder_test.gopkg/sink/kafka/options.gopkg/sink/kafka/options_test.gopkg/sink/kafka/sarama_factory.go
🚧 Files skipped from review as they are similar to previous changes (6)
- downstreamadapter/sink/helper/helper.go
- pkg/sink/codec/canal/canal_json_encoder_test.go
- pkg/sink/codec/common/config_test.go
- pkg/sink/codec/simple/encoder_test.go
- pkg/sink/codec/open/encoder_test.go
- downstreamadapter/sink/kafka/sink.go
|
/test all |
|
/test all |
|
/test all |
|
/test all |
|
/test all |
|
/test all |
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: lidezhu, wk989898 The full list of commands accepted by this bot can be found here. The pull request process is described here DetailsNeeds approval from an approver in each of these files:
Approvers can indicate their approval by writing |
ti-chi-bot
Bot
added
approved
and removed
needs-1-more-lgtm
|
/test pull-cdc-kafka-integration-light |
1 similar comment
|
/test pull-cdc-kafka-integration-light |
|
/cherry-pick release-8.5 |
|
@3AceShowHand: new pull request created to branch DetailsIn response to this:
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. |
4 participants
What problem does this PR solve?
Issue Number: close #1405
Kafka sink currently uses the changefeed
max-message-bytessetting for two different purposes: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 allows20 MiB, and one encoded row event is12 MiB, TiCDC still returnsErrMessageTooLargeor invokeslarge-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 changefeedmax-message-bytessetting;K: the Kafka topicmax.message.bytesfor an existing topic, or the brokermessage.max.bytesfor a topic that does not exist yet;K;min(T, K).The implementation:
TasMaxBatchedBytesbefore reading Kafka metadata;MaxMessageBytestoKand uses it for both the producer and the codec's final message-size check;MaxBatchedBytesonly when deciding whether to add another row to the current batch;large-message-handleonly when the original encoded message exceedsMaxMessageBytes(K), and performs a final size check after handling;Tas the producer limit if the Kafka topic or broker size configuration cannot be read.As a result:
Tbut not larger thanKis sent directly;min(T, K);Kinvokes the configuredlarge-message-handle, or returnsErrMessageTooLargewhen it still cannot fit;max-message-bytessetting.This PR does not introduce a separate model for Sarama's request-size limit or cap
Kusingsarama.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
min(T, K).kafka_big_messagescovers Canal JSON, Open Protocol, Simple JSON, Simple Avro, and Avro.ErrMessageTooLarge.max.message.bytes, without updating, pausing, or manually resuming the changefeed, and verifies that synchronization automatically recovers and passes sync-diff.Questions
Will it cause performance regression or break compatibility?
No performance regression or compatibility break is expected.
The batching behavior of
max-message-bytesis preserved. This PR fixes the previous conflation of the batch threshold and Kafka's actual message-size limit: whenT < message size <= K, Kafka can accept the message, so TiCDC sends it instead of incorrectly returningErrMessageTooLargeor invokinglarge-message-handle. IfK < T, the batch threshold is reduced toKso 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-bytescontrols Kafka batching, while Kafka topic or broker configuration controls the final message limit. No monitoring update is required.Release note