wombatu-kun commented on code in PR #17925:
URL: https://github.com/apache/iceberg/pull/17925#discussion_r3975123221
##########
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:
Both new tests drive `addReady` with the same observable input - two
`DataComplete`s each yielding `("src-topic", 0)`, since production reads only
`topic()` and `partition()`. Give the overlapping case a second topic so it
also pins that the key is `(topic, partition)` and not the partition number.
##########
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:
The set makes coverage checkable, but `isCommitReady` still compares
cardinality against `Coordinator`'s summed `totalPartitionCount`, so a
partition no worker owns any more can still stand in for one that never
reported. Was passing the expected `Set<TopicPartition>` down and checking
`containsAll` considered here?
##########
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:
With the zombie payload and the current payload both carrying partition 0,
this test now passes with the commit-id guard in `addReady` deleted. Give the
zombie payload partitions the current one does not report, so a broken guard
still fails the `isCommitReady(2)` assertion.
##########
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:
Both `DataComplete` payloads carry a null timestamp, so `validThroughTs`
stays null and no `kafka.connect.valid-through-ts` reaches the snapshot. Give
partition 0 a later timestamp than partition 1 and assert
`VALID_THROUGH_TS_SNAPSHOT_PROP` the way `testCommitAppend` does.
##########
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:
Everything from the config stubs down to `initConsumer` repeats
`startCoordinator`, which differs only in passing `ImmutableList.of()` for the
members. Add a `Collection<MemberDescription>` overload to `startCoordinator`
and call it here.
--
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]
