kafka: use default option when meeting authorization and ACLs fails#5562
Conversation
|
Skipping CI for Draft Pull Request. |
ti-chi-bot
Bot
added
release-note
📝 WalkthroughWalkthroughError handling in pkg/sink/kafka/options.go is relaxed for Kafka configuration lookups. Failures reading topic or broker max message bytes now log warnings and fall back to configured/default values instead of returning errors. Similarly, min.insync.replicas fetch errors are now treated as non-fatal, logging a warning and assuming valid configuration. ChangesKafka Configuration Fallback Handling
Estimated code review effort: 2 (Simple) | ~10 minutes Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 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/S
![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 introduces fallback mechanisms in pkg/sink/kafka/options.go when retrieving Kafka configurations (such as max.message.bytes and message.max.bytes) fails, falling back to options.MaxMessageBytes instead of returning an error. It also updates validateMinInsyncReplicas to log a warning and assume validity if retrieving min.insync.replicas fails. However, the reviewer identified critical bugs in the fallback implementations: if retrieving the configuration fails, the code still attempts to parse the uninitialized configuration string using strconv.Atoi, which will fail and return an error anyway. The reviewer provided code suggestions to wrap the parsing logic in else blocks to ensure the fallback values are correctly used.
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.
| var topicMaxMessageBytes int | ||
| if err != nil { | ||
| return errors.Trace(err) | ||
| log.Warn("TiCDC cannot find `max.message.bytes` from topic's configuration, use the option `MaxMessageBytes` as default") | ||
| topicMaxMessageBytes = options.MaxMessageBytes | ||
| // return errors.Trace(err) | ||
| } | ||
| topicMaxMessageBytes, err := strconv.Atoi(topicMaxMessageBytesStr) | ||
| topicMaxMessageBytes, err = strconv.Atoi(topicMaxMessageBytesStr) | ||
| if err != nil { | ||
| return errors.Trace(err) | ||
| } |
There was a problem hiding this comment.
If getTopicConfig fails, err is not nil, and we set topicMaxMessageBytes = options.MaxMessageBytes. However, the code immediately proceeds to execute strconv.Atoi(topicMaxMessageBytesStr). Since topicMaxMessageBytesStr is empty or invalid, strconv.Atoi will fail and return an error, which is then returned by the function. This completely defeats the fallback mechanism. We should only parse topicMaxMessageBytesStr when getTopicConfig succeeds.
| var topicMaxMessageBytes int | |
| if err != nil { | |
| return errors.Trace(err) | |
| log.Warn("TiCDC cannot find `max.message.bytes` from topic's configuration, use the option `MaxMessageBytes` as default") | |
| topicMaxMessageBytes = options.MaxMessageBytes | |
| // return errors.Trace(err) | |
| } | |
| topicMaxMessageBytes, err := strconv.Atoi(topicMaxMessageBytesStr) | |
| topicMaxMessageBytes, err = strconv.Atoi(topicMaxMessageBytesStr) | |
| if err != nil { | |
| return errors.Trace(err) | |
| } | |
| var topicMaxMessageBytes int | |
| if err != nil { | |
| log.Warn("TiCDC cannot find `max.message.bytes` from topic's configuration, use the option `MaxMessageBytes` as default") | |
| topicMaxMessageBytes = options.MaxMessageBytes | |
| } else { | |
| topicMaxMessageBytes, err = strconv.Atoi(topicMaxMessageBytesStr) | |
| if err != nil { | |
| return errors.Trace(err) | |
| } | |
| } |
| var brokerMessageMaxBytes int | ||
| brokerMessageMaxBytesStr, err := admin.GetBrokerConfig(BrokerMessageMaxBytesConfigName) | ||
| if err != nil { | ||
| log.Warn("TiCDC cannot find `message.max.bytes` from broker's configuration") | ||
| return errors.Trace(err) | ||
| log.Warn("TiCDC cannot find `message.max.bytes` from broker's configuration, use the option `MaxMessageBytes` as default") | ||
| brokerMessageMaxBytes = options.MaxMessageBytes | ||
| } | ||
| brokerMessageMaxBytes, err := strconv.Atoi(brokerMessageMaxBytesStr) | ||
| brokerMessageMaxBytes, err = strconv.Atoi(brokerMessageMaxBytesStr) | ||
| if err != nil { | ||
| return errors.Trace(err) | ||
| } |
There was a problem hiding this comment.
If admin.GetBrokerConfig fails, err is not nil, and we set brokerMessageMaxBytes = options.MaxMessageBytes. However, the code immediately proceeds to execute strconv.Atoi(brokerMessageMaxBytesStr). Since brokerMessageMaxBytesStr is empty or invalid, strconv.Atoi will fail and return an error, which is then returned by the function. This completely defeats the fallback mechanism. We should only parse brokerMessageMaxBytesStr when admin.GetBrokerConfig succeeds.
| var brokerMessageMaxBytes int | |
| brokerMessageMaxBytesStr, err := admin.GetBrokerConfig(BrokerMessageMaxBytesConfigName) | |
| if err != nil { | |
| log.Warn("TiCDC cannot find `message.max.bytes` from broker's configuration") | |
| return errors.Trace(err) | |
| log.Warn("TiCDC cannot find `message.max.bytes` from broker's configuration, use the option `MaxMessageBytes` as default") | |
| brokerMessageMaxBytes = options.MaxMessageBytes | |
| } | |
| brokerMessageMaxBytes, err := strconv.Atoi(brokerMessageMaxBytesStr) | |
| brokerMessageMaxBytes, err = strconv.Atoi(brokerMessageMaxBytesStr) | |
| if err != nil { | |
| return errors.Trace(err) | |
| } | |
| var brokerMessageMaxBytes int | |
| brokerMessageMaxBytesStr, err := admin.GetBrokerConfig(BrokerMessageMaxBytesConfigName) | |
| if err != nil { | |
| log.Warn("TiCDC cannot find `message.max.bytes` from broker's configuration, use the option `MaxMessageBytes` as default") | |
| brokerMessageMaxBytes = options.MaxMessageBytes | |
| } else { | |
| brokerMessageMaxBytes, err = strconv.Atoi(brokerMessageMaxBytesStr) | |
| if err != nil { | |
| return errors.Trace(err) | |
| } | |
| } |
ti-chi-bot
Bot
added
size/M
ti-chi-bot
Bot
removed
the
do-not-merge/work-in-progress
ti-chi-bot
Bot
added
needs-1-more-lgtm
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: asddongmen, lidezhu 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
lgtm
and removed
needs-1-more-lgtm
[LGTM Timeline notifier]Timeline:
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
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 (2)
pkg/sink/kafka/options.go (2)
610-633: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winFallback value self-triggers spurious overhead subtraction.
When
getTopicConfigfails entirely,topicMaxMessageBytesis set equal tooptions.MaxMessageBytes(Line 613). That equal value then flows into the unchanged comparison at Line 622 (topicMaxMessageBytes <= options.MaxMessageBytes), which is trivially true on equality, causing Line 628 to executeoptions.MaxMessageBytes = topicMaxMessageBytes - maxMessageBytesOverhead. The net effect: every time this fallback triggers, the user's configuredMaxMessageBytesis silently shrunk by 128 bytes for no real reason, and the emitted warning at Line 623-627 misleadingly claims the topic's actualmax.message.bytesis smaller than the configured option, even though no real topic data was obtained.This undermines the fallback's intent of "keep using the option value as-is" — it should leave
options.MaxMessageBytesuntouched when the fallback path is taken.🐛 Proposed fix to skip the overhead adjustment on fallback
var topicMaxMessageBytes int + usedFallback := false if err != nil { log.Warn("TiCDC cannot find `max.message.bytes` from topic's configuration, use the option `MaxMessageBytes` as default") topicMaxMessageBytes = options.MaxMessageBytes + usedFallback = true } else { topicMaxMessageBytes, err = strconv.Atoi(topicMaxMessageBytesStr) if err != nil { return errors.Trace(err) } } - maxMessageBytes := topicMaxMessageBytes - maxMessageBytesOverhead - if topicMaxMessageBytes <= options.MaxMessageBytes { - log.Warn("topic's `max.message.bytes` less than the `max-message-bytes`,"+ - "use topic's `max.message.bytes` to initialize the Kafka producer", - zap.Int("max.message.bytes", topicMaxMessageBytes), - zap.Int("max-message-bytes", options.MaxMessageBytes), - zap.Int("real-max-message-bytes", maxMessageBytes)) - options.MaxMessageBytes = maxMessageBytes - } else { - if maxMessageBytes < options.MaxMessageBytes { - options.MaxMessageBytes = maxMessageBytes + if !usedFallback { + maxMessageBytes := topicMaxMessageBytes - maxMessageBytesOverhead + if topicMaxMessageBytes <= options.MaxMessageBytes { + log.Warn("topic's `max.message.bytes` less than the `max-message-bytes`,"+ + "use topic's `max.message.bytes` to initialize the Kafka producer", + zap.Int("max.message.bytes", topicMaxMessageBytes), + zap.Int("max-message-bytes", options.MaxMessageBytes), + zap.Int("real-max-message-bytes", maxMessageBytes)) + options.MaxMessageBytes = maxMessageBytes + } else { + if maxMessageBytes < options.MaxMessageBytes { + options.MaxMessageBytes = 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/kafka/options.go` around lines 610 - 633, The fallback path in the Kafka topic config handling is incorrectly treated like a real topic value, so `options.MaxMessageBytes` gets reduced by `maxMessageBytesOverhead` even when `getTopicConfig` fails. Update the logic in the `options.go` flow around `topicMaxMessageBytes` so the defaulting branch that uses `options.MaxMessageBytes` bypasses the `topicMaxMessageBytes <= options.MaxMessageBytes` comparison and the subsequent overwrite, leaving the configured `MaxMessageBytes` unchanged when no topic config is available.
649-677: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winSame self-triggering overhead-reduction bug on the broker fallback path.
Identical pattern to the topic-config fallback: on lookup failure
brokerMessageMaxBytesis set tooptions.MaxMessageBytes(Line 653), which then makes the unchanged Line 666 (brokerMessageMaxBytes <= options.MaxMessageBytes) trivially true, shrinkingoptions.MaxMessageBytesbymaxMessageBytesOverheadand logging a misleading "broker's message.max.bytes less than max-message-bytes" warning based on fabricated equal values rather than real broker data.🐛 Proposed fix mirroring the topic-side fix
var brokerMessageMaxBytes int + usedFallback := false brokerMessageMaxBytesStr, err := admin.GetBrokerConfig(BrokerMessageMaxBytesConfigName) if err != nil { log.Warn("TiCDC cannot find `message.max.bytes` from broker's configuration, use the option `MaxMessageBytes` as default") brokerMessageMaxBytes = options.MaxMessageBytes + usedFallback = true } else { brokerMessageMaxBytes, err = strconv.Atoi(brokerMessageMaxBytesStr) if err != nil { return errors.Trace(err) } } - maxMessageBytes := brokerMessageMaxBytes - maxMessageBytesOverhead - if brokerMessageMaxBytes <= options.MaxMessageBytes { - ... - options.MaxMessageBytes = maxMessageBytes - } else { - if maxMessageBytes < options.MaxMessageBytes { - options.MaxMessageBytes = maxMessageBytes + if !usedFallback { + maxMessageBytes := brokerMessageMaxBytes - maxMessageBytesOverhead + if brokerMessageMaxBytes <= options.MaxMessageBytes { + ... + options.MaxMessageBytes = maxMessageBytes + } else { + if maxMessageBytes < options.MaxMessageBytes { + options.MaxMessageBytes = 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/kafka/options.go` around lines 649 - 677, In the broker fallback path in options.go, the current use of options.MaxMessageBytes as the default for brokerMessageMaxBytes causes the later comparison in the MaxMessageBytes adjustment block to self-trigger and incorrectly shrink the producer limit. Update the GetBrokerConfig fallback in the brokerMessageMaxBytes handling so it does not feed a fabricated value into the comparison; instead preserve a separate fallback state or skip the broker-based override when the broker config lookup fails, matching the topic-config fix. Keep the logic in the MaxMessageBytes adjustment section and the warning around broker's message.max.bytes consistent with the real broker value, using the existing symbols BrokerMessageMaxBytesConfigName, brokerMessageMaxBytes, and options.MaxMessageBytes.
🤖 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 `@pkg/sink/kafka/options.go`:
- Around line 610-633: The fallback path in the Kafka topic config handling is
incorrectly treated like a real topic value, so `options.MaxMessageBytes` gets
reduced by `maxMessageBytesOverhead` even when `getTopicConfig` fails. Update
the logic in the `options.go` flow around `topicMaxMessageBytes` so the
defaulting branch that uses `options.MaxMessageBytes` bypasses the
`topicMaxMessageBytes <= options.MaxMessageBytes` comparison and the subsequent
overwrite, leaving the configured `MaxMessageBytes` unchanged when no topic
config is available.
- Around line 649-677: In the broker fallback path in options.go, the current
use of options.MaxMessageBytes as the default for brokerMessageMaxBytes causes
the later comparison in the MaxMessageBytes adjustment block to self-trigger and
incorrectly shrink the producer limit. Update the GetBrokerConfig fallback in
the brokerMessageMaxBytes handling so it does not feed a fabricated value into
the comparison; instead preserve a separate fallback state or skip the
broker-based override when the broker config lookup fails, matching the
topic-config fix. Keep the logic in the MaxMessageBytes adjustment section and
the warning around broker's message.max.bytes consistent with the real broker
value, using the existing symbols BrokerMessageMaxBytesConfigName,
brokerMessageMaxBytes, and options.MaxMessageBytes.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 3689eaaf-d98e-4ef8-9878-d8b3a18ef730
📒 Files selected for processing (1)
pkg/sink/kafka/options.go
|
/cherry-pick release-nextgen-202603 |
|
@wk989898: 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 #5563
What is changed and how it works?
If ticdc fails to authorize acl on the kafka server, try using the default option instead of throwing an error when creating the changefeed. The change will fail later.
This is to prevent the changefeed from being created if the user does not grant the relevant acl permissions.
Check List
Tests
DescribeConfigsgranted?changefeed createmax.message.bytes=1024warning, broker rejects large DML with message-too-largemax.message.bytes=1024warning, TiCDC detectsErrMessageTooLargeusing adjustedmaxMessageBytes=896max.message.bytes=1024warning, broker rejects large DML with message-too-largemax.message.bytes=1024warning, TiCDC detectsErrMessageTooLargeusing adjustedmaxMessageBytes=896min.insync.replicas=2, RF=1warning, Kafka send fails after DDLmin.insync.replicas=2, RF=1min.insync.replicas=2, RF=1normalin this local broker testmin.insync.replicas=2, RF=1Questions
Will it cause performance regression or break compatibility?
Do you need to update user documentation, design documentation or monitoring documentation?
Release note
Summary by CodeRabbit