nahidupa commented on code in PR #17925:
URL: https://github.com/apache/iceberg/pull/17925#discussion_r3978479358
##########
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:
Follow-up on the negative-control results above. Those applied to the
previous revision, where readiness compared `reportedPartitions.size()` against
an expected count — replacing the set with a duplicate-accepting list inflated
that size and produced the reported failures. That evidence was valid for that
implementation.
In `65320ff8f` readiness requires
`reportedPartitions.containsAll(expectedPartitions)`. Duplicates do not change
a membership check, so the same Set-to-List mutation would now pass and is no
longer a discriminating control. The set retains efficient lookup and avoids
storing duplicate identities, but it is `containsAll` that establishes
readiness.
Controls re-run against the current implementation: restoring cardinality
readiness fails both unexpected-partition tests; restoring duplicate-accepting
counting fails the replay unit test and both coordinator replay variants (the
unit test fails its readiness assertion; both coordinator variants produce
premature snapshots carrying the later timestamp); removing the commit-id guard
fails the zombie test; dropping topic identity fails the cross-topic test. All
mutations were reverted and source hashes confirmed identical. Full connector
check: 149 tests, zero failures.
One clarification to the description: a stale-id `DataComplete` still enters
`readyBuffer`, so it can lower `validThroughTs` *or suppress it entirely*,
since `hasValidThroughTs` requires a non-null timestamp across the whole
buffer. Excluding stale-commit entries from the `validThroughTs` calculation
was proposed in #17080, which was closed unmerged. That change filters the
calculation, not insertion into `readyBuffer`; treating it as a separate
concern 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]
