emlynazuma commented on code in PR #17376:
URL: https://github.com/apache/iceberg/pull/17376#discussion_r3662308456
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -123,7 +151,7 @@ protected void consumeAvailable(Duration pollDuration) {
record -> {
// the consumer stores the offsets that corresponds to the next
record to consume,
// so increment the record offset by one
- controlTopicOffsets.put(record.partition(), record.offset() + 1);
+ controlTopicOffsets.merge(record.partition(), record.offset() + 1,
Long::max);
Review Comment:
Looks like this landed in `onPartitionsAssigned` via `seekToTrackedOffsets`
rather than clearing on
revoke — matches what I raised above about revoke-time clearing risking
dropping buffered records
that haven't committed yet. And it's still doing work independent of the
fencing: `open()`/`close()`
starting and stopping the coordinator is itself a `-coord` membership
change, so the rewind can still
happen on a clean single-coordinator handoff, not just during a split-brain
window.
Will test this against our repro setup and report back.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
