kafka: unify sink errors and replace failpoint tests - #5786
Conversation
ti-chi-bot
Bot
added
do-not-merge/needs-linked-issue
release-note
|
Warning Review limit reached
Next review available in: 49 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (2)
📝 WalkthroughWalkthroughKafka sink errors are consolidated around configuration, admin, producer, and sink-level classifications. Admin configuration lookups now report presence explicitly, while producer, sink, topic-manager, claim-check, and integration tests follow the revised contracts. ChangesKafka error taxonomy and producer runtime
Admin configuration contract and option resolution
Sink and topic-manager propagation
Integration scenario updates
Estimated code review effort: 3 (Moderate) | ~30 minutes 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 |
![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/kafka/sarama_factory.go`:
- Around line 125-127: Close the previously created client before returning from
both producer-construction failure paths: the sarama.NewSyncProducerFromClient
call at pkg/sink/kafka/sarama_factory.go:125-127 and the
sarama.NewAsyncProducerFromClient call at
pkg/sink/kafka/sarama_factory.go:152-154. Preserve the existing wrapped error
returns after performing the cleanup.
🪄 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 Plus
Run ID: 0de4272f-86ca-4b46-9a28-e301df7632e1
📒 Files selected for processing (21)
downstreamadapter/sink/eventrouter/topic/expression.godownstreamadapter/sink/eventrouter/topic/expression_test.godownstreamadapter/sink/kafka/helper.godownstreamadapter/sink/kafka/sink.godownstreamadapter/sink/pulsar/helper.godownstreamadapter/sink/topicmanager/kafka_topic_manager.godownstreamadapter/sink/topicmanager/kafka_topic_manager_test.gopkg/errors/error.gopkg/errors/error_test.gopkg/sink/kafka/admin.gopkg/sink/kafka/admin_test.gopkg/sink/kafka/cluster_admin_client.gopkg/sink/kafka/cluster_admin_client_mock.gopkg/sink/kafka/options.gopkg/sink/kafka/options_test.gopkg/sink/kafka/sarama_async_producer.gopkg/sink/kafka/sarama_config.gopkg/sink/kafka/sarama_factory.gopkg/sink/kafka/sarama_sync_producer.gopkg/sink/kafka/sarama_sync_producer_test.gotests/integration_tests/http_api/util/test_case.py
3AceShowHand
changed the title
ti-chi-bot
Bot
added
release-note-none
3AceShowHand
changed the title
ti-chi-bot
Bot
added
size/XXL
|
/test all |
|
/test all |
ti-chi-bot
Bot
added
needs-1-more-lgtm
ti-chi-bot
Bot
added
lgtm
and removed
needs-1-more-lgtm
[LGTM Timeline notifier]Timeline:
|
|
/test all |
There was a problem hiding this comment.
🧹 Nitpick comments (2)
tests/integration_tests/mq_sink_error_resume/run.sh (1)
26-33: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueQuote variable expansions flagged by shellcheck.
$WORK_DIR,$CDC_BINARY,$pd_addr, and$SINK_URIare unquoted on these lines, risking globbing/word-splitting.🔧 Suggested fix
- run_pulsar_cluster $WORK_DIR normal + run_pulsar_cluster "$WORK_DIR" normal SINK_URI="pulsar://127.0.0.1:6650/$TOPIC_NAME?protocol=canal-json&enable-tidb-extension=true" # Return one failpoint error to fail the changefeed. # Note we return one error for the failpoint, if owner retry changefeed frequently, it may break the test. export GO_FAILPOINTS='github.com/pingcap/ticdc/downstreamadapter/sink/pulsar/PulsarSinkAsyncSendError=1*return(true)' - run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY --addr "127.0.0.1:8300" --pd $pd_addr + run_cdc_server --workdir "$WORK_DIR" --binary "$CDC_BINARY" --addr "127.0.0.1:8300" --pd "$pd_addr" - changefeed_id=$(cdc_cli_changefeed create --pd=$pd_addr --sink-uri="$SINK_URI" | grep '^ID:' | head -n1 | awk '{print $2}') + changefeed_id=$(cdc_cli_changefeed create --pd="$pd_addr" --sink-uri="$SINK_URI" | grep '^ID:' | head -n1 | awk '{print $2}') - run_pulsar_consumer --upstream-uri $SINK_URI + run_pulsar_consumer --upstream-uri "$SINK_URI"🤖 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 `@tests/integration_tests/mq_sink_error_resume/run.sh` around lines 26 - 33, Quote the variable expansions in the integration test commands using WORK_DIR, CDC_BINARY, pd_addr, and SINK_URI, including the run_cdc_server, cdc_cli_changefeed, and run_pulsar_consumer invocations, to prevent shell globbing and word splitting while preserving the existing command behavior.Source: Linters/SAST tools
pkg/sink/kafka/sarama_sync_producer.go (1)
58-67: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicate log+wrap logic; verify logging guideline compliance.
SendMessageandSendMessagesrepeat the samelog.Error(...)block (keyspace/changefeed/eventContext fields) before wrapping witherrors.ErrKafkaSendMessage. Consider extracting a small helper (e.g.p.logAndWrapSendError(err, message)) to avoid field drift between the two call sites.Separately, this introduces a new structured log call and removes the prior failpoint-based error injection — worth double-checking against the repo's logging conventions (log level, field set, avoiding duplicate logging if the caller in
downstreamadapter/sink/kafka/sink.goalso logs on the same failure path).♻️ Suggested helper extraction
+func (p *saramaSyncProducer) logAndWrapSendError(err error, message *codecCommon.Message) error { + log.Error("send message to kafka failed", + zap.String("keyspace", p.id.Keyspace()), + zap.String("changefeed", p.id.Name()), + zap.String("eventContext", BuildEventLogContext(p.id.Keyspace(), p.id.Name(), message.LogInfo)), + zap.Error(err)) + return errors.WrapError(errors.ErrKafkaSendMessage, err) +} + func (p *saramaSyncProducer) SendMessage(topic string, partitionNum int32, message *codecCommon.Message) error { ... _, _, err := p.producer.SendMessage(msg) if err == nil { return nil } - log.Error("send message to kafka failed", - zap.String("keyspace", p.id.Keyspace()), - zap.String("changefeed", p.id.Name()), - zap.String("eventContext", BuildEventLogContext(p.id.Keyspace(), p.id.Name(), message.LogInfo)), - zap.Error(err)) - return errors.WrapError(errors.ErrKafkaSendMessage, err) + return p.logAndWrapSendError(err, message) }As per coding guidelines: "Logs are operational signals; see docs/agents/logging.md before adding, removing, or rewriting logs."
Also applies to: 84-92
🤖 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/sarama_sync_producer.go` around lines 58 - 67, The SendMessage and SendMessages paths duplicate Kafka send-error logging and wrapping, with potential logging-convention or duplicate-log issues. Extract shared handling into a helper such as logAndWrapSendError using the existing keyspace, changefeed, eventContext, and ErrKafkaSendMessage behavior, then verify it complies with the repository logging guidance and does not duplicate caller logging.Source: Coding guidelines
🤖 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.
Nitpick comments:
In `@pkg/sink/kafka/sarama_sync_producer.go`:
- Around line 58-67: The SendMessage and SendMessages paths duplicate Kafka
send-error logging and wrapping, with potential logging-convention or
duplicate-log issues. Extract shared handling into a helper such as
logAndWrapSendError using the existing keyspace, changefeed, eventContext, and
ErrKafkaSendMessage behavior, then verify it complies with the repository
logging guidance and does not duplicate caller logging.
In `@tests/integration_tests/mq_sink_error_resume/run.sh`:
- Around line 26-33: Quote the variable expansions in the integration test
commands using WORK_DIR, CDC_BINARY, pd_addr, and SINK_URI, including the
run_cdc_server, cdc_cli_changefeed, and run_pulsar_consumer invocations, to
prevent shell globbing and word splitting while preserving the existing command
behavior.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: ce5d84ab-2bf2-416c-9863-fe6442613c8d
📒 Files selected for processing (23)
downstreamadapter/sink/kafka/helper.godownstreamadapter/sink/kafka/sink.godownstreamadapter/sink/kafka/sink_test.godownstreamadapter/sink/pulsar/helper.gopkg/sink/kafka/admin_test.gopkg/sink/kafka/claimcheck/claim_check.gopkg/sink/kafka/claimcheck/claim_check_test.gopkg/sink/kafka/logutil.gopkg/sink/kafka/logutil_test.gopkg/sink/kafka/oauth2_token_provider.gopkg/sink/kafka/oauth2_token_provider_test.gopkg/sink/kafka/options_test.gopkg/sink/kafka/sarama_async_producer.gopkg/sink/kafka/sarama_factory.gopkg/sink/kafka/sarama_sync_producer.gopkg/sink/kafka/sarama_sync_producer_test.gopkg/util/external_storage.gotests/integration_tests/http_api/util/test_case.pytests/integration_tests/kafka_big_messages/run.shtests/integration_tests/kafka_log_info/run.shtests/integration_tests/mq_sink_error_resume/run.shtests/integration_tests/run_heavy_it_in_ci.shtests/integration_tests/run_light_it_in_ci.sh
💤 Files with no reviewable changes (1)
- tests/integration_tests/kafka_log_info/run.sh
🚧 Files skipped from review as they are similar to previous changes (3)
- tests/integration_tests/http_api/util/test_case.py
- pkg/sink/kafka/sarama_factory.go
- pkg/sink/kafka/options_test.go
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: asddongmen, hongyunyan, 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 |
|
/retest |
|
/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. |
5 participants
What problem does this PR solve?
Kafka sink errors are overly fine-grained and inconsistently wrapped. Some propagation paths add multiple TiCDC error codes, lose the original cause, or use an error category owned by another sink. This makes retry classification and troubleshooting harder.
Kafka producer failure tests also rely on failpoints even when the behavior can be covered by focused unit tests and real Kafka failures.
Issue Number: close #5790
What is changed and how it works?
ErrKafkaInvalidConfig.ErrKafkaAdminAPI.ErrKafkaSendMessage.ErrNewKafkaSinkandErrKafkaSinkClosed, and delete superseded Kafka error definitions.The Kafka failpoints are removed without dropping their behavioral coverage:
KafkaSinkAsyncSendError: replaced by a real broker-rejection scenario inkafka_big_messages. The producer starts with a large topic limit, the test lowers the broker-side topic limit, and Kafka rejects an already-enqueued message through SaramaErrors(). The test checksErrKafkaSendMessage, warning-to-normal recovery, replay, and final data consistency.AsyncRunCallbackerror makessink.Runreturn the error and mark the sink abnormal without depending on Sarama.KafkaSinkSyncSendMessageError: single-message failure, cause preservation, and singleErrKafkaSendMessagewrapping are covered byTestSyncProducerErrorWrappedOnce.KafkaSinkSyncSendMessagesError: batch failure, cause preservation, and singleErrKafkaSendMessagewrapping are covered byTestSyncProducerErrorWrappedOnce.BuildEventLogContextunit tests.Check List
Tests
The Kafka integration scenario requires the repository Linux test environment and is left for CI; local unit tests, shell validation, and static checks pass on macOS/ARM.
Questions
Will it cause performance regression or break compatibility?
No performance regression is expected. This PR intentionally removes obsolete exported Kafka error variables and changes affected RFC codes and messages; compatibility aliases are not retained.
Do you need to update user documentation, design documentation or monitoring documentation?
No user-facing documentation or monitoring changes are required.
Release note
Summary by CodeRabbit
Bug Fixes
Tests