wombatu-kun commented on code in PR #17925:
URL: https://github.com/apache/iceberg/pull/17925#discussion_r3964288781
##########
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:
Both partial-replay cases pass with the guard removed: the replay is exactly
one offset behind, so the unguarded put rewrites the same value, and
`distinctByKey` collapses the duplicate envelope. Consume two records on the
replayed partition before the reassignment and replay only the first, so the
offsets map would actually regress and the snapshot-offsets assertion catches
it.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestChannel.java:
##########
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.tuple;
+import static org.mockito.Mockito.mock;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.UUID;
+import org.apache.iceberg.connect.IcebergSinkConfig;
+import org.apache.iceberg.connect.events.AvroUtil;
+import org.apache.iceberg.connect.events.Event;
+import org.apache.iceberg.connect.events.StartCommit;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.connect.sink.SinkTaskContext;
+import org.junit.jupiter.api.Test;
+
+class TestChannel extends ChannelTestBase {
Review Comment:
A `TestChannel` already exists at this path on `main`, and its
`controlTopicOffsetsTrackTheHighestPositionConsumed` asserts seven dispatched
envelopes after replaying two already-consumed offsets - this guard makes that
five. Fold these cases into that class and change the assertion deliberately,
since it currently encodes the opposite expectation.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -121,6 +121,20 @@ protected void consumeAvailable(Duration pollDuration) {
while (!records.isEmpty()) {
records.forEach(
record -> {
+ // A rebalance can reassign this partition to the same consumer,
which then resumes
+ // from the group's committed offset -- behind the position this
channel already
+ // reached. Skipping the re-delivered records before the offset
update keeps
+ // controlTopicOffsets monotonic, and skipping before dispatch
keeps a replayed
+ // DataComplete from being counted toward readiness a second time.
+ Long nextOffset = controlTopicOffsets.get(record.partition());
Review Comment:
`main` now merges this offset with `Long::max` in the same loop, so the
monotonicity half of the guard has already landed and only the
skip-before-dispatch is new here. Rebase onto that and lead with the replayed
`DataComplete` double-count in `CommitState.addReady`, which the merge does not
address.
##########
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:
No record is re-delivered here - `rebalance` clears the mock's buffer and
the LATEST reset lands the position at 2, where the earlier poll already left
it - so this never reaches the guard. Was the intent to pin buffer retention
rather than the skip, and is that still worth a test now that no rebalance
listener exists?
##########
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:
`dataWrittenEvent` in this class builds exactly this event, and this test
already calls it for the second file. Use it here and in
`retainsBufferedFilesWhenRebalanceResetsToLatest`.
--
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]
