(hudi) branch master updated: fix(kafka-connect): retry start commit when the topic partition lookup fails (#20312)

[danny0405]

This is an automated email from the ASF dual-hosted git repository.

danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git


The following commit(s) were added to refs/heads/master by this push:
     new 9086d9819a5d fix(kafka-connect): retry start commit when the topic 
partition lookup fails (#20312)
9086d9819a5d is described below

commit 9086d9819a5d15ac4b0cf40d6a7bd39e58e0b894
Author: Vova Kolmakov <wombatu...@gmail.com>
AuthorDate: Thu Oct 15 10:04:45 2026 +0700

    fix(kafka-connect): retry start commit when the topic partition lookup 
fails (#20312)
    
    Co-authored-by: Vova Kolmakov <wombatu...@apache.org>
---
 .../transaction/ConnectTransactionCoordinator.java |  2 +-
 .../connect/TestConnectTransactionCoordinator.java | 42 ++++++++++++++++++++++
 2 files changed, 43 insertions(+), 1 deletion(-)

diff --git 
a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/transaction/ConnectTransactionCoordinator.java
 
b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/transaction/ConnectTransactionCoordinator.java
index 7ab4bb3a040c..55bc71a98ad3 100644
--- 
a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/transaction/ConnectTransactionCoordinator.java
+++ 
b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/transaction/ConnectTransactionCoordinator.java
@@ -302,9 +302,9 @@ public class ConnectTransactionCoordinator implements 
TransactionCoordinator, Ru
   }
 
   private void startNewCommit() {
-    numPartitions = 
partitionProvider.getLatestNumPartitions(configs.getString(BOOTSTRAP_SERVERS_CFG),
 partition.topic());
     partitionsWriteStatusReceived.clear();
     try {
+      numPartitions = 
partitionProvider.getLatestNumPartitions(configs.getString(BOOTSTRAP_SERVERS_CFG),
 partition.topic());
       currentCommitTime = transactionServices.startCommit();
       
kafkaControlClient.publishMessage(buildControlMessage(ControlMessage.EventType.START_COMMIT));
       currentState = State.STARTED_COMMIT;
diff --git 
a/hudi-kafka-connect/src/test/java/org/apache/hudi/connect/TestConnectTransactionCoordinator.java
 
b/hudi-kafka-connect/src/test/java/org/apache/hudi/connect/TestConnectTransactionCoordinator.java
index 5ce1710bb785..8d5128a3d2f3 100644
--- 
a/hudi-kafka-connect/src/test/java/org/apache/hudi/connect/TestConnectTransactionCoordinator.java
+++ 
b/hudi-kafka-connect/src/test/java/org/apache/hudi/connect/TestConnectTransactionCoordinator.java
@@ -21,6 +21,7 @@ package org.apache.hudi.connect;
 import org.apache.hudi.client.WriteStatus;
 import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.util.Option;
+import org.apache.hudi.connect.kafka.KafkaControlAgent;
 import org.apache.hudi.connect.transaction.ConnectTransactionCoordinator;
 import org.apache.hudi.connect.transaction.TransactionCoordinator;
 import org.apache.hudi.connect.transaction.TransactionParticipant;
@@ -34,6 +35,7 @@ import lombok.Getter;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
 
@@ -44,9 +46,13 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.argThat;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
 
 public class TestConnectTransactionCoordinator {
 
@@ -101,6 +107,42 @@ public class TestConnectTransactionCoordinator {
     participant.stop();
   }
 
+  @Test
+  public void testRetryStartCommitWhenPartitionLookupFails() {
+    KafkaControlAgent controlAgent = mock(KafkaControlAgent.class);
+    AtomicInteger partitionLookups = new AtomicInteger();
+    // The first lookup fails, e.g. because the Kafka brokers are briefly 
unreachable
+    ConnectTransactionCoordinator.KafkaPartitionProvider partitionProvider = 
(bootstrapServers, topicName) -> {
+      if (partitionLookups.getAndIncrement() == 0) {
+        throw new HoodieException("Failed to fetch the latest number of 
partitions");
+      }
+      return TOTAL_KAFKA_PARTITIONS;
+    };
+    configs = KafkaConnectConfigs.newBuilder()
+        .withCommitIntervalSecs(60L)
+        .withCoordinatorWriteTimeoutSecs(60L)
+        .build();
+
+    TransactionCoordinator coordinator = new ConnectTransactionCoordinator(
+        configs,
+        new TopicPartition(TOPIC_NAME, 0),
+        controlAgent,
+        transactionServices,
+        partitionProvider);
+    try {
+      coordinator.start();
+
+      // The coordinator retries instead of stalling, and the failed attempt 
started no instant,
+      // so the commit that starts is the first one handed out by 
MockConnectTransactionServices
+      verify(controlAgent, timeout(TimeUnit.SECONDS.toMillis(10)))
+          .publishMessage(argThat(message -> message.getType() == 
ControlMessage.EventType.START_COMMIT
+              && message.getCommitTime().equals("101")));
+      assertEquals(2, partitionLookups.get());
+    } finally {
+      coordinator.stop();
+    }
+  }
+
   /**
    * A mock Transaction Participant, that exercises all the test scenarios
    * for the coordinator as mentioned in {@link TestScenarios}.