From e1565dc7858c4eb1cf32e84e8188e055a1cc053e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Mon, 19 Oct 2026 09:39:28 +0300 Subject: [PATCH 1/2] CAMEL-25812: camel-milo - end idle client threads and run reconnects on a thread of their own Since CAMEL-25713, a client connection handled the outcome of its requests and ran its reconnects on a scheduled thread pool of its own with one thread per processor. The threads of a scheduled pool never end while the connection is open, so 30 connections on a machine with 8 processors kept 240 idle threads. With virtual threads, the pool had a single worker, so after awaited requests the routes of a connection ran one at a time: one connection with 32 callers and 10 ms of work after each awaited write fell from about 670 to 85 exchanges per second. A connection now handles the outcome of its write, method call, read and browse requests on a thread pool with up to one thread per processor, whose idle threads end after the keep-alive time of the default thread pool profile. With virtual threads and more than one processor, the pool starts a virtual thread for each task, and the same measurement gives about 2,100 to 2,300 exchanges per second. The pool stays bounded with platform threads, because when the connection is lost every request in flight fails and waits for the disconnect of the first failure: a pool that starts a thread for each of them ran out of threads with writes that are not awaited, and the connection was not rebuilt. The reconnects run one after the other on a single-thread scheduled executor of the connection, so a reconnect no longer waits behind the outcomes of failed requests. Disposing the connection cancels a reconnect that is still waiting, so the thread ends when the connection is closed. Neither executor is the shared scheduled executor of Milo, so a failed request still disconnects without waiting for a thread that only the disconnect could free. A request that completes after the connection is closed still fails with a RejectedExecutionException. Co-authored-by: Claude Opus 5.5 --- .../milo/client/MiloClientConnection.java | 44 ++++++++++++++----- .../client/internal/SubscriptionManager.java | 18 +++++--- .../milo/MiloClientExecutorTest.java | 32 +++++++++++++- .../pages/camel-4x-upgrade-guide-4_24.adoc | 13 +++--- 4 files changed, 83 insertions(+), 24 deletions(-) diff --git a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConnection.java b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConnection.java index 02226f6aa2ba2..f0837e2fe1aa8 100644 --- a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConnection.java +++ b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConnection.java @@ -20,6 +20,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -29,6 +30,7 @@ import org.apache.camel.CamelContextAware; import org.apache.camel.RuntimeCamelException; import org.apache.camel.component.milo.client.internal.SubscriptionManager; +import org.apache.camel.spi.ExecutorServiceManager; import org.apache.camel.spi.ThreadPoolProfile; import org.apache.camel.util.ObjectHelper; import org.apache.camel.util.concurrent.ThreadPoolRejectedPolicy; @@ -49,7 +51,8 @@ public class MiloClientConnection implements AutoCloseable, CamelContextAware { private final Lock lock = new ReentrantLock(); private final MiloClientConfiguration configuration; private volatile CamelContext camelContext; - private ScheduledExecutorService executor; + private ExecutorService executor; + private ScheduledExecutorService reconnectExecutor; private SubscriptionManager manager; private volatile boolean initialized; @@ -101,18 +104,28 @@ protected void init() { ObjectHelper.notNull(this.camelContext, "camelContext"); // A failed request disconnects the client, which can wait for a reconnect attempt that Milo schedules on its shared - // scheduled executor, so this connection handles its requests and reconnects on an executor of its own. + // scheduled executor, so this connection handles its requests and reconnects on executors of its own. + final ExecutorServiceManager executorServiceManager = this.camelContext.getExecutorServiceManager(); final ThreadPoolProfile profile = new ThreadPoolProfile("MiloClient"); - // as many threads as that executor; with one processor or with virtual threads this is one thread, so a route - // that continues on it after an awaited request must not wait for another request of this connection - profile.setPoolSize(Runtime.getRuntime().availableProcessors()); - // not bounded, like that executor, because a rejected reconnect would end the reconnects of this connection - profile.setMaxQueueSize(-1); + // as many threads as that executor, but they end when they are idle; with virtual threads and more than one + // processor, a virtual thread for each task. Not more platform threads, because when the connection is lost, + // every request in flight fails and waits for the disconnect of the first failure. With one processor this is + // one thread, so a route that continues on it after an awaited request must not wait for another request of + // this connection. + final int threads = Runtime.getRuntime().availableProcessors(); + profile.setPoolSize(threads); + profile.setMaxPoolSize(threads); + // not bounded, so that no request outcome is rejected while the connection is open; a size of -1 would give a + // SynchronousQueue, which rejects a task while all the threads are busy + profile.setMaxQueueSize(Integer.MAX_VALUE); // a request that completes after the connection is closed fails instead of being dropped profile.setRejectedPolicy(ThreadPoolRejectedPolicy.Abort); - this.executor = this.camelContext.getExecutorServiceManager().newScheduledThreadPool(this, "MiloClient", profile); + this.executor = executorServiceManager.newThreadPool(this, "MiloClient", profile); + // the reconnects of the connection run one after the other, on a thread that starts with the first one and ends + // when the connection is closed + this.reconnectExecutor = executorServiceManager.newSingleThreadScheduledExecutor(this, "MiloClientReconnect"); - this.manager = new SubscriptionManager(this.configuration, this.executor, 10_000); + this.manager = new SubscriptionManager(this.configuration, this.reconnectExecutor, this.executor, 10_000); } @Override @@ -125,9 +138,16 @@ public void close() throws Exception { this.manager = null; } } finally { - if (this.executor != null) { - this.camelContext.getExecutorServiceManager().shutdown(this.executor); - this.executor = null; + try { + if (this.reconnectExecutor != null) { + this.camelContext.getExecutorServiceManager().shutdown(this.reconnectExecutor); + this.reconnectExecutor = null; + } + } finally { + if (this.executor != null) { + this.camelContext.getExecutorServiceManager().shutdown(this.executor); + this.executor = null; + } } } } finally { diff --git a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java index 314a55b4625db..d2a6efcbd417a 100644 --- a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java +++ b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java @@ -717,7 +717,8 @@ nodeId, direction, NodeIds.References, includeSubTypes, uint(nodeClasses), } private final MiloClientConfiguration configuration; - private final ScheduledExecutorService executor; + private final ScheduledExecutorService reconnectExecutor; + private final Executor executor; private final long reconnectTimeout; private final ServerCertificateTrust serverCertificateTrust; private final Lock lock = new ReentrantLock(); @@ -728,10 +729,11 @@ nodeId, direction, NodeIds.References, includeSubTypes, uint(nodeClasses), private Future reconnectJob; private final Map subscriptions = new HashMap<>(); - public SubscriptionManager(final MiloClientConfiguration configuration, final ScheduledExecutorService executor, - final long reconnectTimeout) { + public SubscriptionManager(final MiloClientConfiguration configuration, final ScheduledExecutorService reconnectExecutor, + final Executor executor, final long reconnectTimeout) { this.configuration = configuration; + this.reconnectExecutor = reconnectExecutor; this.executor = executor; this.reconnectTimeout = reconnectTimeout; this.serverCertificateTrust = new ServerCertificateTrust(configuration); @@ -1101,6 +1103,11 @@ public void dispose() { } this.disposed = true; connected = this.connected; + if (this.reconnectJob != null) { + // a waiting reconnect would keep the thread of the reconnect executor until it is due + this.reconnectJob.cancel(false); + this.reconnectJob = null; + } } finally { lock.unlock(); } @@ -1128,9 +1135,10 @@ private void triggerReconnect(final boolean immediate) { } if (immediate) { - this.reconnectJob = this.executor.submit(this::connect); + this.reconnectJob = this.reconnectExecutor.submit(this::connect); } else { - this.reconnectJob = this.executor.schedule(this::connect, this.reconnectTimeout, TimeUnit.MILLISECONDS); + this.reconnectJob + = this.reconnectExecutor.schedule(this::connect, this.reconnectTimeout, TimeUnit.MILLISECONDS); } } finally { lock.unlock(); diff --git a/components/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientExecutorTest.java b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientExecutorTest.java index a4be2a00135be..6701000742956 100644 --- a/components/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientExecutorTest.java +++ b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientExecutorTest.java @@ -27,6 +27,7 @@ import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.milo.server.MiloServerComponent; import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.test.AvailablePortFinder; import org.eclipse.milo.opcua.stack.core.Stack; import org.eclipse.milo.opcua.stack.core.types.builtin.Variant; import org.junit.jupiter.api.Test; @@ -45,6 +46,11 @@ public class MiloClientExecutorTest extends AbstractMiloServerTest { = "milo-client:opc.tcp://foo:bar@localhost:@@port@@?node=" + nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "myitem1") + "&overrideHost=true"; + // no server listens on this port + private static final String MILO_CLIENT_UNREACHABLE + = "milo-client:opc.tcp://localhost:" + AvailablePortFinder.getNextAvailable() + "?node=" + + nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "myitem1") + "&overrideHost=true"; + @EndpointInject("mock:item") protected MockEndpoint item; @@ -55,6 +61,7 @@ protected RoutesBuilder createRouteBuilder() { public void configure() { from("milo-server:myitem1").to("mock:item"); from("direct:write").routeId("write").to(resolve(MILO_CLIENT_ITEM)); + from(MILO_CLIENT_UNREACHABLE).routeId("unreachable").autoStartup(false).to("mock:unreachable"); } }; } @@ -101,7 +108,8 @@ void awaitedWriteCompletesWhileTheSharedScheduledExecutorOfMiloIsBusy() throws E } /** - * The executor of a connection is shut down when the last endpoint releases the connection. + * The pool that handles the outcome of the requests of a connection is shut down when the last endpoint releases + * the connection. */ @Test void executorIsShutDownWithTheConnection() throws Exception { @@ -114,14 +122,34 @@ void executorIsShutDownWithTheConnection() throws Exception { await().atMost(30, TimeUnit.SECONDS).until(() -> executorThreads().isEmpty()); } + /** + * A connection that cannot connect waits for its next reconnect on a thread of its own. The thread ends when the + * last endpoint releases the connection, without waiting until that reconnect is due. + */ + @Test + void reconnectThreadEndsWithTheConnection() throws Exception { + context.getRouteController().startRoute("unreachable"); + await().atMost(30, TimeUnit.SECONDS).until(() -> !threads(" - MiloClientReconnect").isEmpty()); + + // the route is the only user of the connection + context.getRouteController().stopRoute("unreachable"); + + // the reconnect is due 10 seconds after the failed connect + await().atMost(5, TimeUnit.SECONDS).until(() -> threads(" - MiloClientReconnect").isEmpty()); + } + private void write(final String value) { template.sendBodyAndHeader("direct:write", new Variant(value), MiloConstants.HEADER_AWAIT, true); } private List executorThreads() { + return threads(" - MiloClient"); + } + + private List threads(final String suffix) { final String prefix = "Camel (" + context.getName() + ") thread #"; return Thread.getAllStackTraces().keySet().stream() - .filter(thread -> thread.getName().startsWith(prefix) && thread.getName().endsWith(" - MiloClient")) + .filter(thread -> thread.getName().startsWith(prefix) && thread.getName().endsWith(suffix)) .toList(); } } diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index 08ed3d96753bf..5136676b0d7e5 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc @@ -889,13 +889,16 @@ the content of the earlier file was lost. The later entry is now added with a nu `orders (1).csv`, `orders (2).csv`, and so on. Entry names can therefore differ from the file names (or the `CamelFileName` header). Use `preserveFolderStructure=true` to keep the folders in the entry names instead. -=== camel-milo - each client connection has a thread pool of its own +=== camel-milo - each client connection has thread pools of its own A `milo-client` or `milo-browse` connection now handles the outcome of its write, method call, read and browse -requests, and its reconnects, on a thread pool of its own named `MiloClient`. Before, it used the shared scheduled -executor of Milo (`milo-shared-scheduled-executor`). A route that continues after an awaited request now runs on a -thread of this pool. A connection has up to as many of these threads as there are processors, and one with virtual -threads enabled; they end when the connection is closed. +requests on a thread pool of its own named `MiloClient`, and runs its reconnects on a single thread named +`MiloClientReconnect`. Before, it used the shared scheduled executor of Milo (`milo-shared-scheduled-executor`) for +both. A route that continues after an awaited request now runs on a thread of the `MiloClient` pool. The pool has up +to as many threads as there are processors, which end after they have been idle for the keep-alive time of the +default thread pool profile (60 seconds). With virtual threads enabled and more than one processor, it starts a +virtual thread for each task instead. The `MiloClientReconnect` thread starts with the first reconnect of the +connection, and its reconnects run one after the other. The threads of a connection end when it is closed. A `MiloClientConnection` created outside an endpoint needs a `CamelContext`, set with `setCamelContext`, before it is used. From 78da9a0709a5d4df2be2826a843c0780536fc88d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Mon, 19 Oct 2026 15:09:48 +0300 Subject: [PATCH 2/2] CAMEL-25812: camel-milo - hold the port of the unreachable endpoint in the test AvailablePortFinder.getNextAvailable() is deprecated. The test now holds the port as an AvailablePortFinder.Port extension, as AbstractMiloServerTest does for the server port, so it stays reserved while the test runs. Co-authored-by: Claude Opus 5.5 --- .../camel/component/milo/MiloClientExecutorTest.java | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/components/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientExecutorTest.java b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientExecutorTest.java index 6701000742956..8326debb6face 100644 --- a/components/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientExecutorTest.java +++ b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientExecutorTest.java @@ -31,6 +31,7 @@ import org.eclipse.milo.opcua.stack.core.Stack; import org.eclipse.milo.opcua.stack.core.types.builtin.Variant; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; import static org.apache.camel.component.milo.NodeIds.nodeValue; import static org.awaitility.Awaitility.await; @@ -47,9 +48,8 @@ public class MiloClientExecutorTest extends AbstractMiloServerTest { + nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "myitem1") + "&overrideHost=true"; // no server listens on this port - private static final String MILO_CLIENT_UNREACHABLE - = "milo-client:opc.tcp://localhost:" + AvailablePortFinder.getNextAvailable() + "?node=" - + nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "myitem1") + "&overrideHost=true"; + @RegisterExtension + AvailablePortFinder.Port unreachablePort = AvailablePortFinder.find(); @EndpointInject("mock:item") protected MockEndpoint item; @@ -61,7 +61,9 @@ protected RoutesBuilder createRouteBuilder() { public void configure() { from("milo-server:myitem1").to("mock:item"); from("direct:write").routeId("write").to(resolve(MILO_CLIENT_ITEM)); - from(MILO_CLIENT_UNREACHABLE).routeId("unreachable").autoStartup(false).to("mock:unreachable"); + from("milo-client:opc.tcp://localhost:" + unreachablePort.getPort() + "?node=" + + nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "myitem1") + "&overrideHost=true") + .routeId("unreachable").autoStartup(false).to("mock:unreachable"); } }; }