nahidupa commented on code in PR #17925:
URL: https://github.com/apache/iceberg/pull/17925#discussion_r3978568782
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java:
##########
@@ -119,18 +131,18 @@ boolean isCommitReady(int expectedPartitionCount) {
return false;
}
- if (receivedPartitionCount >= expectedPartitionCount) {
+ if (reportedPartitions.size() >= expectedPartitionCount) {
Review Comment:
Implemented in 65320ff8f. `Coordinator` now captures an immutable
`Set<TopicPartition>` from the member assignments, and
`CommitState.isCommitReady` requires
`reportedPartitions.containsAll(expectedPartitions)`. An unexpected partition
can no longer substitute for a missing expected one.
Added `unexpectedPartitionsDoNotSatisfyReadiness` and
`unexpectedPartitionDoesNotCompleteCommit`; both fail when readiness is changed
back to a cardinality comparison. This checks coverage against the assignment
captured at coordinator construction, not its freshness after a rebalance or
topic expansion.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCommitState.java:
##########
@@ -60,16 +69,70 @@ public void testIsCommitReady() {
assertThat(commitState.isCommitReady(4)).isFalse();
}
+ @Test
+ public void testReplayedReadyDoesNotSatisfyQuorumTwice() {
+ CommitState commitState = new CommitState(mock(IcebergSinkConfig.class));
+ commitState.startNewCommit();
+
+ // one worker owning source partition 0 reports; a control-topic replay
redelivers it
+ TopicPartitionOffset tp0 = partition(0);
+ DataComplete payload = mock(DataComplete.class);
+ when(payload.commitId()).thenReturn(commitState.currentCommitId());
+ when(payload.assignments()).thenReturn(ImmutableList.of(tp0));
+
+ commitState.addReady(wrapInEnvelope(payload));
+ commitState.addReady(wrapInEnvelope(payload));
+
+ assertThat(commitState.isCommitReady(2))
+ .as("a redelivered response must not stand in for a partition that
never reported")
+ .isFalse();
+
+ // the partition that was actually missing reports
+ TopicPartitionOffset tp1 = partition(1);
+ DataComplete second = mock(DataComplete.class);
+ when(second.commitId()).thenReturn(commitState.currentCommitId());
+ when(second.assignments()).thenReturn(ImmutableList.of(tp1));
+ commitState.addReady(wrapInEnvelope(second));
+
+ assertThat(commitState.isCommitReady(2)).isTrue();
+ }
+
+ @Test
+ public void testOverlappingAssignmentsDoNotSatisfyQuorumTwice() {
Review Comment:
Addressed in 65320ff8f. `testOverlappingAssignmentsDoNotSatisfyQuorumTwice`
now expects `src-topic/0` and `other-topic/0`: two reports for `src-topic/0`
remain insufficient, and a report for `other-topic/0` completes that expected
set. It also checks that an additional expected `src-topic/1` is still missing.
The test fails when topic identity is dropped from the recorded key. That
gives it a distinct assertion beyond the duplicate-response case.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCommitState.java:
##########
@@ -89,7 +152,7 @@ public void testIsCommitReadyResetsBetweenCommits() {
@Test
public void testIsCommitReadyIgnoresZombieCoordinatorPayloads() {
- TopicPartitionOffset tp = mock(TopicPartitionOffset.class);
+ TopicPartitionOffset tp = partition(0);
Review Comment:
Addressed in 65320ff8f. The stale-ID payload now carries partition 1 while
the current-ID payload carries partition 0. Readiness for both partitions stays
false until partition 1 is reported with the current commit ID.
Removing the commit-ID guard makes
`testIsCommitReadyIgnoresZombieCoordinatorPayloads` fail. The guard was
restored after the negative-control run.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -480,6 +483,90 @@ private void assertCommitComplete(int idx, UUID commitId,
OffsetDateTime ts) {
assertThat(commitCompletePayload.validThroughTs()).isEqualTo(ts);
}
+ @Test
+ public void testReplayedDataCompleteStillCommitsTheFileExactlyOnce() {
Review Comment:
Addressed in 65320ff8f. Added
`startCoordinator(Collection<MemberDescription> members)` and made the
no-argument helper delegate with `ImmutableList.of()`. The assignment-aware
tests now use the overload, sharing the config stubs, coordinator startup, and
consumer initialization.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -480,6 +483,90 @@ private void assertCommitComplete(int idx, UUID commitId,
OffsetDateTime ts) {
assertThat(commitCompletePayload.validThroughTs()).isEqualTo(ts);
}
+ @Test
+ public void testReplayedDataCompleteStillCommitsTheFileExactlyOnce() {
+ 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
+ MemberDescription member =
+ new MemberDescription(
+ "member",
+ Optional.empty(),
+ "client",
+ "host",
+ new MemberAssignment(
+ ImmutableSet.of(
+ new TopicPartition(SRC_TOPIC_NAME, 0), new
TopicPartition(SRC_TOPIC_NAME, 1))));
+ Coordinator coordinator =
+ new Coordinator(
+ catalog, config, ImmutableList.of(member), clientFactory,
mock(SinkTaskContext.class));
+ coordinator.start();
+ initConsumer();
+
+ coordinator.process();
+ UUID commitId =
+ ((StartCommit)
AvroUtil.decode(producer.history().get(0).value()).payload()).commitId();
+
+ DataFile dataFile = EventTestUtil.createDataFile();
+ Event written =
+ new Event(
+ config.connectGroupId(),
+ new DataWritten(
+ StructType.of(),
+ commitId,
+ TableReference.of("catalog", TABLE_IDENTIFIER, table.uuid()),
+ ImmutableList.of(dataFile),
+ ImmutableList.of()));
+ Event firstPartitionReady =
+ new Event(
+ config.connectGroupId(),
+ new DataComplete(
+ commitId, ImmutableList.of(new
TopicPartitionOffset(SRC_TOPIC_NAME, 0, 1L, null))));
Review Comment:
Addressed in 65320ff8f. Partition 0 now reports a timestamp one second later
than partition 1. `repeatedDataCompleteWaitsForEveryExpectedPartition` asserts
that no snapshot, completion event, or checkpoint advance occurs while
partition 1 is missing, then checks that `VALID_THROUGH_TS_SNAPSHOT_PROP` and
both completion-event timestamps equal partition 1's earlier timestamp.
This runs for both a same-offset rewind and duplicate payloads at new
offsets. Restoring duplicate-accepting counting makes both coordinator variants
fail with premature snapshots carrying the later timestamp. All mutations were
reverted; the forced connector check passed 149 tests, with no live-broker
rebalance exercised.
--
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]
