Kafka Connect: Fix coordinator potentially committing files from prior commit in certain rebalance scenarios - #17713
Conversation
Control consumer group rebalances can rewind a live channel to an earlier committed offset. Skip records already handled by that channel so offsets remain monotonic and commit responses are not buffered twice. Add focused coverage for event handling and duplicate file registration across snapshots. Generated-by: OpenAI Codex
Generated-by: Codex
ajreid21
changed the title
Generated-by: Codex
Generated-by: Codex
| @@ -119,21 +120,32 @@ protected void send(List<Event> events, Map<TopicPartition, Offset> sourceOffset | |||
| protected void consumeAvailable(Duration pollDuration) { | |||
There was a problem hiding this comment.
Please note that there might be some unresolved cross-PR consolidation here. This exact replay-dedup guard is duplicated, byte-equivalent, in the earlier open #17376 and #17450 (same file, same method). Should we pick one vehicle for the fix and reconcile the others to avoid merge conflicts? The focused TestControlTopicReplay regression coverage added here is worth preserving, whichever PR lands. This is just a consolidation decision, not a code defect. Otherwise, looks good - thank you @ajreid21!
There was a problem hiding this comment.
I created this PR to target this specific case as it is a much simpler case (and fix) to reason about. This PR is not intended to try and fix all the potential coordinator replay and zombie issues -- these larger issues require much more discussion and larger PRs (like the 2 referenced) that may take many review cycles to get in.
I think it still makes sense to get this (ultimately, ~5 line code addition) in in the meantime to help prevent this particular issue from happening.
2 participants
Suppose one live Coordinator has consumed through offset 109:
Kafka committed offset: 100
Kafka consumer position: 110
Channel next offset: 110
CommitState: -> 109
A rebalance begins:
KafkaConsumer remains the same Java object
Channel remains the same Java object
CommitState remains populated
Under eager assignment, Kafka clears the partition assignment and its associated fetch position.
The same consumer then receives the same control topic again:
Kafka committed offset: 100
new Kafka consumer position: 100
Channel next offset: 110
CommitState: –>109
The next poll() can return offsets 100–109 again.