nahidupa commented on code in PR #17925:
URL: https://github.com/apache/iceberg/pull/17925#discussion_r3955926167
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java:
##########
@@ -102,6 +102,21 @@ void clearResponses() {
commitBuffer.clear();
}
+ /**
+ * Discard all in-flight commit state -- buffered responses, buffered ready
events, the readiness
+ * counter, and the current commit id. Used when a control-topic rebalance
invalidates the commit
+ * this coordinator was assembling; the underlying events remain on the
control topic and are
+ * re-read by whichever coordinator takes over.
+ *
+ * <p>{@code startTime} is deliberately left alone, so the coordinator
re-drives the abandoned
+ * commit on its next cycle rather than waiting out another full commit
interval. That is what we
+ * want when a rebalance interrupted a commit that was already due.
+ */
+ void reset() {
+ clearResponses();
Review Comment:
Thanks @wombatu-kun. Your review opened my eyes to a recovery assumption I
had missed. That data-loss window was **not intentional**, and my earlier
statement that discarded events would always be re-read was incorrect. With no
committed coordinator offset and `latest`, clearing the buffer can lose the
only remaining announcement for files whose source offsets the worker has
already committed.
The revised direction is to separate two problems:
- **Replay within the same coordinator instance:** preserve the in-flight
buffers and skip already-consumed control records per partition in
`consumeAvailable`, before updating offsets or dispatching events. This avoids
replay-induced readiness double-counting and offset regression without
discarding pending files. It overlaps with
[#17713](https://github.com/apache/iceberg/pull/17713), so consolidation makes
sense.
- **Recovery after coordinator replacement:** [New BUG Discovered]
[#18006](https://github.com/apache/iceberg/pull/18006) changes the
coordinator’s default to `earliest` when control-topic offsets are missing or
out of range, while retaining the worker default of `latest` and respecting
explicit overrides. The recovery tests check actual table contents, including
committed files mixed with pending files and a subsequent replay. Please have a
look at this PR too.
Your feedback also prompted a deeper investigation that uncovered additional
problems. Found a duplicate file registration when replay occurs after the
snapshot containing the connector’s offset boundary has expired, even though
the files remain live. This also helped identify a startup-memory risk from
buffering large retained histories before commit-time filtering. Those need
separate follow-ups; `earliest` alone does not make every recovery scenario
safe.
Would you recommend consolidating the same-instance replay fix into #17713
and keeping #18006 focused on replacement recovery? I’d appreciate any better
approach you see, particularly for preserving pending files while keeping
replay deduplication reliable across restarts and snapshot expiration.
--
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]
