This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 35800edf39f8 CAMEL-26050: camel-thrift - stop the TAsyncClientManager
of the asynchronous producer when it stops (#28537)
35800edf39f8 is described below
commit 35800edf39f8fb1cbb59091a81eaa4d07105d8d0
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Oct 27 19:06:54 2026 +0530
CAMEL-26050: camel-thrift - stop the TAsyncClientManager of the
asynchronous producer when it stops (#28537)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/component/thrift/ThriftProducer.java | 136 ++++++++-
.../apache/camel/component/thrift/ThriftUtils.java | 27 +-
.../ThriftProducerAsyncInFlightStopTest.java | 308 +++++++++++++++++++++
.../thrift/ThriftProducerAsyncStopTest.java | 67 +++++
4 files changed, 524 insertions(+), 14 deletions(-)
diff --git
a/components/camel-thrift/src/main/java/org/apache/camel/component/thrift/ThriftProducer.java
b/components/camel-thrift/src/main/java/org/apache/camel/component/thrift/ThriftProducer.java
index 1ee34cafd47a..73cb2b3bed73 100644
---
a/components/camel-thrift/src/main/java/org/apache/camel/component/thrift/ThriftProducer.java
+++
b/components/camel-thrift/src/main/java/org/apache/camel/component/thrift/ThriftProducer.java
@@ -17,9 +17,13 @@
package org.apache.camel.component.thrift;
import java.io.IOException;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.camel.AsyncCallback;
import org.apache.camel.AsyncProducer;
+import org.apache.camel.CamelExchangeException;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.component.thrift.client.AsyncClientMethodCallback;
@@ -28,6 +32,7 @@ import org.apache.camel.support.ResourceHelper;
import org.apache.camel.support.jsse.SSLContextParameters;
import org.apache.camel.util.ObjectHelper;
import org.apache.thrift.TException;
+import org.apache.thrift.async.TAsyncClientManager;
import org.apache.thrift.transport.TNonblockingSocket;
import org.apache.thrift.transport.TNonblockingTransport;
import org.apache.thrift.transport.TSSLTransportFactory;
@@ -48,7 +53,10 @@ public class ThriftProducer extends DefaultAsyncProducer
implements AsyncProduce
protected final ThriftEndpoint endpoint;
private TTransport syncTransport;
private TNonblockingTransport asyncTransport;
- private Object thriftClient;
+ private TAsyncClientManager asyncClientManager;
+ private volatile Object thriftClient;
+ // the callbacks of the asynchronous calls that have not completed yet,
failed when this producer stops
+ private final Set<PendingCallback> pendingCallbacks =
ConcurrentHashMap.newKeySet();
public ThriftProducer(ThriftEndpoint endpoint, ThriftConfiguration
configuration) {
super(endpoint);
@@ -60,10 +68,23 @@ public class ThriftProducer extends DefaultAsyncProducer
implements AsyncProduce
public boolean process(Exchange exchange, AsyncCallback callback) {
Message message = exchange.getIn();
+ PendingCallback pending = new PendingCallback(exchange, callback);
+ pendingCallbacks.add(pending);
+ // read after the registration: doStop clears the client before it
fails the pending calls
+ Object client = thriftClient;
+ if (client == null) {
+ // the producer is stopped
+ pending.stopped();
+ return false;
+ }
try {
- ThriftUtils.invokeAsyncMethod(thriftClient,
configuration.getMethod(), message.getBody(),
- new AsyncClientMethodCallback(exchange, callback));
+ ThriftUtils.invokeAsyncMethod(client, configuration.getMethod(),
message.getBody(), pending);
} catch (Exception e) {
+ // the call was not made
+ if (!pending.discard()) {
+ // a stop of the producer completed the exchange meanwhile,
with done(false)
+ return false;
+ }
if (e.getCause() instanceof TException) {
exchange.setException(e.getCause());
} else {
@@ -72,6 +93,12 @@ public class ThriftProducer extends DefaultAsyncProducer
implements AsyncProduce
callback.done(true);
return true;
}
+ if (client != thriftClient) {
+ // the producer stopped meanwhile, and may have started again with
a new client: the stop may have failed
+ // the pending calls before this one was registered, and the
selector thread of the stopped client manager,
+ // which may still be ending, accepts the call without ever
completing it
+ pending.stopped();
+ }
return false;
}
@@ -124,27 +151,112 @@ public class ThriftProducer extends DefaultAsyncProducer
implements AsyncProduce
if (asyncTransport == null) {
initializeAsyncTransport();
LOG.info("Getting asynchronous client implementation");
+ asyncClientManager = new TAsyncClientManager();
thriftClient =
ThriftUtils.constructAsyncClientInstance(endpoint.getServicePackage(),
endpoint.getServiceName(),
asyncTransport,
- configuration.getExchangeProtocol(),
endpoint.getCamelContext());
+ configuration.getExchangeProtocol(),
endpoint.getCamelContext(), asyncClientManager);
}
}
}
@Override
protected void doStop() throws Exception {
- if (syncTransport != null) {
- LOG.debug("Terminating synchronous transport the remote Thrift
server");
- syncTransport.close();
- syncTransport = null;
- } else if (asyncTransport != null) {
- LOG.debug("Terminating asynchronous transport the remote Thrift
server");
- asyncTransport.close();
- asyncTransport = null;
+ try {
+ if (syncTransport != null) {
+ LOG.debug("Terminating synchronous transport the remote Thrift
server");
+ syncTransport.close();
+ syncTransport = null;
+ } else if (asyncTransport != null) {
+ // before the pending calls are failed: process() fails a call
whose client was cleared or replaced
+ thriftClient = null;
+ LOG.debug("Terminating asynchronous transport the remote
Thrift server");
+ asyncTransport.close();
+ asyncTransport = null;
+ }
+ } finally {
+ if (asyncClientManager != null) {
+ // stops the selector thread and closes the selector of the
async client
+ asyncClientManager.stop();
+ asyncClientManager = null;
+ }
+ // after the stop, so that a call registered meanwhile is failed
too
+ failPendingCallbacks();
}
super.doStop();
}
+ /**
+ * Fails the exchanges of the asynchronous calls that have not completed:
the stopped selector thread does not
+ * complete them. A call that completes meanwhile is completed once, and a
continuation that throws does not skip
+ * the other calls.
+ */
+ private void failPendingCallbacks() {
+ for (PendingCallback pending : pendingCallbacks) {
+ try {
+ pending.stopped();
+ } catch (Exception e) {
+ LOG.warn("Error completing the exchange of a pending call of
method {} when stopping the Thrift producer",
+ configuration.getMethod(), e);
+ }
+ }
+ }
+
+ /**
+ * The number of asynchronous calls that have not completed yet (for
tests).
+ */
+ int getPendingCallbackCount() {
+ return pendingCallbacks.size();
+ }
+
+ /**
+ * Completes the exchange of an asynchronous call once: with the response
or error of the call, or with an error
+ * when the producer stops first.
+ */
+ private final class PendingCallback extends AsyncClientMethodCallback {
+ private final Exchange exchange;
+ private final AtomicBoolean done = new AtomicBoolean();
+
+ PendingCallback(Exchange exchange, AsyncCallback callback) {
+ super(exchange, callback);
+ this.exchange = exchange;
+ }
+
+ @Override
+ public void onComplete(Object response) {
+ if (finish()) {
+ super.onComplete(response);
+ }
+ }
+
+ @Override
+ public void onError(Exception exception) {
+ if (finish()) {
+ super.onError(exception);
+ }
+ }
+
+ void stopped() {
+ if (finish()) {
+ super.onError(new CamelExchangeException(
+ "The Thrift producer was stopped before the call of
method " + configuration.getMethod()
+ + " completed",
+ exchange));
+ }
+ }
+
+ boolean discard() {
+ return finish();
+ }
+
+ private boolean finish() {
+ if (done.compareAndSet(false, true)) {
+ pendingCallbacks.remove(this);
+ return true;
+ }
+ return false;
+ }
+ }
+
protected void initializeSyncTransport() throws TTransportException {
if (!ObjectHelper.isEmpty(configuration.getHost()) &&
!ObjectHelper.isEmpty(configuration.getPort())) {
LOG.info("Creating transport to the remote Thrift server {}:{}",
configuration.getHost(), configuration.getPort());
diff --git
a/components/camel-thrift/src/main/java/org/apache/camel/component/thrift/ThriftUtils.java
b/components/camel-thrift/src/main/java/org/apache/camel/component/thrift/ThriftUtils.java
index 86a75308d885..aff764d9b6af 100644
---
a/components/camel-thrift/src/main/java/org/apache/camel/component/thrift/ThriftUtils.java
+++
b/components/camel-thrift/src/main/java/org/apache/camel/component/thrift/ThriftUtils.java
@@ -87,10 +87,33 @@ public final class ThriftUtils {
return clientInstance;
}
+ /**
+ * Constructs an async client with a new client manager, which the caller
cannot stop: its selector thread and
+ * selector stay until the JVM exits.
+ *
+ * @deprecated use the overload with a {@link TAsyncClientManager}
argument, and stop that manager when the client
+ * is no longer used.
+ */
+ @Deprecated(since = "4.24.0")
@SuppressWarnings({ "rawtypes", "unchecked" })
public static Object constructAsyncClientInstance(
String packageName, String serviceName, TNonblockingTransport
transport, ThriftExchangeProtocol exchangeProtocol,
final CamelContext context) {
+ try {
+ return constructAsyncClientInstance(packageName, serviceName,
transport, exchangeProtocol, context,
+ new TAsyncClientManager());
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Thrift async client manager
cannot be created", e);
+ }
+ }
+
+ /**
+ * Constructs an async client that uses the given client manager; the
caller owns the manager and must stop it.
+ */
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ public static Object constructAsyncClientInstance(
+ String packageName, String serviceName, TNonblockingTransport
transport, ThriftExchangeProtocol exchangeProtocol,
+ final CamelContext context, TAsyncClientManager clientManager) {
Object asynClientInstance = null;
Class[] getterParamTypes = { TNonblockingTransport.class };
Class[] constructorParamTypes = { TAsyncClientManager.class,
TProtocolFactory.class };
@@ -101,7 +124,7 @@ public final class ThriftUtils {
Class clientClass =
context.getClassResolver().resolveMandatoryClass(clientClassName);
Constructor factoryConstructor =
clientClass.getConstructor(constructorParamTypes);
Object factoryInstance
- = factoryConstructor.newInstance(new
TAsyncClientManager(), constructAsyncProtocol(exchangeProtocol));
+ = factoryConstructor.newInstance(clientManager,
constructAsyncProtocol(exchangeProtocol));
Method asyncClientGetter =
ReflectionHelper.findMethod(clientClass,
ThriftConstants.THRIFT_ASYNC_CLIENT_GETTER_NAME,
getterParamTypes);
if (asyncClientGetter == null) {
@@ -116,7 +139,7 @@ public final class ThriftUtils {
} catch (NoSuchMethodException e) {
throw new IllegalArgumentException("Thrift sync client factory
class not found: " + clientClassName);
} catch (InstantiationException | IllegalAccessException |
IllegalArgumentException | InvocationTargetException
- | IOException | SecurityException e) {
+ | SecurityException e) {
throw new IllegalArgumentException(e);
}
return asynClientInstance;
diff --git
a/components/camel-thrift/src/test/java/org/apache/camel/component/thrift/ThriftProducerAsyncInFlightStopTest.java
b/components/camel-thrift/src/test/java/org/apache/camel/component/thrift/ThriftProducerAsyncInFlightStopTest.java
new file mode 100644
index 000000000000..ead635146885
--- /dev/null
+++
b/components/camel-thrift/src/test/java/org/apache/camel/component/thrift/ThriftProducerAsyncInFlightStopTest.java
@@ -0,0 +1,308 @@
+/*
+ * 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.camel.component.thrift;
+
+import java.io.InputStream;
+import java.lang.reflect.Field;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelExchangeException;
+import org.apache.camel.Exchange;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.apache.thrift.TException;
+import org.apache.thrift.async.AsyncMethodCallback;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Stopping the client manager of the asynchronous producer ends its selector
thread without completing the calls it
+ * still has: the producer must fail their exchanges when it stops, else they
never complete.
+ */
+class ThriftProducerAsyncInFlightStopTest extends CamelTestSupport {
+
+ @Test
+ void stoppingTheProducerFailsAnInFlightCall() throws Exception {
+ CountDownLatch requestReceived = new CountDownLatch(1);
+ try (ServerSocket server = new ServerSocket(0)) {
+ startSilentServer(server, requestReceived);
+
+ String uri = pingUri(server.getLocalPort());
+ ThriftProducer producer = (ThriftProducer)
context.getEndpoint(uri).createProducer();
+ producer.start();
+ try {
+ Exchange exchange = context.getEndpoint(uri).createExchange();
+ CountDownLatch done = new CountDownLatch(1);
+ producer.process(exchange, doneSync -> done.countDown());
+ assertTrue(requestReceived.await(10, TimeUnit.SECONDS), "the
call must be sent");
+ assertEquals(1, producer.getPendingCallbackCount());
+
+ producer.stop();
+
+ assertTrue(done.await(10, TimeUnit.SECONDS), "the exchange of
the call must complete");
+ assertStoppedBeforeTheCallCompleted(exchange);
+ assertEquals(0, producer.getPendingCallbackCount());
+ } finally {
+ producer.stop();
+ }
+ }
+ }
+
+ @Test
+ void aContinuationThatThrowsDoesNotKeepTheSelectorThreadRunning() throws
Exception {
+ CountDownLatch requestReceived = new CountDownLatch(1);
+ try (ServerSocket server = new ServerSocket(0)) {
+ startSilentServer(server, requestReceived);
+
+ String uri = pingUri(server.getLocalPort());
+ ThriftProducer producer = (ThriftProducer)
context.getEndpoint(uri).createProducer();
+ Set<Thread> before = Thread.getAllStackTraces().keySet();
+ producer.start();
+ try {
+ List<Thread> selectors =
Thread.getAllStackTraces().keySet().stream()
+ .filter(t -> !before.contains(t) &&
t.getName().startsWith("TAsyncClientManager#SelectorThread"))
+ .toList();
+ assertEquals(1, selectors.size(), "the selector thread of the
producer");
+
+ Exchange exchange = context.getEndpoint(uri).createExchange();
+ CountDownLatch done = new CountDownLatch(1);
+ producer.process(exchange, doneSync -> {
+ done.countDown();
+ throw new IllegalStateException("the continuation of the
exchange fails");
+ });
+ assertTrue(requestReceived.await(10, TimeUnit.SECONDS), "the
call must be sent");
+
+ // the continuation runs on the stopping thread and throws
+ assertDoesNotThrow(producer::stop);
+
+ assertTrue(done.await(10, TimeUnit.SECONDS), "the exchange of
the call must complete");
+ assertStoppedBeforeTheCallCompleted(exchange);
+ assertEquals(0, producer.getPendingCallbackCount());
+ selectors.get(0).join(TimeUnit.SECONDS.toMillis(10));
+ assertFalse(selectors.get(0).isAlive(), "the selector thread
must end");
+ } finally {
+ producer.stop();
+ }
+ }
+ }
+
+ @Test
+ void aCompletedCallIsNotPending() throws Exception {
+ // control: the callback of a call that completed (here with an error:
nothing listens) is not kept
+ int port;
+ try (ServerSocket closed = new ServerSocket(0)) {
+ port = closed.getLocalPort();
+ }
+ String uri = pingUri(port);
+ ThriftProducer producer = (ThriftProducer)
context.getEndpoint(uri).createProducer();
+ producer.start();
+ try {
+ Exchange exchange = context.getEndpoint(uri).createExchange();
+ CountDownLatch done = new CountDownLatch(1);
+ producer.process(exchange, doneSync -> done.countDown());
+
+ assertTrue(done.await(10, TimeUnit.SECONDS));
+ assertTrue(exchange.isFailed());
+ assertEquals(0, producer.getPendingCallbackCount());
+ } finally {
+ producer.stop();
+ }
+ }
+
+ @Test
+ void aCallThatFailsToStartWhileTheProducerStopsIsCompletedOnce() throws
Exception {
+ ThriftProducer producer = startProducerOnClosedPort();
+ try {
+ // the stop runs between the registration of the call and the
failure to start it: it completes the
+ // exchange (asynchronously), so the failure must not complete it
again
+ setClient(producer, new StoppingClient(producer));
+ Exchange exchange = producer.getEndpoint().createExchange();
+ List<Boolean> done = new CopyOnWriteArrayList<>();
+
+ boolean sync = producer.process(exchange, done::add);
+
+ assertEquals(List.of(false), done, "the exchange must complete
once, asynchronously");
+ assertFalse(sync, "process must return the doneSync passed to the
callback");
+ assertStoppedBeforeTheCallCompleted(exchange);
+ assertEquals(0, producer.getPendingCallbackCount());
+ } finally {
+ producer.stop();
+ }
+ }
+
+ @Test
+ void aCallThatFailsToStartIsCompletedSynchronously() throws Exception {
+ // control: without a stop, the failure to start the call completes
the exchange synchronously
+ ThriftProducer producer = startProducerOnClosedPort();
+ try {
+ setClient(producer, new FailingClient());
+ Exchange exchange = producer.getEndpoint().createExchange();
+ List<Boolean> done = new CopyOnWriteArrayList<>();
+
+ boolean sync = producer.process(exchange, done::add);
+
+ assertEquals(List.of(true), done);
+ assertTrue(sync);
+ assertInstanceOf(TException.class, exchange.getException());
+ assertEquals(0, producer.getPendingCallbackCount());
+ } finally {
+ producer.stop();
+ }
+ }
+
+ @Test
+ void aCallAcceptedAfterTheProducerFailedItsPendingCallsIsFailed() throws
Exception {
+ aCallTheStopDoesNotSeeIsFailed(false);
+ }
+
+ @Test
+ void
aCallAcceptedAfterTheProducerFailedItsPendingCallsIsFailedAlsoWhenTheProducerStartsAgain()
throws Exception {
+ aCallTheStopDoesNotSeeIsFailed(true);
+ }
+
+ private void aCallTheStopDoesNotSeeIsFailed(boolean startAgain) throws
Exception {
+ ThriftProducer producer = startProducerOnClosedPort();
+ try {
+ // the stop fails the pending calls while this call is being
registered, so it does not see it, and a
+ // selector thread that is still ending accepts the call and never
completes it
+ setClient(producer, new UnseenByTheStopClient(producer,
startAgain));
+ Exchange exchange = producer.getEndpoint().createExchange();
+ List<Boolean> done = new CopyOnWriteArrayList<>();
+
+ boolean sync = producer.process(exchange, done::add);
+
+ assertEquals(List.of(false), done, "the exchange must complete
once");
+ assertFalse(sync);
+ assertStoppedBeforeTheCallCompleted(exchange);
+ assertEquals(0, producer.getPendingCallbackCount());
+ } finally {
+ producer.stop();
+ }
+ }
+
+ @Test
+ void aCallToAStoppedProducerIsFailed() throws Exception {
+ ThriftProducer producer = startProducerOnClosedPort();
+ producer.stop();
+
+ Exchange exchange = producer.getEndpoint().createExchange();
+ List<Boolean> done = new CopyOnWriteArrayList<>();
+ boolean sync = producer.process(exchange, done::add);
+
+ assertEquals(List.of(false), done, "the exchange must complete once");
+ assertFalse(sync);
+ assertStoppedBeforeTheCallCompleted(exchange);
+ assertEquals(0, producer.getPendingCallbackCount());
+ }
+
+ private ThriftProducer startProducerOnClosedPort() throws Exception {
+ int port;
+ try (ServerSocket closed = new ServerSocket(0)) {
+ port = closed.getLocalPort();
+ }
+ ThriftProducer producer = (ThriftProducer)
context.getEndpoint(pingUri(port)).createProducer();
+ producer.start();
+ return producer;
+ }
+
+ private static void setClient(ThriftProducer producer, Object client)
throws Exception {
+ Field field = ThriftProducer.class.getDeclaredField("thriftClient");
+ field.setAccessible(true);
+ field.set(producer, client);
+ }
+
+ // stops the producer, then fails to start the call as a stopped client
manager does
+ public static class StoppingClient {
+ private final ThriftProducer producer;
+
+ StoppingClient(ThriftProducer producer) {
+ this.producer = producer;
+ }
+
+ public void ping(AsyncMethodCallback<Void> resultHandler) throws
TException {
+ producer.stop();
+ throw new TException("SelectThread is not running");
+ }
+ }
+
+ public static class FailingClient {
+ public void ping(AsyncMethodCallback<Void> resultHandler) throws
TException {
+ throw new TException("SelectThread is not running");
+ }
+ }
+
+ // accepts the call and never completes it, while the producer stops
without seeing the call (and may start again)
+ public static class UnseenByTheStopClient {
+ private final ThriftProducer producer;
+ private final boolean startAgain;
+
+ UnseenByTheStopClient(ThriftProducer producer, boolean startAgain) {
+ this.producer = producer;
+ this.startAgain = startAgain;
+ }
+
+ public void ping(AsyncMethodCallback<Void> resultHandler) throws
Exception {
+ // the stop fails the pending calls before this one is registered
+ Field field =
ThriftProducer.class.getDeclaredField("pendingCallbacks");
+ field.setAccessible(true);
+ ((Set<?>) field.get(producer)).clear();
+ producer.stop();
+ if (startAgain) {
+ producer.start();
+ }
+ }
+ }
+
+ private static String pingUri(int port) {
+ return "thrift://localhost:" + port +
"/org.apache.camel.component.thrift.generated.Calculator?method=ping";
+ }
+
+ // a server that reads the request and never answers
+ private static void startSilentServer(ServerSocket server, CountDownLatch
requestReceived) {
+ Thread silent = new Thread(() -> {
+ try (Socket socket = server.accept(); InputStream in =
socket.getInputStream()) {
+ if (in.read() != -1) {
+ requestReceived.countDown();
+ }
+ while (in.read() != -1) {
+ // keep the connection open until the client closes it
+ }
+ } catch (Exception e) {
+ // the test ends
+ }
+ });
+ silent.setDaemon(true);
+ silent.start();
+ }
+
+ private static void assertStoppedBeforeTheCallCompleted(Exchange exchange)
{
+ assertInstanceOf(CamelExchangeException.class,
exchange.getException());
+ assertTrue(exchange.getException().getMessage().startsWith(
+ "The Thrift producer was stopped before the call of method
ping completed"),
+ exchange.getException().getMessage());
+ }
+}
diff --git
a/components/camel-thrift/src/test/java/org/apache/camel/component/thrift/ThriftProducerAsyncStopTest.java
b/components/camel-thrift/src/test/java/org/apache/camel/component/thrift/ThriftProducerAsyncStopTest.java
new file mode 100644
index 000000000000..6e2257193354
--- /dev/null
+++
b/components/camel-thrift/src/test/java/org/apache/camel/component/thrift/ThriftProducerAsyncStopTest.java
@@ -0,0 +1,67 @@
+/*
+ * 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.camel.component.thrift;
+
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * The asynchronous producer (the default, synchronous=false) creates a
TAsyncClientManager (a selector thread and a
+ * selector) on every start; stopping the producer must stop it.
+ */
+class ThriftProducerAsyncStopTest extends ThriftProducerBaseTest {
+
+ @Test
+ void routeRestartsStopTheClientManagersOfTheStoppedProducers() throws
Exception {
+ Set<Thread> before = Thread.getAllStackTraces().keySet();
+ for (int i = 0; i < 5; i++) {
+ context.getRouteController().stopRoute("r");
+ context.getRouteController().startRoute("r");
+ // the asynchronous producer of the restarted route works
+ assertNull(template.requestBody("direct:thrift-ping", (Object)
null));
+ }
+ context.getRouteController().stopRoute("r");
+
+ await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> {
+ long alive = Thread.getAllStackTraces().keySet().stream()
+ .filter(t -> !before.contains(t) && t.isAlive()
+ &&
t.getName().startsWith("TAsyncClientManager#SelectorThread"))
+ .count();
+ assertEquals(0, alive, "selector threads of stopped asynchronous
producers");
+ });
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:thrift-ping").routeId("r")
+ .to("thrift://localhost:" + thriftTestPort
+ +
"/org.apache.camel.component.thrift.generated.Calculator?method=ping");
+ }
+ };
+ }
+}
