You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by br...@apache.org on 2025/03/14 17:10:42 UTC
(iceberg) branch main updated: Kafka Connect: Handle no coordinator and data loss in ICR mode (#12372)
This is an automated email from the ASF dual-hosted git repository.
bryanck pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 51abab10e2 Kafka Connect: Handle no coordinator and data loss in ICR mode (#12372)
51abab10e2 is described below
commit 51abab10e22569250ade2b3726844ab7fa9bc0b6
Author: kumarpritam863 <14...@users.noreply.github.com>
AuthorDate: Fri Mar 14 22:40:36 2025 +0530
Kafka Connect: Handle no coordinator and data loss in ICR mode (#12372)
---
.../java/org/apache/iceberg/connect/Committer.java | 25 ++++
.../apache/iceberg/connect/IcebergSinkTask.java | 14 +-
.../iceberg/connect/channel/CommitterImpl.java | 147 +++++++++++++++------
.../apache/iceberg/connect/channel/KafkaUtils.java | 34 +++++
.../iceberg/connect/MockIcebergSinkTask.java} | 41 +++++-
.../iceberg/connect/channel/ChannelTestBase.java | 27 +++-
.../iceberg/connect/channel/CommitterImplTest.java | 4 +-
.../iceberg/connect/channel/CoordinatorTest.java | 23 ++++
8 files changed, 257 insertions(+), 58 deletions(-)
diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/Committer.java b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/Committer.java
index edc217d1b0..c0b813f040 100644
--- a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/Committer.java
+++ b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/Committer.java
@@ -20,13 +20,38 @@ package org.apache.iceberg.connect;
import java.util.Collection;
import org.apache.iceberg.catalog.Catalog;
+import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.connect.sink.SinkRecord;
import org.apache.kafka.connect.sink.SinkTaskContext;
public interface Committer {
+
+ /**
+ * @deprecated will be removed in 2.0.0. Use {@link #open(Catalog, IcebergSinkConfig,
+ * SinkTaskContext, Collection)} instead.
+ */
+ @Deprecated
void start(Catalog catalog, IcebergSinkConfig config, SinkTaskContext context);
+ default void open(
+ Catalog catalog,
+ IcebergSinkConfig config,
+ SinkTaskContext context,
+ Collection<TopicPartition> addedPartitions) {
+ // Default implementation calls the deprecated method. Implementations may override this.
+ start(catalog, config, context);
+ }
+
+ /**
+ * @deprecated will be removed in 2.0.0. Use {@link #close(Collection)} instead.
+ */
+ @Deprecated
void stop();
+ default void close(Collection<TopicPartition> closedPartitions) {
+ // Default implementation calls the deprecated method. Implementations may override this.
+ stop();
+ }
+
void save(Collection<SinkRecord> sinkRecords);
}
diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkTask.java b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkTask.java
index bb9370d3d5..e9428325d1 100644
--- a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkTask.java
+++ b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkTask.java
@@ -21,7 +21,6 @@ package org.apache.iceberg.connect;
import java.util.Collection;
import java.util.Map;
import org.apache.iceberg.catalog.Catalog;
-import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
@@ -46,26 +45,23 @@ public class IcebergSinkTask extends SinkTask {
@Override
public void start(Map<String, String> props) {
this.config = new IcebergSinkConfig(props);
+ this.catalog = CatalogUtils.loadCatalog(config);
+ this.committer = CommitterFactory.createCommitter(config);
}
@Override
public void open(Collection<TopicPartition> partitions) {
- Preconditions.checkArgument(catalog == null, "Catalog already open");
- Preconditions.checkArgument(committer == null, "Committer already open");
-
- catalog = CatalogUtils.loadCatalog(config);
- committer = CommitterFactory.createCommitter(config);
- committer.start(catalog, config, context);
+ committer.open(catalog, config, context, partitions);
}
@Override
public void close(Collection<TopicPartition> partitions) {
- close();
+ committer.close(partitions);
}
private void close() {
if (committer != null) {
- committer.stop();
+ committer.close(context.assignment());
committer = null;
}
diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java
index 53b7b76e8e..853a049f6b 100644
--- a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java
+++ b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java
@@ -20,7 +20,7 @@ package org.apache.iceberg.connect.channel;
import java.util.Collection;
import java.util.Comparator;
-import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.connect.Committer;
import org.apache.iceberg.connect.IcebergSinkConfig;
@@ -43,6 +43,24 @@ public class CommitterImpl implements Committer {
private CoordinatorThread coordinatorThread;
private Worker worker;
+ private Catalog catalog;
+ private IcebergSinkConfig config;
+ private SinkTaskContext context;
+ private KafkaClientFactory clientFactory;
+ private Collection<MemberDescription> membersWhenWorkerIsCoordinator;
+ private final AtomicBoolean isInitialized = new AtomicBoolean(false);
+
+ private void initialize(
+ Catalog icebergCatalog,
+ IcebergSinkConfig icebergSinkConfig,
+ SinkTaskContext sinkTaskContext) {
+ if (isInitialized.compareAndSet(false, true)) {
+ this.catalog = icebergCatalog;
+ this.config = icebergSinkConfig;
+ this.context = sinkTaskContext;
+ this.clientFactory = new KafkaClientFactory(config.kafkaProps());
+ }
+ }
static class TopicPartitionComparator implements Comparator<TopicPartition> {
@@ -56,55 +74,24 @@ public class CommitterImpl implements Committer {
}
}
- @Override
- public void start(Catalog catalog, IcebergSinkConfig config, SinkTaskContext context) {
- KafkaClientFactory clientFactory = new KafkaClientFactory(config.kafkaProps());
-
+ private boolean hasLeaderPartition(Collection<TopicPartition> currentAssignedPartitions) {
ConsumerGroupDescription groupDesc;
try (Admin admin = clientFactory.createAdmin()) {
groupDesc = KafkaUtils.consumerGroupDescription(config.connectGroupId(), admin);
}
-
if (groupDesc.state() == ConsumerGroupState.STABLE) {
Collection<MemberDescription> members = groupDesc.members();
- Set<TopicPartition> partitions = context.assignment();
- if (isLeader(members, partitions)) {
- LOG.info("Task elected leader, starting commit coordinator");
- Coordinator coordinator = new Coordinator(catalog, config, members, clientFactory, context);
- coordinatorThread = new CoordinatorThread(coordinator);
- coordinatorThread.start();
+ if (containsFirstPartition(members, currentAssignedPartitions)) {
+ membersWhenWorkerIsCoordinator = members;
+ return true;
}
}
-
- LOG.info("Starting commit worker");
- SinkWriter sinkWriter = new SinkWriter(catalog, config);
- worker = new Worker(config, clientFactory, sinkWriter, context);
- worker.start();
- }
-
- @Override
- public void save(Collection<SinkRecord> sinkRecords) {
- if (sinkRecords != null && !sinkRecords.isEmpty()) {
- worker.save(sinkRecords);
- }
- processControlEvents();
- }
-
- @Override
- public void stop() {
- if (worker != null) {
- worker.stop();
- worker = null;
- }
-
- if (coordinatorThread != null) {
- coordinatorThread.terminate();
- coordinatorThread = null;
- }
+ return false;
}
@VisibleForTesting
- boolean isLeader(Collection<MemberDescription> members, Collection<TopicPartition> partitions) {
+ boolean containsFirstPartition(
+ Collection<MemberDescription> members, Collection<TopicPartition> partitions) {
// there should only be one task assigned partition 0 of the first topic,
// so elect that one the leader
TopicPartition firstTopicPartition =
@@ -117,6 +104,55 @@ public class CommitterImpl implements Committer {
return partitions.contains(firstTopicPartition);
}
+ @Override
+ public void start(
+ Catalog icebergCatalog,
+ IcebergSinkConfig icebergSinkConfig,
+ SinkTaskContext sinkTaskContext) {
+ throw new UnsupportedOperationException(
+ "The method start(Catalog, IcebergSinkConfig, SinkTaskContext) is deprecated and will be removed in 2.0.0. "
+ + "Use start(Catalog, IcebergSinkConfig, SinkTaskContext, Collection<TopicPartition>) instead.");
+ }
+
+ @Override
+ public void open(
+ Catalog icebergCatalog,
+ IcebergSinkConfig icebergSinkConfig,
+ SinkTaskContext sinkTaskContext,
+ Collection<TopicPartition> addedPartitions) {
+ initialize(icebergCatalog, icebergSinkConfig, sinkTaskContext);
+ if (hasLeaderPartition(addedPartitions)) {
+ LOG.info("Committer received leader partition. Starting Coordinator.");
+ startCoordinator();
+ }
+ }
+
+ @Override
+ public void stop() {
+ throw new UnsupportedOperationException(
+ "The method stop() is deprecated and will be removed in 2.0.0. "
+ + "Use stop(Collection<TopicPartition>) instead.");
+ }
+
+ @Override
+ public void close(Collection<TopicPartition> closedPartitions) {
+ if (hasLeaderPartition(closedPartitions)) {
+ LOG.info("Committer lost leader partition. Stopping Coordinator.");
+ stopCoordinator();
+ }
+ stopWorker();
+ KafkaUtils.seekToLastCommittedOffsets(context);
+ }
+
+ @Override
+ public void save(Collection<SinkRecord> sinkRecords) {
+ if (sinkRecords != null && !sinkRecords.isEmpty()) {
+ startWorker();
+ worker.save(sinkRecords);
+ }
+ processControlEvents();
+ }
+
private void processControlEvents() {
if (coordinatorThread != null && coordinatorThread.isTerminated()) {
throw new NotRunningException("Coordinator unexpectedly terminated");
@@ -125,4 +161,37 @@ public class CommitterImpl implements Committer {
worker.process();
}
}
+
+ private void startWorker() {
+ if (null == this.worker) {
+ LOG.info("Starting commit worker");
+ SinkWriter sinkWriter = new SinkWriter(catalog, config);
+ worker = new Worker(config, clientFactory, sinkWriter, context);
+ worker.start();
+ }
+ }
+
+ private void startCoordinator() {
+ if (null == this.coordinatorThread) {
+ LOG.info("Task elected leader, starting commit coordinator");
+ Coordinator coordinator =
+ new Coordinator(catalog, config, membersWhenWorkerIsCoordinator, clientFactory, context);
+ coordinatorThread = new CoordinatorThread(coordinator);
+ coordinatorThread.start();
+ }
+ }
+
+ private void stopWorker() {
+ if (worker != null) {
+ worker.stop();
+ worker = null;
+ }
+ }
+
+ private void stopCoordinator() {
+ if (coordinatorThread != null) {
+ coordinatorThread.terminate();
+ coordinatorThread = null;
+ }
+ }
}
diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/KafkaUtils.java b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/KafkaUtils.java
index be51fff8bf..bea1c41539 100644
--- a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/KafkaUtils.java
+++ b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/KafkaUtils.java
@@ -18,6 +18,7 @@
*/
package org.apache.iceberg.connect.channel;
+import java.util.Map;
import java.util.concurrent.ExecutionException;
import org.apache.iceberg.common.DynFields;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
@@ -26,11 +27,17 @@ import org.apache.kafka.clients.admin.ConsumerGroupDescription;
import org.apache.kafka.clients.admin.DescribeConsumerGroupsResult;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.sink.SinkTaskContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
class KafkaUtils {
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaUtils.class);
+
private static final String CONTEXT_CLASS_NAME =
"org.apache.kafka.connect.runtime.WorkerSinkTaskContext";
@@ -50,6 +57,33 @@ class KafkaUtils {
return kafkaConsumer(context).groupMetadata();
}
+ static void seekToLastCommittedOffsets(SinkTaskContext context) {
+ Consumer<byte[], byte[]> consumer = kafkaConsumer(context);
+ if (consumer == null) {
+ return;
+ }
+
+ Map<TopicPartition, OffsetAndMetadata> committedOffsets =
+ consumer.committed(consumer.assignment());
+ if (committedOffsets == null || committedOffsets.isEmpty()) {
+ return;
+ }
+
+ committedOffsets.forEach(
+ (topicPartition, offsetAndMetadata) -> {
+ if (offsetAndMetadata != null) {
+ try {
+ consumer.seek(topicPartition, offsetAndMetadata.offset());
+ } catch (IllegalStateException e) {
+ LOG.warn(
+ "Rebalance may have occurred, partition {} lost before seeking",
+ topicPartition,
+ e);
+ }
+ }
+ });
+ }
+
@SuppressWarnings("unchecked")
private static Consumer<byte[], byte[]> kafkaConsumer(SinkTaskContext context) {
String contextClassName = context.getClass().getName();
diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/Committer.java b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/MockIcebergSinkTask.java
similarity index 51%
copy from kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/Committer.java
copy to kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/MockIcebergSinkTask.java
index edc217d1b0..0e016e478c 100644
--- a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/Committer.java
+++ b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/MockIcebergSinkTask.java
@@ -19,14 +19,43 @@
package org.apache.iceberg.connect;
import java.util.Collection;
-import org.apache.iceberg.catalog.Catalog;
+import java.util.Map;
+import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.connect.sink.SinkRecord;
-import org.apache.kafka.connect.sink.SinkTaskContext;
+import org.apache.kafka.connect.sink.SinkTask;
-public interface Committer {
- void start(Catalog catalog, IcebergSinkConfig config, SinkTaskContext context);
+public class MockIcebergSinkTask extends SinkTask {
- void stop();
+ private boolean isCoordinator = false;
- void save(Collection<SinkRecord> sinkRecords);
+ @Override
+ public String version() {
+ return "";
+ }
+
+ @Override
+ public void start(Map<String, String> props) {}
+
+ @Override
+ public void open(Collection<TopicPartition> partitions) {
+ isCoordinator = partitions.stream().anyMatch(tp -> tp.partition() == 0);
+ }
+
+ @Override
+ public void close(Collection<TopicPartition> partitions) {
+ boolean wasCoordinator = partitions.stream().anyMatch(tp -> tp.partition() == 0);
+ if (wasCoordinator) {
+ isCoordinator = false;
+ }
+ }
+
+ @Override
+ public void put(Collection<SinkRecord> collection) {}
+
+ @Override
+ public void stop() {}
+
+ public boolean isCoordinatorRunning() {
+ return isCoordinator;
+ }
}
diff --git a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/ChannelTestBase.java b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/ChannelTestBase.java
index e6ffefbd97..117a5258d3 100644
--- a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/ChannelTestBase.java
+++ b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/ChannelTestBase.java
@@ -26,11 +26,14 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.connect.IcebergSinkConfig;
+import org.apache.iceberg.connect.MockIcebergSinkTask;
import org.apache.iceberg.connect.TableSinkConfig;
import org.apache.iceberg.inmemory.InMemoryCatalog;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
@@ -39,6 +42,7 @@ import org.apache.iceberg.types.Types;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.TopicDescription;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.MockConsumer;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.clients.producer.MockProducer;
@@ -60,6 +64,8 @@ public class ChannelTestBase {
protected KafkaClientFactory clientFactory;
protected MockProducer<String, byte[]> producer;
protected MockConsumer<String, byte[]> consumer;
+ protected MockConsumer<String, byte[]> sourceConsumer;
+ protected MockIcebergSinkTask mockIcebergSinkTask;
protected Admin admin;
private InMemoryCatalog initInMemoryCatalog() {
@@ -109,9 +115,10 @@ public class ChannelTestBase {
producer = new MockProducer<>(false, new StringSerializer(), new ByteArraySerializer());
producer.initTransactions();
-
+ mockIcebergSinkTask = new MockIcebergSinkTask();
consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
-
+ sourceConsumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
+ sourceConsumer.subscribe(Collections.singleton(SRC_TOPIC_NAME), new Listener());
clientFactory = mock(KafkaClientFactory.class);
when(clientFactory.createProducer(any())).thenReturn(producer);
when(clientFactory.createConsumer(any())).thenReturn(consumer);
@@ -128,4 +135,20 @@ public class ChannelTestBase {
consumer.rebalance(ImmutableList.of(tp));
consumer.updateBeginningOffsets(ImmutableMap.of(tp, 0L));
}
+
+ private class Listener implements ConsumerRebalanceListener {
+ @Override
+ public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
+ if (!partitions.isEmpty()) {
+ mockIcebergSinkTask.close(partitions);
+ }
+ }
+
+ @Override
+ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
+ if (!partitions.isEmpty()) {
+ mockIcebergSinkTask.open(partitions);
+ }
+ }
+ }
}
diff --git a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/CommitterImplTest.java b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/CommitterImplTest.java
index 7c8ccf8ef6..add7d0c044 100644
--- a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/CommitterImplTest.java
+++ b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/CommitterImplTest.java
@@ -51,10 +51,10 @@ public class CommitterImplTest {
List<TopicPartition> assignments =
ImmutableList.of(new TopicPartition("topic2", 1), new TopicPartition("topic1", 0));
- assertThat(committer.isLeader(members, assignments)).isTrue();
+ assertThat(committer.containsFirstPartition(members, assignments)).isTrue();
assignments =
ImmutableList.of(new TopicPartition("topic2", 0), new TopicPartition("topic1", 1));
- assertThat(committer.isLeader(members, assignments)).isFalse();
+ assertThat(committer.containsFirstPartition(members, assignments)).isFalse();
}
}
diff --git a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/CoordinatorTest.java b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/CoordinatorTest.java
index f3d0cf3dcf..d523d8f10b 100644
--- a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/CoordinatorTest.java
+++ b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/CoordinatorTest.java
@@ -43,8 +43,10 @@ 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.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.types.Types.StructType;
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;
@@ -206,4 +208,25 @@ public class CoordinatorTest extends ChannelTestBase {
return commitId;
}
+
+ @Test
+ public void testCoordinatorRunning() {
+ TopicPartition tp0 = new TopicPartition(SRC_TOPIC_NAME, 0);
+ TopicPartition tp1 = new TopicPartition(SRC_TOPIC_NAME, 1);
+ TopicPartition tp2 = new TopicPartition(SRC_TOPIC_NAME, 2);
+
+ // Assigning three topic partitions tp0, tp1, and tp2. This will be elected as leader as it has
+ // tp0.
+ sourceConsumer.rebalance(Lists.newArrayList(tp0, tp1, tp2));
+ assertThat(mockIcebergSinkTask.isCoordinatorRunning()).isTrue();
+
+ // Now revoking the partition 2, this should not close the coordinator as this task still has
+ // the zeroth partition
+ sourceConsumer.rebalance(Lists.newArrayList(tp0, tp1));
+ assertThat(mockIcebergSinkTask.isCoordinatorRunning()).isTrue();
+
+ // Now finally revoking partition zero and this should result in the closure of the coordinator
+ sourceConsumer.rebalance(ImmutableList.of(tp1));
+ assertThat(mockIcebergSinkTask.isCoordinatorRunning()).isFalse();
+ }
}