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 be960cf71e [Fix][Connector-V2] Run Kafka cleanup with connector
classloader (#11178)
be960cf71e is described below
commit be960cf71e42cf853ef939c6924e7f8e6a5ba11d
Author: Daniel <[email protected]>
AuthorDate: Sun Jul 5 11:10:18 2026 +0800
[Fix][Connector-V2] Run Kafka cleanup with connector classloader (#11178)
---
.../seatunnel/kafka/KafkaClientUtils.java | 48 +++++++++++++++
.../kafka/sink/KafkaNoTransactionSender.java | 8 ++-
.../seatunnel/kafka/sink/KafkaSinkCommitter.java | 18 +++---
.../kafka/sink/KafkaTransactionSender.java | 13 ++--
.../kafka/source/KafkaPartitionSplitReader.java | 3 +-
.../kafka/source/KafkaSourceSplitEnumerator.java | 3 +-
.../seatunnel/kafka/KafkaClientUtilsTest.java | 69 ++++++++++++++++++++++
7 files changed, 146 insertions(+), 16 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/KafkaClientUtils.java
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/KafkaClientUtils.java
new file mode 100644
index 0000000000..658b6417ae
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/KafkaClientUtils.java
@@ -0,0 +1,48 @@
+/*
+ * 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;
+
+import org.apache.seatunnel.common.utils.TemporaryClassLoaderContext;
+
+/** Utility methods for executing Kafka client operations inside the connector
classloader. */
+public final class KafkaClientUtils {
+
+ private KafkaClientUtils() {}
+
+ /**
+ * Runs a Kafka client action with the connector classloader as the thread
context classloader.
+ *
+ * <p>Kafka clients can lazily resolve cleanup implementation classes
while closing producers,
+ * consumers, and admin clients. SeaTunnel may call connector cleanup from
engine threads whose
+ * context classloader no longer points at the Kafka connector, so cleanup
must restore the
+ * connector classloader instead of preloading Kafka private classes by
name.
+ */
+ public static void runWithConnectorClassLoader(KafkaClientAction action) {
+ try (TemporaryClassLoaderContext ignored =
+
TemporaryClassLoaderContext.of(KafkaClientUtils.class.getClassLoader())) {
+ action.run();
+ }
+ }
+
+ /** Action executed while the Kafka connector classloader is installed. */
+ @FunctionalInterface
+ public interface KafkaClientAction {
+
+ void run();
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaNoTransactionSender.java
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaNoTransactionSender.java
index 46811e5fee..aaa5bbb5f1 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaNoTransactionSender.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaNoTransactionSender.java
@@ -17,6 +17,7 @@
package org.apache.seatunnel.connectors.seatunnel.kafka.sink;
+import org.apache.seatunnel.connectors.seatunnel.kafka.KafkaClientUtils;
import org.apache.seatunnel.connectors.seatunnel.kafka.state.KafkaCommitInfo;
import org.apache.seatunnel.connectors.seatunnel.kafka.state.KafkaSinkState;
@@ -76,7 +77,10 @@ public class KafkaNoTransactionSender<K, V> implements
KafkaProduceSender<K, V>
@Override
public void close() {
- kafkaProducer.flush();
- kafkaProducer.close();
+ KafkaClientUtils.runWithConnectorClassLoader(
+ () -> {
+ kafkaProducer.flush();
+ kafkaProducer.close();
+ });
}
}
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkCommitter.java
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkCommitter.java
index 4be9fba709..da41ce06a5 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkCommitter.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkCommitter.java
@@ -19,6 +19,7 @@ package org.apache.seatunnel.connectors.seatunnel.kafka.sink;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.sink.SinkCommitter;
+import org.apache.seatunnel.connectors.seatunnel.kafka.KafkaClientUtils;
import org.apache.seatunnel.connectors.seatunnel.kafka.state.KafkaCommitInfo;
import org.apache.kafka.clients.producer.KafkaProducer;
@@ -54,10 +55,7 @@ public class KafkaSinkCommitter implements
SinkCommitter<KafkaCommitInfo> {
producer.commitTransaction();
producer.flush();
}
- if (this.kafkaProducer != null) {
- kafkaProducer.close();
- kafkaProducer = null;
- }
+ closeProducer();
return commitInfos;
}
@@ -70,10 +68,7 @@ public class KafkaSinkCommitter implements
SinkCommitter<KafkaCommitInfo> {
KafkaProducer<?, ?> producer = getProducer(commitInfo);
producer.abortTransaction();
}
- if (this.kafkaProducer != null) {
- kafkaProducer.close();
- kafkaProducer = null;
- }
+ closeProducer();
}
private KafkaInternalProducer<?, ?> getProducer(KafkaCommitInfo
commitInfo) {
@@ -91,4 +86,11 @@ public class KafkaSinkCommitter implements
SinkCommitter<KafkaCommitInfo> {
commitInfo.getProducerId(), commitInfo.getEpoch(),
commitInfo.isTxnStarted());
return kafkaProducer;
}
+
+ private void closeProducer() {
+ if (this.kafkaProducer != null) {
+ KafkaClientUtils.runWithConnectorClassLoader(kafkaProducer::close);
+ kafkaProducer = null;
+ }
+ }
}
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 0d5101260a..f211db4119 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
@@ -19,6 +19,7 @@ package org.apache.seatunnel.connectors.seatunnel.kafka.sink;
import org.apache.seatunnel.shade.com.google.common.collect.Lists;
+import org.apache.seatunnel.connectors.seatunnel.kafka.KafkaClientUtils;
import org.apache.seatunnel.connectors.seatunnel.kafka.state.KafkaCommitInfo;
import org.apache.seatunnel.connectors.seatunnel.kafka.state.KafkaSinkState;
@@ -125,10 +126,14 @@ public class KafkaTransactionSender<K, V> implements
KafkaProduceSender<K, V> {
@Override
public void close() {
if (kafkaProducer != null) {
- kafkaProducer.flush();
- // kafkaProducer will abort the transaction if you call close()
without a duration arg
- // which will cause an exception when Committer commit the
transaction later.
- kafkaProducer.close(Duration.ZERO);
+ KafkaClientUtils.runWithConnectorClassLoader(
+ () -> {
+ kafkaProducer.flush();
+ // kafkaProducer will abort the transaction if you
call close() without a
+ // duration arg which will cause an exception when
Committer commit the
+ // transaction later.
+ kafkaProducer.close(Duration.ZERO);
+ });
}
}
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaPartitionSplitReader.java
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaPartitionSplitReader.java
index 8f17be47e3..9609534090 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaPartitionSplitReader.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaPartitionSplitReader.java
@@ -25,6 +25,7 @@ import
org.apache.seatunnel.connectors.seatunnel.common.source.reader.RecordsWit
import
org.apache.seatunnel.connectors.seatunnel.common.source.reader.splitreader.SplitReader;
import
org.apache.seatunnel.connectors.seatunnel.common.source.reader.splitreader.SplitsAddition;
import
org.apache.seatunnel.connectors.seatunnel.common.source.reader.splitreader.SplitsChange;
+import org.apache.seatunnel.connectors.seatunnel.kafka.KafkaClientUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -297,7 +298,7 @@ public class KafkaPartitionSplitReader
@Override
public void close() throws Exception {
- consumer.close();
+ KafkaClientUtils.runWithConnectorClassLoader(consumer::close);
}
public void notifyCheckpointComplete(
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceSplitEnumerator.java
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceSplitEnumerator.java
index 07b7f9ccb4..c4260ce588 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceSplitEnumerator.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceSplitEnumerator.java
@@ -22,6 +22,7 @@ import
org.apache.seatunnel.shade.com.google.common.annotations.VisibleForTestin
import org.apache.seatunnel.api.source.SourceSplitEnumerator;
import org.apache.seatunnel.api.table.catalog.TablePath;
import org.apache.seatunnel.common.config.Common;
+import org.apache.seatunnel.connectors.seatunnel.kafka.KafkaClientUtils;
import
org.apache.seatunnel.connectors.seatunnel.kafka.exception.KafkaConnectorErrorCode;
import
org.apache.seatunnel.connectors.seatunnel.kafka.exception.KafkaConnectorException;
import org.apache.seatunnel.connectors.seatunnel.kafka.state.KafkaSourceState;
@@ -253,7 +254,7 @@ public class KafkaSourceSplitEnumerator
@Override
public void close() throws IOException {
if (this.adminClient != null) {
- adminClient.close();
+ KafkaClientUtils.runWithConnectorClassLoader(adminClient::close);
}
if (scheduledFuture != null) {
scheduledFuture.cancel(false);
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/KafkaClientUtilsTest.java
b/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/KafkaClientUtilsTest.java
new file mode 100644
index 0000000000..78da3301c7
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/KafkaClientUtilsTest.java
@@ -0,0 +1,69 @@
+/*
+ * 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;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+class KafkaClientUtilsTest {
+
+ @Test
+ void shouldRunActionWithConnectorClassLoaderAndRestoreOriginal() {
+ Thread thread = Thread.currentThread();
+ ClassLoader previousClassLoader = thread.getContextClassLoader();
+ ClassLoader originalClassLoader = new ClassLoader(previousClassLoader)
{};
+ AtomicReference<ClassLoader> actionClassLoader = new
AtomicReference<>();
+
+ try {
+ thread.setContextClassLoader(originalClassLoader);
+
+ KafkaClientUtils.runWithConnectorClassLoader(
+ () ->
actionClassLoader.set(thread.getContextClassLoader()));
+
+ Assertions.assertSame(KafkaClientUtils.class.getClassLoader(),
actionClassLoader.get());
+ Assertions.assertSame(originalClassLoader,
thread.getContextClassLoader());
+ } finally {
+ thread.setContextClassLoader(previousClassLoader);
+ }
+ }
+
+ @Test
+ void shouldRestoreOriginalClassLoaderWhenActionFails() {
+ Thread thread = Thread.currentThread();
+ ClassLoader previousClassLoader = thread.getContextClassLoader();
+ ClassLoader originalClassLoader = new ClassLoader(previousClassLoader)
{};
+
+ try {
+ thread.setContextClassLoader(originalClassLoader);
+
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () ->
+ KafkaClientUtils.runWithConnectorClassLoader(
+ () -> {
+ throw new
IllegalStateException("expected");
+ }));
+
+ Assertions.assertSame(originalClassLoader,
thread.getContextClassLoader());
+ } finally {
+ thread.setContextClassLoader(previousClassLoader);
+ }
+ }
+}
