nahidupa commented on code in PR #17925:
URL: https://github.com/apache/iceberg/pull/17925#discussion_r3967730104
##########
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
Review Comment:
The `reset()` method was removed when the reset-on-rebalance approach was
withdrawn, so this Javadoc and the personal pronoun are no longer present. The
remaining implementation rationale is a comment beside `addReady`; the method
contract is not being used to explain the design history.
##########
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:
Following up on both the original recovery concern and your later
consolidation/snapshot-expiration comment: `2e9d2e927` is rebuilt on main and
removes both `reset()` and the dispatch guard. This PR no longer introduces
buffer discard on rebalance; normal successful-commit cleanup is unchanged.
`Channel.java` is untouched relative to the new base.
With the default `latest` and no valid coordinator checkpoint, a replacement
can still skip retained announcements buffered by its predecessor. #18006
addresses that reset default separately; it is not solved by the readiness
change here.
Agreed on the durable boundary. If no reachable ancestor retains the
per-table connector offset summary, the lookup returns an empty map; ordinary
snapshots do not automatically propagate that custom property. Replaying
retained announcements after that boundary is lost can register an
already-committed file again. That limitation is documented on #18006 and
remains outside both changes; `earliest` alone does not solve it.
The remaining readiness scope and the consolidation question are summarized
here: https://github.com/apache/iceberg/pull/17925#issuecomment-5600930892
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -129,6 +133,138 @@ public void testCommitNoFiles() {
assertThat(table.snapshots()).isEmpty();
}
+ @Test
+ public void testControlPartitionsRevokedResetsInFlightCommit() {
+ when(config.commitIntervalMs()).thenReturn(0);
+ when(config.commitTimeoutMs()).thenReturn(Integer.MAX_VALUE);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ Coordinator coordinator =
+ new Coordinator(catalog, config, ImmutableList.of(), clientFactory,
context);
+ coordinator.start();
+ initConsumer();
+
+ // begin a commit and buffer a worker response, but withhold DATA_COMPLETE
so the commit
+ // stays in flight
+ coordinator.process();
+ assertThat(producer.history()).hasSize(1);
+ UUID commitId =
+ ((StartCommit)
AvroUtil.decode(producer.history().get(0).value()).payload()).commitId();
+
+ Event commitResponse =
+ new Event(
+ config.connectGroupId(),
+ new DataWritten(
+ StructType.of(),
+ commitId,
+ TableReference.of("catalog", TableIdentifier.of("db", "tbl"),
null),
+ ImmutableList.of(EventTestUtil.createDataFile()),
+ ImmutableList.of()));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key",
AvroUtil.encode(commitResponse)));
+ coordinator.process();
+
+ // still mid-commit: no further event emitted
+ assertThat(producer.history()).hasSize(1);
+
+ // a control-topic rebalance revokes the partition; the in-flight commit
must be discarded
+ consumer.rebalance(ImmutableList.of());
+
+ // with the in-flight commit reset, the coordinator is free to start a
brand new commit on the
+ // next cycle. Without the reset it would still consider commit `commitId`
in progress and emit
+ // nothing here.
+ coordinator.process();
+
+ assertThat(producer.history()).hasSize(2);
+ Event newStart = AvroUtil.decode(producer.history().get(1).value());
+ assertThat(newStart.type()).isEqualTo(PayloadType.START_COMMIT);
+ assertThat(((StartCommit)
newStart.payload()).commitId()).isNotEqualTo(commitId);
+ }
+
+ /**
Review Comment:
The reset-era test containing this Javadoc was removed in the rebuild. The
replacement readiness tests use short comments at the relevant steps rather
than test-method Javadocs.
--
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]
