emlynazuma commented on code in PR #17376:
URL: https://github.com/apache/iceberg/pull/17376#discussion_r3654374491
##########
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:
Alternative to the `put()` → `merge(..., Long::max)` line, in case it's
useful — handling the rewind
in a `ConsumerRebalanceListener` so the replay never happens:
```java
@Override public void onPartitionsAssigned(Collection<TopicPartition>
partitions) {
for (TopicPartition partition : partitions) {
Long inMemoryOffset = controlTopicOffsets.get(partition.partition());
if (inMemoryOffset != null && inMemoryOffset >
consumer.position(partition)) {
consumer.seek(partition, inMemoryOffset); // + a warn
}
}
}
```
`merge(Long::max)` stops the duplicate files, but the replay still happens:
the replayed
`DataComplete` double-count in `readyBuffer`, so the commit still fires
early and `validThroughTs()`
takes `min(timestamp)` over only the partitions that reported — the snapshot
can claim a
`valid-through-ts` later than the true watermark. Seeking forward avoids
that, and is safe because
the skipped records were consumed by this same `Channel` and are already in
`commitBuffer`.
Context on #16282. Branch on top of 1.11.0 with regression tests:
https://github.com/apache/iceberg/compare/apache-iceberg-1.11.0...emlynazuma:iceberg:tongwai/ikc-1.11.0-rebalance-fixes
--
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]
