kafka: use default option when meeting authorization and ACLs fails (#5562)#5591
Conversation
ti-chi-bot
added
lgtm
release-note
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: The full list of commands accepted by this bot can be found here. DetailsNeeds approval from an approver in each of these files:Approvers can indicate their approval by writing |
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
✨ 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 |
![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 improves the resilience of Kafka sink option adjustments by falling back to default configurations (such as MaxMessageBytes) or assuming validity when failing to retrieve topic or broker configurations (like max.message.bytes, message.max.bytes, and min.insync.replicas), rather than failing immediately. The review feedback suggests enhancing the warning logs in these fallback paths by including the actual underlying error (zap.Error(err)) to assist operators with troubleshooting.
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.
| if err != nil { | ||
| return errors.Trace(err) | ||
| } | ||
| topicMaxMessageBytes, err := strconv.Atoi(topicMaxMessageBytesStr) | ||
| 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 |
There was a problem hiding this comment.
When falling back to the default MaxMessageBytes due to an error fetching the topic configuration, it is important to log the actual error err. This helps operators diagnose whether the failure was due to an authorization/ACL issue, a network timeout, or another problem.
| if err != nil { | |
| return errors.Trace(err) | |
| } | |
| topicMaxMessageBytes, err := strconv.Atoi(topicMaxMessageBytesStr) | |
| 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 | |
| if err != nil { | |
| log.Warn("TiCDC cannot find 'max.message.bytes' from topic's configuration, use the option 'MaxMessageBytes' as default", zap.Error(err)) | |
| topicMaxMessageBytes = options.MaxMessageBytes |
| if err != nil { | ||
| log.Warn("TiCDC cannot find `message.max.bytes` from broker's configuration") | ||
| return errors.Trace(err) | ||
| } | ||
| brokerMessageMaxBytes, err := strconv.Atoi(brokerMessageMaxBytesStr) | ||
| if err != nil { | ||
| 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 |
There was a problem hiding this comment.
When falling back to the default MaxMessageBytes due to an error fetching the broker configuration, the actual error err should be logged to assist with troubleshooting.
| if err != nil { | |
| log.Warn("TiCDC cannot find `message.max.bytes` from broker's configuration") | |
| return errors.Trace(err) | |
| } | |
| brokerMessageMaxBytes, err := strconv.Atoi(brokerMessageMaxBytesStr) | |
| if err != nil { | |
| 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 | |
| if err != nil { | |
| log.Warn("TiCDC cannot find 'message.max.bytes' from broker's configuration, use the option 'MaxMessageBytes' as default", zap.Error(err)) | |
| brokerMessageMaxBytes = options.MaxMessageBytes |
| log.Warn("TiCDC meets error when get `min.insync.replicas` from broker's configuration, assume the config is valid") | ||
| return nil |
There was a problem hiding this comment.
When ignoring the error while fetching min.insync.replicas, the actual error err should be logged so that operators can understand why the configuration could not be retrieved.
| log.Warn("TiCDC meets error when get `min.insync.replicas` from broker's configuration, assume the config is valid") | |
| return nil | |
| log.Warn("TiCDC meets error when get 'min.insync.replicas' from broker's configuration, assume the config is valid", zap.Error(err)) | |
| return nil |
|
/test next-gen |
|
@ti-chi-bot: The following test failed, say
Full PR test history. Your PR dashboard. DetailsInstructions 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 kubernetes-sigs/prow repository. I understand the commands that are listed here. |
2 participants
This is an automated cherry-pick of #5562
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