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 4f41c48887 [Fix][Connector-V2] Run Pulsar cleanup with connector
classloader (#11170)
4f41c48887 is described below
commit 4f41c4888734932f3c907fca3eba327de6e8bea7
Author: Daniel <[email protected]>
AuthorDate: Sat Jul 4 17:03:04 2026 +0800
[Fix][Connector-V2] Run Pulsar cleanup with connector classloader (#11170)
---
.../seatunnel/pulsar/config/PulsarConfigUtil.java | 30 +++++
.../seatunnel/pulsar/sink/PulsarSinkCommitter.java | 7 +-
.../seatunnel/pulsar/sink/PulsarSinkWriter.java | 4 +-
.../source/enumerator/PulsarSplitEnumerator.java | 13 +-
.../pulsar/source/reader/PulsarSourceReader.java | 49 ++++++--
.../source/reader/PulsarSplitReaderThread.java | 19 ++-
.../pulsar/config/PulsarConfigUtilTest.java | 45 +++++++
.../pulsar/sink/PulsarSinkWriterTest.java | 54 ++++++++
.../reader/PulsarSourceReaderRestoreTest.java | 140 +++++++++++++++++++++
9 files changed, 341 insertions(+), 20 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/config/PulsarConfigUtil.java
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/config/PulsarConfigUtil.java
index 667bfab4bf..3cb6c0d78e 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/config/PulsarConfigUtil.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/config/PulsarConfigUtil.java
@@ -18,6 +18,7 @@
package org.apache.seatunnel.connectors.seatunnel.pulsar.config;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.common.utils.TemporaryClassLoaderContext;
import
org.apache.seatunnel.connectors.seatunnel.pulsar.exception.PulsarConnectorErrorCode;
import
org.apache.seatunnel.connectors.seatunnel.pulsar.exception.PulsarConnectorException;
@@ -80,6 +81,23 @@ public class PulsarConfigUtil {
}
}
+ /**
+ * Runs a Pulsar action with the connector classloader as the thread
context classloader.
+ *
+ * <p>Pulsar and its shaded Netty/Jersey stack can lazily resolve
implementation classes while
+ * closing clients, consumers, producers, and admins. SeaTunnel may call
connector close methods
+ * from engine threads whose context classloader no longer points to the
Pulsar connector, so
+ * the cleanup action must restore the connector classloader instead of
preloading
+ * version-specific Pulsar private classes.
+ */
+ public static void runWithConnectorClassLoader(ConnectorClassLoaderAction
action)
+ throws Exception {
+ try (TemporaryClassLoaderContext ignored =
+
TemporaryClassLoaderContext.of(PulsarConfigUtil.class.getClassLoader())) {
+ action.run();
+ }
+ }
+
public static ConsumerBuilder<byte[]> createConsumerBuilder(
PulsarClient client, PulsarConsumerConfig config) {
ConsumerBuilder<byte[]> builder = client.newConsumer(Schema.BYTES);
@@ -198,4 +216,16 @@ public class PulsarConfigUtil {
ProducerBase<byte[]> producerBase = (ProducerBase<byte[]>) producer;
return new TypedMessageBuilderImpl<byte[]>(producerBase, Schema.BYTES,
transaction);
}
+
+ /** Action that may throw while running under the Pulsar connector
classloader. */
+ @FunctionalInterface
+ public interface ConnectorClassLoaderAction {
+
+ /**
+ * Executes the action.
+ *
+ * @throws Exception when the wrapped Pulsar operation fails
+ */
+ void run() throws Exception;
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkCommitter.java
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkCommitter.java
index 99330c889d..3ea1bbca09 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkCommitter.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkCommitter.java
@@ -67,7 +67,12 @@ public class PulsarSinkCommitter implements
SinkCommitter<PulsarCommitInfo> {
client.abort(txnID);
}
if (this.pulsarClient != null) {
- pulsarClient.close();
+ try {
+
PulsarConfigUtil.runWithConnectorClassLoader(pulsarClient::close);
+ } catch (Exception e) {
+ throw new IOException(
+ "Failed to close Pulsar client after aborting
transactions.", e);
+ }
}
}
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java
index 22e3c69448..cee685f69e 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java
@@ -249,14 +249,14 @@ public class PulsarSinkWriter
Throwable closeFailure = null;
for (Producer<byte[]> producer : producerMap.values()) {
try {
- producer.close();
+ PulsarConfigUtil.runWithConnectorClassLoader(producer::close);
} catch (Throwable throwable) {
closeFailure = appendSuppressed(closeFailure, throwable);
}
}
if (pulsarClient != null) {
try {
- pulsarClient.close();
+
PulsarConfigUtil.runWithConnectorClassLoader(pulsarClient::close);
} catch (Throwable throwable) {
closeFailure = appendSuppressed(closeFailure, throwable);
}
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/enumerator/PulsarSplitEnumerator.java
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/enumerator/PulsarSplitEnumerator.java
index c0ae2059d2..ab4b31d2b8 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/enumerator/PulsarSplitEnumerator.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/enumerator/PulsarSplitEnumerator.java
@@ -267,11 +267,16 @@ public class PulsarSplitEnumerator
@Override
public void close() throws IOException {
- if (pulsarAdmin != null) {
- pulsarAdmin.close();
- }
if (executor != null) {
- executor.shutdown();
+ // Stop periodic discovery before closing the shared admin client
it uses.
+ executor.shutdownNow();
+ }
+ if (pulsarAdmin != null) {
+ try {
+
PulsarConfigUtil.runWithConnectorClassLoader(pulsarAdmin::close);
+ } catch (Exception e) {
+ throw new IOException("Failed to close Pulsar admin.", e);
+ }
}
}
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/reader/PulsarSourceReader.java
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/reader/PulsarSourceReader.java
index 92d6b994ee..a4bd22580b 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/reader/PulsarSourceReader.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/reader/PulsarSourceReader.java
@@ -117,19 +117,25 @@ public class PulsarSourceReader<T> implements
SourceReader<T, PulsarPartitionSpl
@Override
public void close() throws IOException {
- if (pulsarClient != null) {
- pulsarClient.close();
- }
+ Throwable closeFailure = null;
+ // Close split readers first so consumers stop before the shared
Pulsar client disappears.
for (PulsarSplitReaderThread pulsarSplitReaderThread :
splitReaders.values()) {
try {
- pulsarSplitReaderThread.close();
- } catch (IOException e) {
- throw new PulsarConnectorException(
- CommonErrorCodeDeprecated.READER_OPERATION_FAILED,
- "Failed to close the split reader thread.",
- e);
+
PulsarConfigUtil.runWithConnectorClassLoader(pulsarSplitReaderThread::close);
+ } catch (Throwable throwable) {
+ closeFailure = appendSuppressed(closeFailure, throwable);
+ }
+ }
+ if (pulsarClient != null) {
+ try {
+
PulsarConfigUtil.runWithConnectorClassLoader(pulsarClient::close);
+ } catch (Throwable throwable) {
+ closeFailure = appendSuppressed(closeFailure, throwable);
}
}
+ if (closeFailure != null) {
+ rethrowCloseFailure(closeFailure);
+ }
}
@Override
@@ -273,6 +279,31 @@ public class PulsarSourceReader<T> implements
SourceReader<T, PulsarPartitionSpl
return tablePath != null ? tablePath : defaultTablePath;
}
+ /**
+ * Preserves all cleanup failures while still allowing the remaining
Pulsar resources to close.
+ */
+ private Throwable appendSuppressed(Throwable existingFailure, Throwable
newFailure) {
+ if (existingFailure == null) {
+ return newFailure;
+ }
+ existingFailure.addSuppressed(newFailure);
+ return existingFailure;
+ }
+
+ /**
+ * Converts deferred cleanup failures to the reader close contract after
all resources are
+ * tried.
+ */
+ private void rethrowCloseFailure(Throwable throwable) throws IOException {
+ if (throwable instanceof IOException) {
+ throw (IOException) throwable;
+ }
+ throw new PulsarConnectorException(
+ CommonErrorCodeDeprecated.READER_OPERATION_FAILED,
+ "Failed to close the Pulsar source reader.",
+ throwable);
+ }
+
private TablePath resolveTablePath(PulsarPartitionSplit split) {
return split.getTablePath() != null ? split.getTablePath() :
defaultTablePath;
}
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/reader/PulsarSplitReaderThread.java
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/reader/PulsarSplitReaderThread.java
index 6073cf26be..a4afab3a74 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/reader/PulsarSplitReaderThread.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/reader/PulsarSplitReaderThread.java
@@ -106,7 +106,7 @@ public class PulsarSplitReaderThread extends Thread
implements Closeable {
} finally {
// make sure the PulsarConsumer is closed
try {
- consumer.close();
+ closeConsumer();
} catch (Throwable t) {
LOG.warn("Error while closing pulsar consumer", t);
} finally {
@@ -118,9 +118,7 @@ public class PulsarSplitReaderThread extends Thread
implements Closeable {
@Override
public void close() throws IOException {
running = false;
- if (consumer != null) {
- consumer.close();
- }
+ closeConsumer();
}
public void committingCursor(MessageId offsetsToCommit) throws
PulsarClientException {
@@ -147,4 +145,17 @@ public class PulsarSplitReaderThread extends Thread
implements Closeable {
e);
}
}
+
+ /**
+ * Closes the Pulsar consumer while exposing the connector classloader to
Pulsar cleanup code.
+ */
+ private void closeConsumer() throws IOException {
+ if (consumer != null) {
+ try {
+ PulsarConfigUtil.runWithConnectorClassLoader(consumer::close);
+ } catch (Exception e) {
+ throw new IOException("Failed to close Pulsar consumer.", e);
+ }
+ }
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/config/PulsarConfigUtilTest.java
b/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/config/PulsarConfigUtilTest.java
new file mode 100644
index 0000000000..b6c6d86807
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/config/PulsarConfigUtilTest.java
@@ -0,0 +1,45 @@
+/*
+ * 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.pulsar.config;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class PulsarConfigUtilTest {
+
+ @Test
+ void
shouldRunActionWithConnectorClassLoaderAndRestoreOriginalClassLoader() throws
Exception {
+ ClassLoader originalClassLoader =
Thread.currentThread().getContextClassLoader();
+ ClassLoader unrelatedClassLoader = new ClassLoader(null) {};
+
+ try {
+ Thread.currentThread().setContextClassLoader(unrelatedClassLoader);
+
+ PulsarConfigUtil.runWithConnectorClassLoader(
+ () ->
+ Assertions.assertSame(
+ PulsarConfigUtil.class.getClassLoader(),
+
Thread.currentThread().getContextClassLoader()));
+
+ Assertions.assertSame(
+ unrelatedClassLoader,
Thread.currentThread().getContextClassLoader());
+ } finally {
+ Thread.currentThread().setContextClassLoader(originalClassLoader);
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriterTest.java
b/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriterTest.java
index e26dd9b6d0..021836b926 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriterTest.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriterTest.java
@@ -21,6 +21,7 @@ import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import
org.apache.seatunnel.connectors.seatunnel.pulsar.config.PulsarConfigUtil;
import
org.apache.seatunnel.connectors.seatunnel.pulsar.exception.PulsarConnectorException;
import org.apache.pulsar.client.api.Producer;
@@ -177,6 +178,43 @@ public class PulsarSinkWriterTest {
assertTrue(clientClosed.get());
}
+ @Test
+ public void testCloseRunsPulsarCleanupWithConnectorClassLoader() throws
Exception {
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put("format", "json");
+ configMap.put("field_delimiter", ",");
+ configMap.put("transaction_timeout", 1000);
+ configMap.put("semantics", "NON");
+ configMap.put("message.routing.mode", "RoundRobinPartition");
+
+ AtomicReference<ClassLoader> producerCloseClassLoader = new
AtomicReference<>();
+ AtomicReference<ClassLoader> clientCloseClassLoader = new
AtomicReference<>();
+ PulsarSinkWriter writer =
+ new PulsarSinkWriter(
+ new SeaTunnelRowType(new String[] {}, new
SeaTunnelDataType[] {}),
+ ReadonlyConfig.fromMap(configMap),
+ java.util.Collections.emptyList(),
+ createPulsarClientProxy(new AtomicBoolean(false),
clientCloseClassLoader),
+ topic ->
+ createProducerProxy(
+ new AtomicBoolean(false),
producerCloseClassLoader));
+ writer.getOrCreateProducer("topic-a");
+
+ ClassLoader originalClassLoader =
Thread.currentThread().getContextClassLoader();
+ ClassLoader unrelatedClassLoader = new ClassLoader(null) {};
+ try {
+ Thread.currentThread().setContextClassLoader(unrelatedClassLoader);
+
+ writer.close();
+
+ assertSame(unrelatedClassLoader,
Thread.currentThread().getContextClassLoader());
+ } finally {
+ Thread.currentThread().setContextClassLoader(originalClassLoader);
+ }
+ assertSame(PulsarConfigUtil.class.getClassLoader(),
producerCloseClassLoader.get());
+ assertSame(PulsarConfigUtil.class.getClassLoader(),
clientCloseClassLoader.get());
+ }
+
@SuppressWarnings("unchecked")
private static Producer<byte[]> createProducerProxy() {
return createProducerProxy(new AtomicBoolean(false));
@@ -184,6 +222,12 @@ public class PulsarSinkWriterTest {
@SuppressWarnings("unchecked")
private static Producer<byte[]> createProducerProxy(AtomicBoolean closed) {
+ return createProducerProxy(closed, new AtomicReference<>());
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Producer<byte[]> createProducerProxy(
+ AtomicBoolean closed, AtomicReference<ClassLoader>
closeClassLoader) {
return (Producer<byte[]>)
Proxy.newProxyInstance(
Producer.class.getClassLoader(),
@@ -191,6 +235,8 @@ public class PulsarSinkWriterTest {
(proxy, method, args) -> {
if ("close".equals(method.getName())) {
closed.set(true);
+ closeClassLoader.set(
+
Thread.currentThread().getContextClassLoader());
return null;
}
Class<?> returnType = method.getReturnType();
@@ -215,6 +261,12 @@ public class PulsarSinkWriterTest {
@SuppressWarnings("unchecked")
private static PulsarClient createPulsarClientProxy(AtomicBoolean closed) {
+ return createPulsarClientProxy(closed, new AtomicReference<>());
+ }
+
+ @SuppressWarnings("unchecked")
+ private static PulsarClient createPulsarClientProxy(
+ AtomicBoolean closed, AtomicReference<ClassLoader>
closeClassLoader) {
return (PulsarClient)
Proxy.newProxyInstance(
PulsarClient.class.getClassLoader(),
@@ -222,6 +274,8 @@ public class PulsarSinkWriterTest {
(proxy, method, args) -> {
if ("close".equals(method.getName())) {
closed.set(true);
+ closeClassLoader.set(
+
Thread.currentThread().getContextClassLoader());
return null;
}
Class<?> returnType = method.getReturnType();
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/reader/PulsarSourceReaderRestoreTest.java
b/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/reader/PulsarSourceReaderRestoreTest.java
index 5487e431aa..c2310edbba 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/reader/PulsarSourceReaderRestoreTest.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/reader/PulsarSourceReaderRestoreTest.java
@@ -31,6 +31,7 @@ import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import
org.apache.seatunnel.connectors.seatunnel.pulsar.config.PulsarClientConfig;
+import
org.apache.seatunnel.connectors.seatunnel.pulsar.config.PulsarConfigUtil;
import
org.apache.seatunnel.connectors.seatunnel.pulsar.config.PulsarConsumerConfig;
import
org.apache.seatunnel.connectors.seatunnel.pulsar.exception.PulsarConnectorErrorCode;
import
org.apache.seatunnel.connectors.seatunnel.pulsar.exception.PulsarConnectorException;
@@ -43,6 +44,7 @@ import
org.apache.seatunnel.connectors.seatunnel.pulsar.source.split.PulsarParti
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClient;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -57,6 +59,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
class PulsarSourceReaderRestoreTest {
@@ -230,6 +233,45 @@ class PulsarSourceReaderRestoreTest {
collector.records.get(1).getTableId()));
}
+ @Test
+ void shouldCloseSplitReadersBeforeClientWithConnectorClassLoader() throws
Exception {
+ TablePath tablePath = TablePath.of("db.orders");
+ List<String> closeOrder = new ArrayList<>();
+ AtomicReference<ClassLoader> splitCloseClassLoader = new
AtomicReference<>();
+ AtomicReference<ClassLoader> clientCloseClassLoader = new
AtomicReference<>();
+ CloseTrackingPulsarSourceReader reader =
+ new CloseTrackingPulsarSourceReader(
+ new TestingReaderContext(),
+ Collections.singletonMap(tablePath,
createMetadata(tablePath)),
+ closeOrder,
+ splitCloseClassLoader);
+ reader.pulsarClient = createPulsarClientProxy(closeOrder,
clientCloseClassLoader);
+ PulsarPartitionSplit split =
+ new PulsarPartitionSplit(
+ new
TopicPartition("persistent://public/default/orders", 0),
+ StopCursor.never(),
+ null,
+ tablePath);
+ reader.addSplits(Collections.singletonList(split));
+
+ ClassLoader originalClassLoader =
Thread.currentThread().getContextClassLoader();
+ ClassLoader unrelatedClassLoader = new ClassLoader(null) {};
+ try {
+ Thread.currentThread().setContextClassLoader(unrelatedClassLoader);
+
+ reader.close();
+
+ Assertions.assertSame(
+ unrelatedClassLoader,
Thread.currentThread().getContextClassLoader());
+ } finally {
+ Thread.currentThread().setContextClassLoader(originalClassLoader);
+ }
+ Assertions.assertEquals(Arrays.asList("split", "client"), closeOrder);
+ Assertions.assertSame(PulsarConfigUtil.class.getClassLoader(),
splitCloseClassLoader.get());
+ Assertions.assertSame(
+ PulsarConfigUtil.class.getClassLoader(),
clientCloseClassLoader.get());
+ }
+
private static PulsarConsumerMetadata createMetadata(TablePath tablePath) {
return createMetadata(tablePath, new TestingDeserializationSchema());
}
@@ -305,6 +347,34 @@ class PulsarSourceReaderRestoreTest {
});
}
+ @SuppressWarnings("unchecked")
+ private static PulsarClient createPulsarClientProxy(
+ List<String> closeOrder, AtomicReference<ClassLoader>
closeClassLoader) {
+ return (PulsarClient)
+ Proxy.newProxyInstance(
+ PulsarClient.class.getClassLoader(),
+ new Class<?>[] {PulsarClient.class},
+ (proxy, method, args) -> {
+ if ("close".equals(method.getName())) {
+ closeOrder.add("client");
+ closeClassLoader.set(
+
Thread.currentThread().getContextClassLoader());
+ return null;
+ }
+ Class<?> returnType = method.getReturnType();
+ if (returnType.equals(boolean.class)) {
+ return false;
+ }
+ if (returnType.equals(int.class)) {
+ return 0;
+ }
+ if (returnType.equals(long.class)) {
+ return 0L;
+ }
+ return null;
+ });
+ }
+
private static final class TestingPulsarSourceReader extends
PulsarSourceReader<SeaTunnelRow> {
private TestingPulsarSourceReader(
@@ -372,6 +442,76 @@ class PulsarSourceReaderRestoreTest {
public void close() throws IOException {}
}
+ /** Source reader that records the close order without opening a real
Pulsar consumer. */
+ private static final class CloseTrackingPulsarSourceReader
+ extends PulsarSourceReader<SeaTunnelRow> {
+
+ private final List<String> closeOrder;
+ private final AtomicReference<ClassLoader> splitCloseClassLoader;
+
+ private CloseTrackingPulsarSourceReader(
+ SourceReader.Context context,
+ Map<TablePath, PulsarConsumerMetadata> metadataMap,
+ List<String> closeOrder,
+ AtomicReference<ClassLoader> splitCloseClassLoader) {
+ super(
+ context,
+
PulsarClientConfig.builder().serviceUrl("pulsar://localhost:6650").build(),
+ metadataMap,
+ false,
+ 100,
+ 50L,
+ 1);
+ this.closeOrder = closeOrder;
+ this.splitCloseClassLoader = splitCloseClassLoader;
+ }
+
+ @Override
+ protected PulsarSplitReaderThread createPulsarSplitReaderThread(
+ PulsarPartitionSplit split) {
+ return new CloseTrackingSplitReaderThread(
+ this, split, handover, closeOrder, splitCloseClassLoader);
+ }
+ }
+
+ /** Split reader that records cleanup metadata without touching a Pulsar
broker. */
+ private static final class CloseTrackingSplitReaderThread extends
PulsarSplitReaderThread {
+
+ private final List<String> closeOrder;
+ private final AtomicReference<ClassLoader> closeClassLoader;
+
+ private CloseTrackingSplitReaderThread(
+ PulsarSourceReader sourceReader,
+ PulsarPartitionSplit split,
+ org.apache.seatunnel.common.Handover<RecordWithSplitId>
handover,
+ List<String> closeOrder,
+ AtomicReference<ClassLoader> closeClassLoader) {
+ super(
+ sourceReader,
+ split,
+ null,
+
PulsarConsumerConfig.builder().subscriptionName("seatunnel-sub").build(),
+ 100,
+ 50L,
+ StartCursor.earliest(),
+ handover);
+ this.closeOrder = closeOrder;
+ this.closeClassLoader = closeClassLoader;
+ }
+
+ @Override
+ public void open() {}
+
+ @Override
+ public synchronized void start() {}
+
+ @Override
+ public void close() {
+ closeOrder.add("split");
+
closeClassLoader.set(Thread.currentThread().getContextClassLoader());
+ }
+ }
+
private static final class TestingReaderContext implements
SourceReader.Context {
private int noMoreElementSignals;
