wombatu-kun commented on code in PR #17925:
URL: https://github.com/apache/iceberg/pull/17925#discussion_r3948388801
##########
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 second paragraph is implementation rationale rather than the method's
contract, and `That is what we want` uses a personal pronoun, which AGENTS.md
rules out in comments. Drop it - the reasoning belongs in the PR description.
##########
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:
No test method under `org.apache.iceberg.connect` carries a Javadoc block;
the house style here is a one-to-three line `//` comment placed at the step it
explains. Trim this to the line that says what the test drives and leave the
mechanism to the PR description.
##########
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);
+ }
+
+ /**
+ * A control-topic rebalance revokes the coordinator's assignment and,
because the coordinator's
+ * consumer resumes from its last <em>committed</em> offset, the coordinator
re-reads every
+ * control-topic record it had already consumed. This test models that
rewind: after a rebalance,
+ * the same {@code DataWritten}/{@code DataComplete} pair for a commit is
delivered again.
+ *
+ * <p>The coordinator expects responses from {@code totalPartitionCount}
partitions (two here).
+ * Before the rebalance it has heard from exactly one (partition 0), so the
commit is in flight
+ * with a readiness count of one. With the in-flight commit state reset (the
fix), the re-read
+ * {@code DataComplete} is a stale event for a commit that no longer exists,
so {@link
+ * CommitState#addReady} ignores it and the coordinator never concludes it
has heard from both
+ * partitions. Without the reset, the stale {@code DataComplete} carries the
same commit id as the
+ * still-in-flight commit, so it is counted a second time and the
coordinator fires a commit that
+ * is missing half its data and stamps a watermark the table does not yet
satisfy.
+ */
+ @Test
+ public void testControlPartitionsRevokedRewindDoesNotDoubleCount() {
+ when(config.commitIntervalMs()).thenReturn(0);
+ when(config.commitTimeoutMs()).thenReturn(Integer.MAX_VALUE);
+
+ // two source partitions, so a commit is only ready once both have reported
+ MemberAssignment assignment =
+ new MemberAssignment(
+ ImmutableSet.of(
+ new TopicPartition(SRC_TOPIC_NAME, 0), new
TopicPartition(SRC_TOPIC_NAME, 1)));
+ MemberDescription member =
+ new MemberDescription(null, Optional.empty(), null, null, assignment);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ Coordinator coordinator =
+ new Coordinator(catalog, config, ImmutableList.of(member),
clientFactory, context);
+ coordinator.start();
+ initConsumer();
+
+ // begin a commit and deliver partition 0's DataWritten + DataComplete, so
the commit is in
+ // flight with a readiness count of one (of two)
+ coordinator.process();
+ assertThat(producer.history()).hasSize(1);
+ UUID commitId =
+ ((StartCommit)
AvroUtil.decode(producer.history().get(0).value()).payload()).commitId();
+
+ OffsetDateTime ts = EventTestUtil.now();
+ Event dataWritten =
+ new Event(
+ config.connectGroupId(),
+ new DataWritten(
+ StructType.of(),
+ commitId,
+ TableReference.of("catalog", TableIdentifier.of("db", "tbl"),
null),
+ ImmutableList.of(EventTestUtil.createDataFile()),
+ ImmutableList.of()));
+ Event dataComplete =
+ new Event(
+ config.connectGroupId(),
+ new DataComplete(
+ commitId, ImmutableList.of(new
TopicPartitionOffset(SRC_TOPIC_NAME, 0, 3L, ts))));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key",
AvroUtil.encode(dataWritten)));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 2, "key",
AvroUtil.encode(dataComplete)));
+ coordinator.process();
+
+ // still mid-commit: only the StartCommit has been emitted
+ assertThat(producer.history()).hasSize(1);
+
+ // a control-topic rebalance revokes the partition; the in-flight commit
must be discarded
+ consumer.rebalance(ImmutableList.of());
+
+ // the consumer rewinds and re-delivers the same DataWritten +
DataComplete pair
+ consumer.rebalance(ImmutableList.of(new TopicPartition(CTL_TOPIC_NAME,
0)));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key",
AvroUtil.encode(dataWritten)));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 2, "key",
AvroUtil.encode(dataComplete)));
+ coordinator.process();
+
+ // The coordinator has heard from exactly one partition (partition 0),
delivered twice. With the
+ // reset, the re-read pair is stale and ignored, so no CommitToTable is
emitted. Without the
+ // reset, the stale DataComplete would push the readiness count to 2 and a
CommitToTable (and
+ // CommitComplete) would appear here.
+ assertThat(producer.history())
+ .noneMatch(record -> AvroUtil.decode(record.value()).type() ==
PayloadType.COMMIT_TO_TABLE);
Review Comment:
Both new tests assert only that nothing was committed early, so nothing
covers that the replayed `DataWritten` still reaches the table exactly once
after the rebalance. Deliver both source partitions' `DataComplete` for the new
commit id and assert the file lands in a single snapshot.
##########
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:
Dropping `commitBuffer` assumes every discarded `DataWritten` is re-read,
but `createConsumer` leaves `auto.offset.reset` at `latest` and the `-coord`
group only gets a committed offset inside `doCommit`, so a revoke before that
group's first successful commit rewinds to the log end instead. Those files
already had their source offsets committed by the worker's transaction - is
that window intentional?
--
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]
