nahidupa commented on code in PR #17925:
URL: https://github.com/apache/iceberg/pull/17925#discussion_r3967736498
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -129,6 +137,272 @@ public void testCommitNoFiles() {
assertThat(table.snapshots()).isEmpty();
}
+ @Test
+ void retainsBufferedFilesWhenRebalanceResetsToLatest() {
Review Comment:
Correct: that test did not exercise the dispatch guard. It was removed when
the PR was narrowed to readiness counting. The replacement coordinator test
delivers duplicate payloads explicitly and fails without partition
deduplication, but it uses newer offsets rather than a same-offset rewind. I
have called out that remaining coverage gap rather than treating it as a
rebalance test.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -129,6 +137,272 @@ public void testCommitNoFiles() {
assertThat(table.snapshots()).isEmpty();
}
+ @Test
+ void retainsBufferedFilesWhenRebalanceResetsToLatest() {
+ when(config.commitIntervalMs()).thenReturn(0);
+ when(config.commitTimeoutMs()).thenReturn(Integer.MAX_VALUE);
+
+ this.consumer = new MockConsumer<>(OffsetResetStrategy.LATEST);
+ when(clientFactory.createConsumer(any())).thenReturn(consumer);
+ TopicPartition controlPartition = new TopicPartition(CTL_TOPIC_NAME, 0);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ Coordinator coordinator =
+ new Coordinator(catalog, config, ImmutableList.of(), clientFactory,
context);
+ coordinator.start();
+ consumer.rebalance(ImmutableList.of(controlPartition));
+ consumer.updateEndOffsets(ImmutableMap.of(controlPartition, 0L));
+ assertThat(consumer.position(controlPartition)).isZero();
+
+ coordinator.process();
+ assertThat(producer.history()).hasSize(1);
+ UUID commitId =
+ ((StartCommit)
AvroUtil.decode(producer.history().get(0).value()).payload()).commitId();
+
+ DataFile dataFile = EventTestUtil.createDataFile();
+ Event commitResponse =
+ new Event(
+ config.connectGroupId(),
+ new DataWritten(
+ StructType.of(),
+ commitId,
+ TableReference.of("catalog", TABLE_IDENTIFIER, table.uuid()),
+ ImmutableList.of(dataFile),
+ ImmutableList.of()));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key",
AvroUtil.encode(commitResponse)));
+ coordinator.process();
+
+ assertThat(producer.history()).hasSize(1);
+
assertThat(consumer.committed(ImmutableSet.of(controlPartition))).isEmpty();
+
+ consumer.updateEndOffsets(ImmutableMap.of(controlPartition, 2L));
+ consumer.rebalance(ImmutableList.of());
+ consumer.rebalance(ImmutableList.of(controlPartition));
+ assertThat(consumer.position(controlPartition)).isEqualTo(2L);
+
+ when(config.commitTimeoutMs()).thenReturn(-1);
+ coordinator.process();
+
+ table.refresh();
+ assertThat(table.snapshots()).hasSize(1);
+ SnapshotChanges changes =
+
SnapshotChanges.builderFor(table).snapshot(table.currentSnapshot()).build();
+ assertThat(changes.addedDataFiles())
+ .extracting(DataFile::location)
+ .containsExactly(dataFile.location());
+ assertThat(table.currentSnapshot().summary())
+ .containsEntry(COMMIT_ID_SNAPSHOT_PROP, commitId.toString())
+ .containsEntry(OFFSETS_SNAPSHOT_PROP, "{\"0\":2}");
+ assertCommitTable(1, commitId, null);
+ assertCommitComplete(2, commitId, null);
+ }
+
+ @Test
+ void commitsReplayedFilesExactlyOnce() {
+ when(config.commitIntervalMs()).thenReturn(0);
+ when(config.commitTimeoutMs()).thenReturn(Integer.MAX_VALUE);
+
+ 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();
+ TopicPartition controlPartition = new TopicPartition(CTL_TOPIC_NAME, 0);
+ consumer.commitSync(ImmutableMap.of(controlPartition, new
OffsetAndMetadata(1L)));
+
+ coordinator.process();
+ assertThat(producer.history()).hasSize(1);
+ UUID commitId =
+ ((StartCommit)
AvroUtil.decode(producer.history().get(0).value()).payload()).commitId();
+
+ OffsetDateTime ts = EventTestUtil.now();
+ DataFile firstFile = EventTestUtil.createDataFile();
+ Event dataWritten =
Review Comment:
Both referenced tests were removed in the rebuild, along with the helper
added for them. The current coordinator regression constructs one `DataWritten`
event inline, so the repeated construction identified in the earlier diff is no
longer present.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -129,6 +137,272 @@ public void testCommitNoFiles() {
assertThat(table.snapshots()).isEmpty();
}
+ @Test
+ void retainsBufferedFilesWhenRebalanceResetsToLatest() {
+ when(config.commitIntervalMs()).thenReturn(0);
+ when(config.commitTimeoutMs()).thenReturn(Integer.MAX_VALUE);
+
+ this.consumer = new MockConsumer<>(OffsetResetStrategy.LATEST);
+ when(clientFactory.createConsumer(any())).thenReturn(consumer);
+ TopicPartition controlPartition = new TopicPartition(CTL_TOPIC_NAME, 0);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ Coordinator coordinator =
+ new Coordinator(catalog, config, ImmutableList.of(), clientFactory,
context);
+ coordinator.start();
+ consumer.rebalance(ImmutableList.of(controlPartition));
+ consumer.updateEndOffsets(ImmutableMap.of(controlPartition, 0L));
+ assertThat(consumer.position(controlPartition)).isZero();
+
+ coordinator.process();
+ assertThat(producer.history()).hasSize(1);
+ UUID commitId =
+ ((StartCommit)
AvroUtil.decode(producer.history().get(0).value()).payload()).commitId();
+
+ DataFile dataFile = EventTestUtil.createDataFile();
+ Event commitResponse =
+ new Event(
+ config.connectGroupId(),
+ new DataWritten(
+ StructType.of(),
+ commitId,
+ TableReference.of("catalog", TABLE_IDENTIFIER, table.uuid()),
+ ImmutableList.of(dataFile),
+ ImmutableList.of()));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key",
AvroUtil.encode(commitResponse)));
+ coordinator.process();
+
+ assertThat(producer.history()).hasSize(1);
+
assertThat(consumer.committed(ImmutableSet.of(controlPartition))).isEmpty();
+
+ consumer.updateEndOffsets(ImmutableMap.of(controlPartition, 2L));
+ consumer.rebalance(ImmutableList.of());
+ consumer.rebalance(ImmutableList.of(controlPartition));
+ assertThat(consumer.position(controlPartition)).isEqualTo(2L);
+
+ when(config.commitTimeoutMs()).thenReturn(-1);
+ coordinator.process();
+
+ table.refresh();
+ assertThat(table.snapshots()).hasSize(1);
+ SnapshotChanges changes =
+
SnapshotChanges.builderFor(table).snapshot(table.currentSnapshot()).build();
+ assertThat(changes.addedDataFiles())
+ .extracting(DataFile::location)
+ .containsExactly(dataFile.location());
+ assertThat(table.currentSnapshot().summary())
+ .containsEntry(COMMIT_ID_SNAPSHOT_PROP, commitId.toString())
+ .containsEntry(OFFSETS_SNAPSHOT_PROP, "{\"0\":2}");
+ assertCommitTable(1, commitId, null);
+ assertCommitComplete(2, commitId, null);
+ }
+
+ @Test
+ void commitsReplayedFilesExactlyOnce() {
+ when(config.commitIntervalMs()).thenReturn(0);
+ when(config.commitTimeoutMs()).thenReturn(Integer.MAX_VALUE);
+
+ 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();
+ TopicPartition controlPartition = new TopicPartition(CTL_TOPIC_NAME, 0);
+ consumer.commitSync(ImmutableMap.of(controlPartition, new
OffsetAndMetadata(1L)));
+
+ coordinator.process();
+ assertThat(producer.history()).hasSize(1);
+ UUID commitId =
+ ((StartCommit)
AvroUtil.decode(producer.history().get(0).value()).payload()).commitId();
+
+ OffsetDateTime ts = EventTestUtil.now();
+ DataFile firstFile = EventTestUtil.createDataFile();
+ Event dataWritten =
+ new Event(
+ config.connectGroupId(),
+ new DataWritten(
+ StructType.of(),
+ commitId,
+ TableReference.of("catalog", TABLE_IDENTIFIER, table.uuid()),
+ ImmutableList.of(firstFile),
+ 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();
+
+ assertThat(producer.history()).hasSize(1);
+
+ consumer.rebalance(ImmutableList.of());
+ consumer.rebalance(ImmutableList.of(controlPartition));
+ assertThat(consumer.position(controlPartition)).isEqualTo(1L);
+ 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();
+
+ assertThat(producer.history()).hasSize(1);
+ assertThat(table.snapshots()).isEmpty();
+
+ DataFile secondFile =
+ DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath("path/to/second-file.parquet")
+ .withFormat(FileFormat.PARQUET)
+ .withRecordCount(firstFile.recordCount())
+ .withFileSizeInBytes(firstFile.fileSizeInBytes())
+ .build();
+ Event secondDataWritten = dataWrittenEvent(commitId, secondFile);
+ Event secondDataComplete =
+ new Event(
+ config.connectGroupId(),
+ new DataComplete(
+ commitId, ImmutableList.of(new
TopicPartitionOffset(SRC_TOPIC_NAME, 1, 4L, ts))));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 3, "key",
AvroUtil.encode(secondDataWritten)));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 4, "key",
AvroUtil.encode(secondDataComplete)));
+ coordinator.process();
+
+ assertThat(producer.history()).hasSize(3);
+ assertCommitTable(1, commitId, ts);
+ assertCommitComplete(2, commitId, ts);
+ table.refresh();
+ Snapshot snapshot = table.currentSnapshot();
+ assertThat(table.snapshots()).hasSize(1);
+
assertThat(SnapshotChanges.builderFor(table).snapshot(snapshot).build().addedDataFiles())
+ .extracting(DataFile::location)
+ .containsExactlyInAnyOrder(firstFile.location(),
secondFile.location());
+ assertThat(snapshot.summary())
+ .containsEntry(OFFSETS_SNAPSHOT_PROP, "{\"0\":5}")
+ .containsEntry(VALID_THROUGH_TS_SNAPSHOT_PROP, ts.toString());
+ CommitToTable committed =
+ (CommitToTable)
AvroUtil.decode(producer.history().get(1).value()).payload();
+ assertThat(committed.snapshotId()).isEqualTo(snapshot.snapshotId());
+
+ consumer.seek(controlPartition, 1L);
+ 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)));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 3, "key",
AvroUtil.encode(secondDataWritten)));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 4, "key",
AvroUtil.encode(secondDataComplete)));
+ when(config.commitTimeoutMs()).thenReturn(-1);
+ coordinator.process();
+
+ table.refresh();
+ assertThat(table.snapshots()).hasSize(1);
+
assertThat(table.currentSnapshot().snapshotId()).isEqualTo(snapshot.snapshotId());
+ assertThat(producer.history())
+ .filteredOn(record -> AvroUtil.decode(record.value()).type() ==
PayloadType.COMMIT_TO_TABLE)
+ .hasSize(1);
+ }
+
+ @Test
+ void retainsFilesDuringPartialControlReplay() {
+ assertPartialControlReplay(false);
+ }
+
+ @Test
+ void retainsFilesWithRetainedControlPartition() {
+ assertPartialControlReplay(true);
+ }
+
+ private void assertPartialControlReplay(boolean retainSecondPartition) {
+ when(config.commitIntervalMs()).thenReturn(0);
+ when(config.commitTimeoutMs()).thenReturn(Integer.MAX_VALUE);
+
+ Coordinator coordinator =
+ new Coordinator(
+ catalog, config, ImmutableList.of(), clientFactory,
mock(SinkTaskContext.class));
+ coordinator.start();
+ TopicPartition firstPartition = new TopicPartition(CTL_TOPIC_NAME, 0);
+ TopicPartition secondPartition = new TopicPartition(CTL_TOPIC_NAME, 1);
+ consumer.rebalance(ImmutableList.of(firstPartition, secondPartition));
+ consumer.updateBeginningOffsets(ImmutableMap.of(firstPartition, 1L,
secondPartition, 1L));
+ consumer.commitSync(
+ ImmutableMap.of(
+ firstPartition, new OffsetAndMetadata(1L),
+ secondPartition, new OffsetAndMetadata(1L)));
+ coordinator.process();
+ UUID commitId =
+ ((StartCommit)
AvroUtil.decode(producer.history().get(0).value()).payload()).commitId();
+
+ DataFile firstFile = EventTestUtil.createDataFile();
+ DataFile secondFile =
+ DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath("path/to/second-file.parquet")
+ .withFormat(FileFormat.PARQUET)
+ .withRecordCount(firstFile.recordCount())
+ .withFileSizeInBytes(firstFile.fileSizeInBytes())
+ .build();
+ Event firstResponse = dataWrittenEvent(commitId, firstFile);
+ Event secondResponse = dataWrittenEvent(commitId, secondFile);
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key",
AvroUtil.encode(firstResponse)));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 1, 1, "key",
AvroUtil.encode(secondResponse)));
+ coordinator.process();
+ assertThat(producer.history()).hasSize(1);
+
+ consumer.rebalance(
+ retainSecondPartition ? ImmutableList.of(secondPartition) :
ImmutableList.of());
+ consumer.rebalance(ImmutableList.of(firstPartition, secondPartition));
+ assertThat(consumer.position(firstPartition)).isEqualTo(1L);
+
assertThat(consumer.position(secondPartition)).isEqualTo(retainSecondPartition
? 2L : 1L);
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key",
AvroUtil.encode(firstResponse)));
Review Comment:
Agreed that the old offset arrangement did not distinguish the guard from
the unguarded path. Those cases were removed, and this PR no longer changes
offset tracking or dispatch.
The new readiness tests have been checked with a negative control: replacing
the reported-partition set with a duplicate-accepting list fails exactly
`testReplayedReadyDoesNotSatisfyQuorumTwice`,
`testOverlappingAssignmentsDoNotSatisfyQuorumTwice`, and
`testReplayedDataCompleteStillCommitsTheFileExactlyOnce`; the other 21 focused
tests pass. Restoring the set passes all 24. That demonstrates sensitivity to
partition double-counting, not coverage of an actual Kafka rewind. The full
validation results and limitations are recorded in the summary comment.
--
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]
