From 53720bbacaafc72e5ba71a86d3d41bc3f444f2e5 Mon Sep 17 00:00:00 2001 From: Alex Reid <5721775+ajreid21@users.noreply.github.com> Date: Mon, 17 Aug 2026 10:03:23 -0700 Subject: [PATCH 1/4] Kafka Connect: Ignore replayed control topic records Control consumer group rebalances can rewind a live channel to an earlier committed offset. Skip records already handled by that channel so offsets remain monotonic and commit responses are not buffered twice. Add focused coverage for event handling and duplicate file registration across snapshots. Generated-by: OpenAI Codex --- .../iceberg/connect/channel/Channel.java | 40 ++-- .../channel/TestControlTopicReplay.java | 215 ++++++++++++++++++ 2 files changed, 240 insertions(+), 15 deletions(-) create mode 100644 kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestControlTopicReplay.java diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java index 01cf165de66b..04b172a4bf2d 100644 --- a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java +++ b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java @@ -32,6 +32,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.kafka.clients.admin.Admin; import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.clients.producer.Producer; @@ -119,21 +120,30 @@ protected void send(List events, Map sourceOffset protected void consumeAvailable(Duration pollDuration) { ConsumerRecords records = consumer.poll(pollDuration); while (!records.isEmpty()) { - records.forEach( - record -> { - // the consumer stores the offsets that corresponds to the next record to consume, - // so increment the record offset by one - controlTopicOffsets.put(record.partition(), record.offset() + 1); - - Event event = AvroUtil.decode(record.value()); - - if (event.groupId().equals(connectGroupId)) { - LOG.debug("Received event of type: {}", event.type().name()); - if (receive(new Envelope(event, record.partition(), record.offset()))) { - LOG.info("Handled event of type: {}", event.type().name()); - } - } - }); + for (ConsumerRecord record : records) { + Long nextOffset = controlTopicOffsets.get(record.partition()); + // A consumer group rebalance can rewind the fetch position without recreating this channel. + if (nextOffset != null && record.offset() < nextOffset) { + LOG.debug( + "Skipping already-consumed control topic offset {} for partition {}", + record.offset(), + record.partition()); + continue; + } + + // the consumer stores the offsets that corresponds to the next record to consume, + // so increment the record offset by one + controlTopicOffsets.put(record.partition(), record.offset() + 1); + + Event event = AvroUtil.decode(record.value()); + + if (event.groupId().equals(connectGroupId)) { + LOG.debug("Received event of type: {}", event.type().name()); + if (receive(new Envelope(event, record.partition(), record.offset()))) { + LOG.info("Handled event of type: {}", event.type().name()); + } + } + } records = consumer.poll(pollDuration); } } diff --git a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestControlTopicReplay.java b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestControlTopicReplay.java new file mode 100644 index 000000000000..2eaaea8acf64 --- /dev/null +++ b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestControlTopicReplay.java @@ -0,0 +1,215 @@ +/* + * 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.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.UUID; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DataFiles; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.connect.IcebergSinkConfig; +import org.apache.iceberg.connect.events.AvroUtil; +import org.apache.iceberg.connect.events.DataComplete; +import org.apache.iceberg.connect.events.DataWritten; +import org.apache.iceberg.connect.events.Event; +import org.apache.iceberg.connect.events.StartCommit; +import org.apache.iceberg.connect.events.TableReference; +import org.apache.iceberg.connect.events.TopicPartitionOffset; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; +import org.apache.iceberg.types.Types.StructType; +import org.apache.kafka.clients.admin.MemberAssignment; +import org.apache.kafka.clients.admin.MemberDescription; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.connect.sink.SinkTaskContext; +import org.junit.jupiter.api.Test; + +class TestControlTopicReplay extends ChannelTestBase { + + @Test + void handlesReplayedControlRecordsOnce() { + TrackingChannel channel = + new TrackingChannel(config, clientFactory, mock(SinkTaskContext.class)); + channel.start(); + initConsumer(); + + addStartCommit(0L); + addStartCommit(1L); + channel.process(); + + assertThat(channel.receivedCount()).isEqualTo(2); + assertThat(channel.nextOffset(0)).isEqualTo(2L); + + consumer.seek(new TopicPartition(CTL_TOPIC_NAME, 0), 0L); + addStartCommit(0L); + addStartCommit(1L); + channel.process(); + + assertThat(channel.receivedCount()).isEqualTo(2); + assertThat(channel.nextOffset(0)).isEqualTo(2L); + } + + @Test + void doesNotCommitReplayedDataFilesTwice() throws IOException { + 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), + new TopicPartition(SRC_TOPIC_NAME, 2))); + MemberDescription member = + new MemberDescription(null, Optional.empty(), null, null, assignment); + 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 file1 = dataFile("path/to/file-1.parquet"); + DataFile file2 = dataFile("path/to/file-2.parquet"); + + addDataWritten(0L, commitId, file1); + addDataComplete(1L, commitId, 0); + addDataWritten(2L, commitId, file2); + addDataComplete(3L, commitId, 1); + coordinator.process(); + + consumer.seek(new TopicPartition(CTL_TOPIC_NAME, 0), 0L); + addDataWritten(0L, commitId, file1); + addDataComplete(1L, commitId, 0); + addDataWritten(2L, commitId, file2); + addDataComplete(3L, commitId, 1); + coordinator.process(); + + addDataComplete(4L, commitId, 2); + coordinator.process(); + + // Start another cycle and force a partial commit. Before the replay guard, the replayed tail + // remains buffered and this cycle commits file2 a second time. + when(config.commitTimeoutMs()).thenReturn(-1); + coordinator.process(); + + table.refresh(); + List snapshots = ImmutableList.copyOf(table.snapshots()); + assertThat(snapshots).hasSize(1); + assertThat(snapshots.get(0).summary()).containsEntry(OFFSETS_SNAPSHOT_PROP, "{\"0\":5}"); + + List locations = new ArrayList<>(); + try (CloseableIterable tasks = table.newScan().planFiles()) { + tasks.forEach(task -> locations.add(task.file().location())); + } + + assertThat(locations) + .containsExactlyInAnyOrder(file1.location().toString(), file2.location().toString()); + } + + private void addStartCommit(long offset) { + Event event = new Event(config.connectGroupId(), new StartCommit(UUID.randomUUID())); + addControlRecord(offset, event); + } + + private void addDataWritten(long offset, UUID commitId, DataFile file) { + Event event = + new Event( + config.connectGroupId(), + new DataWritten( + StructType.of(), + commitId, + TableReference.of("catalog", TABLE_IDENTIFIER, table.uuid()), + ImmutableList.of(file), + ImmutableList.of())); + addControlRecord(offset, event); + } + + private void addDataComplete(long offset, UUID commitId, int sourcePartition) { + Event event = + new Event( + config.connectGroupId(), + new DataComplete( + commitId, + ImmutableList.of( + new TopicPartitionOffset( + SRC_TOPIC_NAME, sourcePartition, 1L, EventTestUtil.now())))); + addControlRecord(offset, event); + } + + private void addControlRecord(long offset, Event event) { + consumer.addRecord( + new ConsumerRecord<>(CTL_TOPIC_NAME, 0, offset, "key", AvroUtil.encode(event))); + } + + private DataFile dataFile(String location) { + return DataFiles.builder(PartitionSpec.unpartitioned()) + .withPath(location) + .withFormat(FileFormat.PARQUET) + .withFileSizeInBytes(100L) + .withRecordCount(1L) + .build(); + } + + private static class TrackingChannel extends Channel { + private int receivedCount = 0; + + private TrackingChannel( + IcebergSinkConfig config, KafkaClientFactory clientFactory, SinkTaskContext context) { + super("tracking", "tracking-group", config, clientFactory, context); + } + + @Override + protected boolean receive(Envelope envelope) { + receivedCount += 1; + return true; + } + + private void process() { + consumeAvailable(Duration.ZERO); + } + + private int receivedCount() { + return receivedCount; + } + + private Long nextOffset(int partition) { + return controlTopicOffsets().get(partition); + } + } +} From 58dc592a1b79d861d8094235cd6679116f8dc1be Mon Sep 17 00:00:00 2001 From: Alex Reid <5721775+ajreid21@users.noreply.github.com> Date: Tue, 18 Aug 2026 12:51:15 -0700 Subject: [PATCH 2/4] Kafka Connect: Clarify control offset rewind comment Generated-by: Codex --- .../main/java/org/apache/iceberg/connect/channel/Channel.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java index 04b172a4bf2d..127830d1996b 100644 --- a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java +++ b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java @@ -122,7 +122,9 @@ protected void consumeAvailable(Duration pollDuration) { while (!records.isEmpty()) { for (ConsumerRecord record : records) { Long nextOffset = controlTopicOffsets.get(record.partition()); - // A consumer group rebalance can rewind the fetch position without recreating this channel. + // A rebalance may revoke and later reassign this control partition to the same + // consumer. Kafka then initializes it from the group's committed offset, which + // can lag this channel's retained in-memory position. if (nextOffset != null && record.offset() < nextOffset) { LOG.debug( "Skipping already-consumed control topic offset {} for partition {}", From e390f70cd198ef5ecd83117493e579d0cfa557b4 Mon Sep 17 00:00:00 2001 From: Alex Reid <5721775+ajreid21@users.noreply.github.com> Date: Tue, 18 Aug 2026 13:41:04 -0700 Subject: [PATCH 3/4] Kafka Connect: Apply test formatting Generated-by: Codex --- .../iceberg/connect/channel/TestControlTopicReplay.java | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestControlTopicReplay.java b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestControlTopicReplay.java index 2eaaea8acf64..8677e5fb5166 100644 --- a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestControlTopicReplay.java +++ b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestControlTopicReplay.java @@ -93,11 +93,7 @@ void doesNotCommitReplayedDataFilesTwice() throws IOException { new MemberDescription(null, Optional.empty(), null, null, assignment); Coordinator coordinator = new Coordinator( - catalog, - config, - ImmutableList.of(member), - clientFactory, - mock(SinkTaskContext.class)); + catalog, config, ImmutableList.of(member), clientFactory, mock(SinkTaskContext.class)); coordinator.start(); initConsumer(); coordinator.process(); From b987b9c1784870d0c2ba499337ff4ffcec36a65b Mon Sep 17 00:00:00 2001 From: Alex Reid <5721775+ajreid21@users.noreply.github.com> Date: Tue, 18 Aug 2026 14:07:46 -0700 Subject: [PATCH 4/4] Kafka Connect: Fix replay test checkstyle Generated-by: Codex --- .../iceberg/connect/channel/TestControlTopicReplay.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestControlTopicReplay.java b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestControlTopicReplay.java index 8677e5fb5166..7f35ccbd593b 100644 --- a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestControlTopicReplay.java +++ b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestControlTopicReplay.java @@ -24,7 +24,6 @@ import java.io.IOException; import java.time.Duration; -import java.util.ArrayList; import java.util.List; import java.util.Optional; import java.util.UUID; @@ -45,6 +44,7 @@ import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.types.Types.StructType; import org.apache.kafka.clients.admin.MemberAssignment; import org.apache.kafka.clients.admin.MemberDescription; @@ -129,7 +129,7 @@ void doesNotCommitReplayedDataFilesTwice() throws IOException { assertThat(snapshots).hasSize(1); assertThat(snapshots.get(0).summary()).containsEntry(OFFSETS_SNAPSHOT_PROP, "{\"0\":5}"); - List locations = new ArrayList<>(); + List locations = Lists.newArrayList(); try (CloseableIterable tasks = table.newScan().planFiles()) { tasks.forEach(task -> locations.add(task.file().location())); }