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 da78989d9377 fix(kafka-connect): terminate transaction coordinator
event loop on stop (#20271)
da78989d9377 is described below
commit da78989d9377c6603b1ef63bf5c82256266c2325
Author: fhan <[email protected]>
AuthorDate: Tue Oct 13 17:19:05 2026 +0800
fix(kafka-connect): terminate transaction coordinator event loop on stop
(#20271)
* fix(kafka-connect): terminate transaction coordinator event loop on stop
* fix(kafka-connect): clean up coordinator resources on shutdown
* fix(kafka-connect): retry coordinator service cleanup on repeated stop
---------
Co-authored-by: fhan <[email protected]>
---
.../transaction/ConnectTransactionCoordinator.java | 121 +++++++---
.../writers/ConnectTransactionServices.java | 5 +-
.../writers/KafkaConnectTransactionServices.java | 5 +
...TestConnectTransactionCoordinatorLifecycle.java | 251 +++++++++++++++++++++
.../helper/MockConnectTransactionServices.java | 4 +
5 files changed, 358 insertions(+), 28 deletions(-)
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 b3bd854a1e7e..7ab4bb3a040c 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
@@ -20,6 +20,7 @@ package org.apache.hudi.connect.transaction;
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.common.util.VisibleForTesting;
import org.apache.hudi.connect.ControlMessage;
import org.apache.hudi.connect.kafka.KafkaControlAgent;
import org.apache.hudi.connect.utils.KafkaConnectUtils;
@@ -42,6 +43,7 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -63,6 +65,7 @@ public class ConnectTransactionCoordinator implements
TransactionCoordinator, Ru
private static final Long START_COMMIT_INIT_DELAY_MS = 100L;
private static final Long RESTART_COMMIT_DELAY_MS = 500L;
private static final int COORDINATOR_EVENT_LOOP_TIMEOUT_MS = 1000;
+ private static final int COORDINATOR_EVENT_LOOP_SHUTDOWN_TIMEOUT_MS = 5000;
private final KafkaConnectConfigs configs;
@Getter
@@ -70,9 +73,11 @@ public class ConnectTransactionCoordinator implements
TransactionCoordinator, Ru
private final KafkaControlAgent kafkaControlClient;
private final ConnectTransactionServices transactionServices;
private final KafkaPartitionProvider partitionProvider;
+ private final int eventLoopShutdownTimeoutMs;
private final Map<Integer, List<WriteStatus>> partitionsWriteStatusReceived;
private final Map<Integer, Long> currentConsumedKafkaOffsets;
private final AtomicBoolean hasStarted = new AtomicBoolean(false);
+ private final AtomicBoolean transactionServicesClosed = new
AtomicBoolean(false);
private final BlockingQueue<CoordinatorEvent> events;
private final ExecutorService executorService;
private final ScheduledExecutorService scheduler;
@@ -81,6 +86,8 @@ public class ConnectTransactionCoordinator implements
TransactionCoordinator, Ru
private Map<Integer, Long> globalCommittedKafkaOffsets;
private State currentState;
private int numPartitions;
+ private boolean deregistered;
+ private boolean shutdownInitiated;
public ConnectTransactionCoordinator(KafkaConnectConfigs configs,
TopicPartition partition,
@@ -97,11 +104,23 @@ public class ConnectTransactionCoordinator implements
TransactionCoordinator, Ru
KafkaControlAgent kafkaControlClient,
ConnectTransactionServices
transactionServices,
KafkaPartitionProvider
partitionProvider) {
+ this(configs, partition, kafkaControlClient, transactionServices,
partitionProvider,
+ COORDINATOR_EVENT_LOOP_SHUTDOWN_TIMEOUT_MS);
+ }
+
+ @VisibleForTesting
+ ConnectTransactionCoordinator(KafkaConnectConfigs configs,
+ TopicPartition partition,
+ KafkaControlAgent kafkaControlClient,
+ ConnectTransactionServices transactionServices,
+ KafkaPartitionProvider partitionProvider,
+ int eventLoopShutdownTimeoutMs) {
this.configs = configs;
this.partition = partition;
this.kafkaControlClient = kafkaControlClient;
this.transactionServices = transactionServices;
this.partitionProvider = partitionProvider;
+ this.eventLoopShutdownTimeoutMs = eventLoopShutdownTimeoutMs;
this.events = new LinkedBlockingQueue<>();
scheduler = Executors.newSingleThreadScheduledExecutor();
executorService = Executors.newSingleThreadExecutor();
@@ -131,25 +150,20 @@ public class ConnectTransactionCoordinator implements
TransactionCoordinator, Ru
}
@Override
- public void stop() {
- kafkaControlClient.deregisterTransactionCoordinator(this);
- scheduler.shutdownNow();
+ public synchronized void stop() {
hasStarted.set(false);
- if (executorService != null) {
- boolean terminated = false;
- try {
- log.info("Shutting down executor service.");
- executorService.shutdown();
- log.info("Awaiting termination.");
- terminated = executorService.awaitTermination(100,
TimeUnit.MILLISECONDS);
- } catch (InterruptedException e) {
- // ignored
+ try {
+ if (!deregistered) {
+ kafkaControlClient.deregisterTransactionCoordinator(this);
+ deregistered = true;
}
-
- if (!terminated) {
- log.warn(
- "Unclean Kafka Control Manager executor service shutdown ");
- executorService.shutdownNow();
+ } finally {
+ if (!shutdownInitiated) {
+ shutdownInitiated = true;
+ shutdownEventLoop();
+ }
+ if (executorService.isTerminated()) {
+ closeTransactionServices();
}
}
}
@@ -173,14 +187,54 @@ public class ConnectTransactionCoordinator implements
TransactionCoordinator, Ru
@Override
public void run() {
- while (true) {
- try {
- CoordinatorEvent event =
events.poll(COORDINATOR_EVENT_LOOP_TIMEOUT_MS, TimeUnit.MILLISECONDS);
- if (event != null) {
- processCoordinatorEvent(event);
+ try {
+ while (hasStarted.get()) {
+ try {
+ CoordinatorEvent event =
events.poll(COORDINATOR_EVENT_LOOP_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ if (event != null && hasStarted.get()) {
+ processCoordinatorEvent(event);
+ }
+ } catch (InterruptedException exception) {
+ Thread.currentThread().interrupt();
+ log.debug("Transaction coordinator event loop interrupted for topic
{} partition {}",
+ partition.topic(), partition.partition());
+ break;
}
- } catch (InterruptedException exception) {
- log.warn("Error received while polling the event loop in Partition
Coordinator", exception);
+ }
+ } finally {
+ closeTransactionServices();
+ }
+ }
+
+ @VisibleForTesting
+ boolean isEventLoopTerminated() {
+ return executorService.isTerminated();
+ }
+
+ private void shutdownEventLoop() {
+ scheduler.shutdownNow();
+ events.clear();
+ executorService.shutdownNow();
+ try {
+ if (!executorService.awaitTermination(eventLoopShutdownTimeoutMs,
TimeUnit.MILLISECONDS)) {
+ log.warn("Transaction coordinator event loop did not terminate within
{} ms for topic {} partition {}",
+ eventLoopShutdownTimeoutMs, partition.topic(),
partition.partition());
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ log.warn("Interrupted while waiting for transaction coordinator event
loop to terminate for topic {} partition {}",
+ partition.topic(), partition.partition());
+ }
+ }
+
+ private void closeTransactionServices() {
+ if (transactionServicesClosed.compareAndSet(false, true)) {
+ try {
+ transactionServices.close();
+ } catch (RuntimeException e) {
+ transactionServicesClosed.set(false);
+ log.warn("Failed to close transaction services for topic {} partition
{}",
+ partition.topic(), partition.partition(), e);
}
}
}
@@ -190,9 +244,22 @@ public class ConnectTransactionCoordinator implements
TransactionCoordinator, Ru
}
private void submitEvent(CoordinatorEvent event, long delay, TimeUnit unit) {
- scheduler.schedule(() -> {
- events.add(event);
- }, delay, unit);
+ if (!hasStarted.get()) {
+ return;
+ }
+ try {
+ scheduler.schedule(() -> {
+ if (hasStarted.get()) {
+ events.add(event);
+ }
+ }, delay, unit);
+ } catch (RejectedExecutionException e) {
+ if (hasStarted.get()) {
+ throw e;
+ }
+ log.debug("Ignoring coordinator event submitted during shutdown for
topic {} partition {}",
+ partition.topic(), partition.partition());
+ }
}
private void processCoordinatorEvent(CoordinatorEvent event) {
diff --git
a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/ConnectTransactionServices.java
b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/ConnectTransactionServices.java
index 2ce44ff802bc..62f00d5d80d2 100644
---
a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/ConnectTransactionServices.java
+++
b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/ConnectTransactionServices.java
@@ -28,11 +28,14 @@ import java.util.Map;
* Transaction service APIs used by
* {@link TransactionCoordinator}.
*/
-public interface ConnectTransactionServices {
+public interface ConnectTransactionServices extends AutoCloseable {
String startCommit();
boolean endCommit(String commitTime, List<WriteStatus> writeStatuses,
Map<String, String> extraMetadata);
Map<String, String> fetchLatestExtraCommitMetadata();
+
+ @Override
+ void close();
}
diff --git
a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectTransactionServices.java
b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectTransactionServices.java
index 21c76f5aa8c6..6b89627fad9c 100644
---
a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectTransactionServices.java
+++
b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectTransactionServices.java
@@ -151,6 +151,11 @@ public class KafkaConnectTransactionServices implements
ConnectTransactionServic
throw new HoodieException("Fatal error retrieving Hoodie Extra Metadata
since Table Meta Client is absent");
}
+ @Override
+ public void close() {
+ javaClient.close();
+ }
+
private boolean isAsyncCompactionEnabled() {
return tableMetaClient.isPresent()
&&
HoodieTableType.MERGE_ON_READ.equals(tableMetaClient.get().getTableType())
diff --git
a/hudi-kafka-connect/src/test/java/org/apache/hudi/connect/transaction/TestConnectTransactionCoordinatorLifecycle.java
b/hudi-kafka-connect/src/test/java/org/apache/hudi/connect/transaction/TestConnectTransactionCoordinatorLifecycle.java
new file mode 100644
index 000000000000..3967c583defa
--- /dev/null
+++
b/hudi-kafka-connect/src/test/java/org/apache/hudi/connect/transaction/TestConnectTransactionCoordinatorLifecycle.java
@@ -0,0 +1,251 @@
+/*
+ * 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.hudi.connect.transaction;
+
+import org.apache.hudi.connect.ControlMessage;
+import org.apache.hudi.connect.kafka.KafkaControlAgent;
+import org.apache.hudi.connect.writers.ConnectTransactionServices;
+import org.apache.hudi.connect.writers.KafkaConnectConfigs;
+
+import org.apache.kafka.common.TopicPartition;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class TestConnectTransactionCoordinatorLifecycle {
+
+ private static final String TOPIC_NAME = "coordinator-lifecycle-test-topic";
+
+ @Test
+ void stopTerminatesEventLoopAndIgnoresFurtherEvents() throws Exception {
+ KafkaControlAgent controlAgent = mock(KafkaControlAgent.class);
+ ConnectTransactionServices transactionServices =
mock(ConnectTransactionServices.class);
+ CountDownLatch eventProcessed = new CountDownLatch(1);
+
when(transactionServices.fetchLatestExtraCommitMetadata()).thenReturn(Collections.emptyMap());
+ when(transactionServices.startCommit()).thenAnswer(invocation -> {
+ eventProcessed.countDown();
+ return "001";
+ });
+
+ KafkaConnectConfigs configs = KafkaConnectConfigs.newBuilder()
+ .withCommitIntervalSecs(60L)
+ .withCoordinatorWriteTimeoutSecs(60L)
+ .build();
+ ConnectTransactionCoordinator coordinator = new
ConnectTransactionCoordinator(
+ configs,
+ new TopicPartition(TOPIC_NAME,
ConnectTransactionCoordinator.COORDINATOR_KAFKA_PARTITION),
+ controlAgent,
+ transactionServices,
+ (bootstrapServers, topicName) -> 1);
+
+ try {
+ coordinator.start();
+ assertTrue(eventProcessed.await(5, TimeUnit.SECONDS));
+
+ coordinator.stop();
+
+ assertTrue(coordinator.isEventLoopTerminated());
+ assertDoesNotThrow(coordinator::stop);
+ assertDoesNotThrow(() ->
coordinator.processControlEvent(writeStatusEvent()));
+ verify(controlAgent,
times(1)).deregisterTransactionCoordinator(coordinator);
+ verify(transactionServices, times(1)).close();
+ } finally {
+ coordinator.stop();
+ }
+ }
+
+ @Test
+ void stopClosesTransactionServicesAfterEventProcessingExits() throws
Exception {
+ KafkaControlAgent controlAgent = mock(KafkaControlAgent.class);
+ ConnectTransactionServices transactionServices =
mock(ConnectTransactionServices.class);
+ CountDownLatch eventProcessingStarted = new CountDownLatch(1);
+ CountDownLatch releaseEventProcessing = new CountDownLatch(1);
+ AtomicBoolean eventProcessingExited = new AtomicBoolean(false);
+ AtomicBoolean servicesClosedAfterEventProcessing = new
AtomicBoolean(false);
+
when(transactionServices.fetchLatestExtraCommitMetadata()).thenReturn(Collections.emptyMap());
+ when(transactionServices.startCommit()).thenAnswer(invocation -> {
+ eventProcessingStarted.countDown();
+ try {
+ releaseEventProcessing.await();
+ } finally {
+ eventProcessingExited.set(true);
+ }
+ return "001";
+ });
+ doAnswer(invocation -> {
+ servicesClosedAfterEventProcessing.set(eventProcessingExited.get());
+ return null;
+ }).when(transactionServices).close();
+
+ ConnectTransactionCoordinator coordinator =
createCoordinator(controlAgent, transactionServices);
+ try {
+ coordinator.start();
+ assertTrue(eventProcessingStarted.await(5, TimeUnit.SECONDS));
+
+ coordinator.stop();
+
+ assertTrue(eventProcessingExited.get());
+ assertTrue(servicesClosedAfterEventProcessing.get());
+ verify(transactionServices, times(1)).close();
+ } finally {
+ releaseEventProcessing.countDown();
+ coordinator.stop();
+ }
+ }
+
+ @Test
+ void stopRetriesTransactionServicesCleanupAfterDelayedEventLoopExit() throws
Exception {
+ KafkaControlAgent controlAgent = mock(KafkaControlAgent.class);
+ ConnectTransactionServices transactionServices =
mock(ConnectTransactionServices.class);
+ CountDownLatch eventProcessingStarted = new CountDownLatch(1);
+ CountDownLatch releaseEventProcessing = new CountDownLatch(1);
+ CountDownLatch firstCloseAttempted = new CountDownLatch(1);
+ AtomicBoolean eventProcessingInterrupted = new AtomicBoolean(false);
+ AtomicBoolean failFirstClose = new AtomicBoolean(true);
+
when(transactionServices.fetchLatestExtraCommitMetadata()).thenReturn(Collections.emptyMap());
+ when(transactionServices.startCommit()).thenAnswer(invocation -> {
+ eventProcessingStarted.countDown();
+ boolean released = false;
+ while (!released) {
+ try {
+ releaseEventProcessing.await();
+ released = true;
+ } catch (InterruptedException e) {
+ eventProcessingInterrupted.set(true);
+ }
+ }
+ return "001";
+ });
+ doAnswer(invocation -> {
+ if (failFirstClose.compareAndSet(true, false)) {
+ firstCloseAttempted.countDown();
+ throw new IllegalStateException("close failed");
+ }
+ return null;
+ }).when(transactionServices).close();
+
+ ConnectTransactionCoordinator coordinator =
createCoordinator(controlAgent, transactionServices, 100);
+ try {
+ coordinator.start();
+ assertTrue(eventProcessingStarted.await(5, TimeUnit.SECONDS));
+
+ coordinator.stop();
+
+ assertTrue(eventProcessingInterrupted.get());
+ assertFalse(coordinator.isEventLoopTerminated());
+ verify(transactionServices, never()).close();
+
+ releaseEventProcessing.countDown();
+ assertTrue(firstCloseAttempted.await(5, TimeUnit.SECONDS));
+ waitForEventLoopTermination(coordinator);
+ verify(transactionServices, times(1)).close();
+
+ assertDoesNotThrow(coordinator::stop);
+ verify(transactionServices, times(2)).close();
+ } finally {
+ releaseEventProcessing.countDown();
+ coordinator.stop();
+ }
+ }
+
+ @Test
+ void stopRetriesDeregistrationAfterFailure() throws Exception {
+ KafkaControlAgent controlAgent = mock(KafkaControlAgent.class);
+ ConnectTransactionServices transactionServices =
mock(ConnectTransactionServices.class);
+ CountDownLatch eventProcessed = new CountDownLatch(1);
+
when(transactionServices.fetchLatestExtraCommitMetadata()).thenReturn(Collections.emptyMap());
+ when(transactionServices.startCommit()).thenAnswer(invocation -> {
+ eventProcessed.countDown();
+ return "001";
+ });
+
+ ConnectTransactionCoordinator coordinator =
createCoordinator(controlAgent, transactionServices);
+ try {
+ coordinator.start();
+ assertTrue(eventProcessed.await(5, TimeUnit.SECONDS));
+ doThrow(new IllegalStateException("deregistration failed"))
+ .doNothing()
+ .when(controlAgent).deregisterTransactionCoordinator(coordinator);
+
+ assertThrows(IllegalStateException.class, coordinator::stop);
+ assertTrue(coordinator.isEventLoopTerminated());
+ assertDoesNotThrow(coordinator::stop);
+ assertDoesNotThrow(coordinator::stop);
+
+ verify(controlAgent,
times(2)).deregisterTransactionCoordinator(coordinator);
+ verify(transactionServices, times(1)).close();
+ } finally {
+ coordinator.stop();
+ }
+ }
+
+ private static ConnectTransactionCoordinator createCoordinator(
+ KafkaControlAgent controlAgent, ConnectTransactionServices
transactionServices) {
+ return createCoordinator(controlAgent, transactionServices, 5000);
+ }
+
+ private static ConnectTransactionCoordinator createCoordinator(
+ KafkaControlAgent controlAgent,
+ ConnectTransactionServices transactionServices,
+ int eventLoopShutdownTimeoutMs) {
+ KafkaConnectConfigs configs = KafkaConnectConfigs.newBuilder()
+ .withCommitIntervalSecs(60L)
+ .withCoordinatorWriteTimeoutSecs(60L)
+ .build();
+ return new ConnectTransactionCoordinator(
+ configs,
+ new TopicPartition(TOPIC_NAME,
ConnectTransactionCoordinator.COORDINATOR_KAFKA_PARTITION),
+ controlAgent,
+ transactionServices,
+ (bootstrapServers, topicName) -> 1,
+ eventLoopShutdownTimeoutMs);
+ }
+
+ private static void
waitForEventLoopTermination(ConnectTransactionCoordinator coordinator)
+ throws InterruptedException {
+ long deadlineNanos = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+ while (!coordinator.isEventLoopTerminated() && System.nanoTime() <
deadlineNanos) {
+ TimeUnit.MILLISECONDS.sleep(10);
+ }
+ assertTrue(coordinator.isEventLoopTerminated());
+ }
+
+ private static ControlMessage writeStatusEvent() {
+ return ControlMessage.newBuilder()
+ .setType(ControlMessage.EventType.WRITE_STATUS)
+ .setTopicName(TOPIC_NAME)
+ .setCommitTime("001")
+ .build();
+ }
+}
diff --git
a/hudi-kafka-connect/src/test/java/org/apache/hudi/helper/MockConnectTransactionServices.java
b/hudi-kafka-connect/src/test/java/org/apache/hudi/helper/MockConnectTransactionServices.java
index b3314ade3d5b..5f23ed8b6531 100644
---
a/hudi-kafka-connect/src/test/java/org/apache/hudi/helper/MockConnectTransactionServices.java
+++
b/hudi-kafka-connect/src/test/java/org/apache/hudi/helper/MockConnectTransactionServices.java
@@ -55,4 +55,8 @@ public class MockConnectTransactionServices implements
ConnectTransactionService
public Map<String, String> fetchLatestExtraCommitMetadata() {
return new HashMap<>();
}
+
+ @Override
+ public void close() {
+ }
}
