consumer: revert https://github.com/pingcap/ticdc/pull/4052#5438
consumer: revert https://github.com/pingcap/ticdc/pull/4052#5438wk989898 wants to merge 3 commits into
Conversation
ti-chi-bot
Bot
added
do-not-merge/needs-linked-issue
release-note
|
/test all |
📝 WalkthroughWalkthroughRemoves the ChangesRemove AppliedWatermark and rework DML fallback routing
Estimated code review effort🎯 4 (Complex) | ⏱️ ~75 minutes Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ 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
![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 simplifies the DML event handling and watermark tracking in both the Kafka and Pulsar consumers. It removes the AppliedWatermark tracking and resolvedGroups logic from the writers and EventsGroup, and refactors appendRow2Group to handle fallback events based on the protocol type. Additionally, corresponding tests for the removed fallback merging logic have been cleaned up. The review feedback highlights two improvement opportunities in cmd/pulsar-consumer/writer.go: changing a high-frequency log from Info to Debug level to prevent log flooding, and removing a redundant logging entry for progress.watermark.
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 commitTs >= group.HighWatermark { | ||
| group.Append(dml, false) | ||
| log.Info("DML event append to the group", |
There was a problem hiding this comment.
Logging every DML event append at Info level can cause severe log flooding and performance issues under high throughput. It should be logged at Debug level, consistent with the Kafka consumer implementation.
| log.Info("DML event append to the group", | |
| log.Debug("DML event append to the group", |
| } | ||
| log.Warn("DML event fallback row, since less than the group high watermark, ignore it", | ||
| zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), | ||
| zap.Any("partitionWatermark", progress.watermark), zap.Any("watermark", progress.watermark), |
There was a problem hiding this comment.
The progress.watermark is logged twice redundantly under different keys (partitionWatermark and watermark). This appears to be a copy-paste error from the Kafka consumer where watermarkOffset was logged. We should simplify this to log it only once, and use zap.Uint64 for better type safety.
| zap.Any("partitionWatermark", progress.watermark), zap.Any("watermark", progress.watermark), | |
| zap.Uint64("partitionWatermark", progress.watermark), |
![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
🧹 Nitpick comments (2)
cmd/kafka-consumer/writer.go (1)
560-624: 💤 Low valueStale comment mentions "open protocol" in Simple/Debezium case block.
The comment on lines 589-592 mentions "open protocol set the partition table id..." but this case block only handles
ProtocolSimpleandProtocolDebezium. The open protocol handling is in the next case block (lines 601-620). This is misleading and should be updated.📝 Suggested comment fix
case config.ProtocolSimple, config.ProtocolDebezium: - // simple protocol set the table id for all row message, it can be known which table the row message belongs to, - // also consider the table partition. - // open protocol set the partition table id if the table is partitioned. - // for normal table, the table id is generated by the fake table id generator by using schema and table name. - // so one event group for one normal table or one table partition, replayed messages can be ignored. + // Simple and Debezium protocols include the table ID for all row messages, making it possible + // to identify which table the row belongs to. Since each event group corresponds to one table + // (or one table partition), replayed messages with commitTs below HighWatermark can be safely ignored. log.Warn("DML event fallback row, since less than the group high watermark, ignore it",🤖 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 `@cmd/kafka-consumer/writer.go` around lines 560 - 624, The comment block in the case config.ProtocolSimple, config.ProtocolDebezium branch incorrectly mentions open protocol behavior, which is actually handled in the separate case config.ProtocolCanalJSON, config.ProtocolOpen, config.ProtocolAvro branch. Update the comment to only describe the Simple and Debezium protocol behavior without referencing open protocol, and remove the part about fake table id generation and how open protocol assigns partition table ids since those details belong in the other case block.cmd/pulsar-consumer/writer.go (1)
490-510: Pulsar sink intentionally supports only CanalJSON protocol, which is enforced by validation before the writer is initialized.The protocol validation in
options.go(lines 88-91) explicitly panics if any protocol other thanProtocolCanalJSONis used, preventing unsupported protocols from reaching the writer. The switch statement at lines 490-510 is unreachable defensive code.However, line 115 contains unnecessary code: the comparison
o.protocol == config.ProtocolAvroalways evaluates to false since Avro is not a supported protocol for Pulsar. This dead code should be simplified to passfalsedirectly toNewEventRouter.🤖 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 `@cmd/pulsar-consumer/writer.go` around lines 490 - 510, Since the Pulsar sink only supports ProtocolCanalJSON and this is enforced by validation in options.go before the writer is initialized, the switch statement in the writer is unreachable defensive code that should be simplified. Additionally, remove the dead code comparison at line 115 where o.protocol is compared to config.ProtocolAvro (which always evaluates to false since Avro is not a supported protocol for Pulsar) and instead pass false directly to the NewEventRouter call. The switch statement and protocol check create unnecessary complexity that contradicts the enforced protocol restrictions.
🤖 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 `@cmd/pulsar-consumer/writer.go`:
- Around line 474-481: The log level for the DML event append operation is
inconsistent with the equivalent Kafka writer implementation. Change the
log.Info call for the "DML event append to the group" message to log.Debug to
match the log level used in the Kafka writer and prevent excessive log volume in
production during high-throughput scenarios.
---
Nitpick comments:
In `@cmd/kafka-consumer/writer.go`:
- Around line 560-624: The comment block in the case config.ProtocolSimple,
config.ProtocolDebezium branch incorrectly mentions open protocol behavior,
which is actually handled in the separate case config.ProtocolCanalJSON,
config.ProtocolOpen, config.ProtocolAvro branch. Update the comment to only
describe the Simple and Debezium protocol behavior without referencing open
protocol, and remove the part about fake table id generation and how open
protocol assigns partition table ids since those details belong in the other
case block.
In `@cmd/pulsar-consumer/writer.go`:
- Around line 490-510: Since the Pulsar sink only supports ProtocolCanalJSON and
this is enforced by validation in options.go before the writer is initialized,
the switch statement in the writer is unreachable defensive code that should be
simplified. Additionally, remove the dead code comparison at line 115 where
o.protocol is compared to config.ProtocolAvro (which always evaluates to false
since Avro is not a supported protocol for Pulsar) and instead pass false
directly to the NewEventRouter call. The switch statement and protocol check
create unnecessary complexity that contradicts the enforced protocol
restrictions.
🪄 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: e3c9de01-fe8b-49be-ae5f-a28117c48606
📒 Files selected for processing (6)
cmd/kafka-consumer/writer.gocmd/kafka-consumer/writer_test.gocmd/pulsar-consumer/writer.gocmd/pulsar-consumer/writer_test.gocmd/util/event_group.gocmd/util/event_group_test.go
💤 Files with no reviewable changes (2)
- cmd/util/event_group_test.go
- cmd/pulsar-consumer/writer_test.go
| if commitTs >= group.HighWatermark { | ||
| group.Append(dml, false) | ||
| log.Info("DML event append to the group", | ||
| zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), | ||
| zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), | ||
| zap.Stringer("eventType", dml.RowTypes[0])) | ||
| return | ||
| } |
There was a problem hiding this comment.
Log level inconsistency with Kafka writer.
The normal append path uses log.Info here (line 476), while the equivalent code in the Kafka writer uses log.Debug (line 570). This could cause excessive log volume in production for high-throughput scenarios.
📝 Suggested fix
if commitTs >= group.HighWatermark {
group.Append(dml, false)
- log.Info("DML event append to the group",
+ log.Debug("DML event append to the group",
zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark),
zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
zap.Stringer("eventType", dml.RowTypes[0]))
return
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if commitTs >= group.HighWatermark { | |
| group.Append(dml, false) | |
| log.Info("DML event append to the group", | |
| zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), | |
| zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), | |
| zap.Stringer("eventType", dml.RowTypes[0])) | |
| return | |
| } | |
| if commitTs >= group.HighWatermark { | |
| group.Append(dml, false) | |
| log.Debug("DML event append to the group", | |
| zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), | |
| zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), | |
| zap.Stringer("eventType", dml.RowTypes[0])) | |
| return | |
| } |
🤖 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 `@cmd/pulsar-consumer/writer.go` around lines 474 - 481, The log level for the
DML event append operation is inconsistent with the equivalent Kafka writer
implementation. Change the log.Info call for the "DML event append to the group"
message to log.Debug to match the log level used in the Kafka writer and prevent
excessive log volume in production during high-throughput scenarios.
|
/test pull-cdc-kafka-integration-light |
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:
|
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: 3AceShowHand, lidezhu, wlwilliamx 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 |
1 similar comment
|
/retest |
|
@wk989898: 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. |
4 participants
What problem does this PR solve?
Issue Number: ref #4051
What is changed and how it works?
TiCDC must send commit-ts events with linear growth.
The only possibility for receiving small commi-ts with large offsets is that the event is repeated.
Check List
Tests
Questions
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
Bug Fixes
CREATE TABLE ... LIKE ...) to prevent incorrect routing and missing data.Refactor
Tests