kumarpritam863 commented on code in PR #17376:
URL: https://github.com/apache/iceberg/pull/17376#discussion_r3654729755
##########
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:
@emlynazuma I am adding a proper rebalance consumer listener but I do not
think it will be needed if the zombie coordinator scenario is properly handled
which after the deterministic changes that I have made will make it almost
impossible to have two coordinators running. But to be on the safer side I will
add a proper rebalance listener. Also I think the best place to clear thinks
are in the revoked partitions. As revoked path is called first if there is any
think to revoke so we can just clear the Data Written, Data Complete and
control topic offsets stored against those partitions. But I was just analyzing
for it to not become complicated. Let me know about your thoughts on these.
--
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]
