This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new e9dc8c90dd [Fix][Connector-V2] Prevent Kafka transaction cleanup loop
(#11588)
e9dc8c90dd is described below
commit e9dc8c90dd082170f80457cec098ca2af7811579
Author: Jast <[email protected]>
AuthorDate: Thu Jul 30 23:28:22 2026 +0800
[Fix][Connector-V2] Prevent Kafka transaction cleanup loop (#11588)
Co-authored-by: zhangshenghang <[email protected]>
---
.../kafka/sink/KafkaTransactionSender.java | 47 ++++++------
.../kafka/sink/KafkaTransactionSenderTest.java | 85 ++++++++++++++++++++++
2 files changed, 110 insertions(+), 22 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaTransactionSender.java
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaTransactionSender.java
index f211db4119..cc6f71f4d6 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaTransactionSender.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaTransactionSender.java
@@ -64,7 +64,7 @@ public class KafkaTransactionSender<K, V> implements
KafkaProduceSender<K, V> {
@Override
public void beginTransaction(String transactionId) {
this.transactionId = transactionId;
- this.kafkaProducer = getTransactionProducer(kafkaProperties,
transactionId);
+ this.kafkaProducer = getTransactionProducer(transactionId);
kafkaProducer.beginTransaction();
recordNumInTransaction = 0;
}
@@ -88,26 +88,21 @@ public class KafkaTransactionSender<K, V> implements
KafkaProduceSender<K, V> {
@Override
public void abortTransaction(long checkpointId) {
-
- KafkaInternalProducer<K, V> producer;
- if (this.kafkaProducer != null) {
- producer = this.kafkaProducer;
- } else {
- producer =
- getTransactionProducer(
- this.kafkaProperties,
- generateTransactionId(this.transactionPrefix,
checkpointId));
- }
-
for (long i = checkpointId; ; i++) {
String transactionId =
generateTransactionId(this.transactionPrefix, i);
- producer.initTransactionId(transactionId);
- if (log.isDebugEnabled()) {
- log.debug("Abort kafka transaction: {}", transactionId);
- }
- producer.flush();
- if (producer.getEpoch() == 0) {
- break;
+ KafkaInternalProducer<K, V> producer =
createTransactionProducer(transactionId);
+ try {
+ if (log.isDebugEnabled()) {
+ log.debug("Abort kafka transaction: {}", transactionId);
+ }
+ // Initializing a producer with the same transactional ID
fences and aborts any
+ // transaction left by a previous attempt. Each ID needs its
own producer: changing
+ // the ID on a reused producer can retain a non-zero epoch
indefinitely.
+ if (producer.getEpoch() == 0) {
+ return;
+ }
+ } finally {
+ closeTemporaryProducer(producer);
}
}
}
@@ -137,14 +132,22 @@ public class KafkaTransactionSender<K, V> implements
KafkaProduceSender<K, V> {
}
}
- private KafkaInternalProducer<K, V> getTransactionProducer(
- Properties properties, String transactionId) {
+ private KafkaInternalProducer<K, V> getTransactionProducer(String
transactionId) {
close();
- Properties transactionProperties = (Properties) properties.clone();
+ return createTransactionProducer(transactionId);
+ }
+
+ /** Creates and initializes a producer used for one transactional ID. */
+ protected KafkaInternalProducer<K, V> createTransactionProducer(String
transactionId) {
+ Properties transactionProperties = (Properties)
kafkaProperties.clone();
transactionProperties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,
transactionId);
KafkaInternalProducer<K, V> transactionProducer =
new KafkaInternalProducer<>(transactionProperties,
transactionId);
transactionProducer.initTransactions();
return transactionProducer;
}
+
+ private void closeTemporaryProducer(KafkaInternalProducer<K, V> producer) {
+ KafkaClientUtils.runWithConnectorClassLoader(() ->
producer.close(Duration.ZERO));
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaTransactionSenderTest.java
b/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaTransactionSenderTest.java
new file mode 100644
index 0000000000..8b34fc8237
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaTransactionSenderTest.java
@@ -0,0 +1,85 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.kafka.sink;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.time.Duration;
+import java.util.ArrayDeque;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Properties;
+
+class KafkaTransactionSenderTest {
+
+ @Test
+ void abortTransactionUsesFreshProducerForEachTransactionalId() {
+ KafkaInternalProducer<byte[], byte[]> existingTransaction =
+ Mockito.mock(KafkaInternalProducer.class);
+ KafkaInternalProducer<byte[], byte[]> unusedTransaction =
+ Mockito.mock(KafkaInternalProducer.class);
+ Mockito.when(existingTransaction.getEpoch()).thenReturn((short) 1);
+ Mockito.when(unusedTransaction.getEpoch()).thenReturn((short) 0);
+
+ TestingKafkaTransactionSender sender =
+ new TestingKafkaTransactionSender(existingTransaction,
unusedTransaction);
+
+ sender.abortTransaction(7L);
+
+ Assertions.assertEquals(
+ Arrays.asList("test-prefix-7", "test-prefix-8"),
sender.createdTransactionIds);
+ Mockito.verify(existingTransaction).close(Duration.ZERO);
+ Mockito.verify(unusedTransaction).close(Duration.ZERO);
+ }
+
+ private static class TestingKafkaTransactionSender
+ extends KafkaTransactionSender<byte[], byte[]> {
+
+ private final ArrayDeque<KafkaInternalProducer<byte[], byte[]>>
producers;
+ private final List<String> createdTransactionIds = new
java.util.ArrayList<>();
+
+ @SafeVarargs
+ private TestingKafkaTransactionSender(KafkaInternalProducer<byte[],
byte[]>... producers) {
+ super("test-prefix", kafkaProperties());
+ this.producers = new ArrayDeque<>(Arrays.asList(producers));
+ }
+
+ @Override
+ protected KafkaInternalProducer<byte[], byte[]>
createTransactionProducer(
+ String transactionId) {
+ createdTransactionIds.add(transactionId);
+ return producers.removeFirst();
+ }
+
+ private static Properties kafkaProperties() {
+ Properties properties = new Properties();
+ properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
"localhost:9092");
+ properties.put(
+ ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
+
"org.apache.kafka.common.serialization.ByteArraySerializer");
+ properties.put(
+ ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
+
"org.apache.kafka.common.serialization.ByteArraySerializer");
+ return properties;
+ }
+ }
+}
