From a6043c3c0b7937dc1e199d060e4b49477cfdd600 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Sun, 18 Oct 2026 13:26:30 +0300 Subject: [PATCH 1/2] CAMEL-25713: camel-milo - give each client connection its own executor A client connection handled the outcome of its write, method call, read and browse requests, and ran its reconnect job, on the shared scheduled executor of Milo. A failed request disconnects the client, and the CloseSession of the disconnect can wait for a reconnect attempt of the channel, which Milo schedules on that same executor. The executor has as many threads as the JVM has processors, so with one processor its only thread waited for a task that only it could run: the connection was not rebuilt, its consumers stayed silent, and every call that needs the lock of the connection waited. Each connection now has a scheduled thread pool of its own from the ExecutorServiceManager of the CamelContext, with up to as many threads as the shared scheduled executor of Milo, and shuts it down when it is closed. Like the queue of that executor, its queue is not bounded, because a rejected reconnect would end the reconnects of the connection. A request that completes after the connection is closed fails with a RejectedExecutionException instead of being dropped. Co-authored-by: Claude Opus 5.5 --- .../milo/browse/MiloBrowseEndpoint.java | 4 +- .../milo/client/MiloClientConnection.java | 52 ++++++- .../milo/client/MiloClientEndpoint.java | 5 +- .../client/internal/SubscriptionManager.java | 4 +- .../milo/MiloClientExecutorTest.java | 127 ++++++++++++++++++ .../pages/camel-4x-upgrade-guide-4_24.adoc | 11 ++ 6 files changed, 193 insertions(+), 10 deletions(-) create mode 100644 components/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientExecutorTest.java diff --git a/components/camel-milo/src/main/java/org/apache/camel/component/milo/browse/MiloBrowseEndpoint.java b/components/camel-milo/src/main/java/org/apache/camel/component/milo/browse/MiloBrowseEndpoint.java index e70d789432175..7d0354d26738d 100644 --- a/components/camel-milo/src/main/java/org/apache/camel/component/milo/browse/MiloBrowseEndpoint.java +++ b/components/camel-milo/src/main/java/org/apache/camel/component/milo/browse/MiloBrowseEndpoint.java @@ -150,7 +150,9 @@ public Consumer createConsumer(final Processor processor) throws Exception { } public MiloClientConnection createConnection() { - return this.connectionManager.createConnection(configuration, null); + final MiloClientConnection connection = this.connectionManager.createConnection(configuration, null); + connection.setCamelContext(getCamelContext()); + return connection; } public void releaseConnection(MiloClientConnection connection) { 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 73fbe7ae6a20c..71c427c27fa42 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,14 +20,19 @@ import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import java.util.function.Consumer; +import org.apache.camel.CamelContext; +import org.apache.camel.CamelContextAware; import org.apache.camel.RuntimeCamelException; import org.apache.camel.component.milo.client.internal.SubscriptionManager; +import org.apache.camel.spi.ThreadPoolProfile; +import org.apache.camel.util.ObjectHelper; +import org.apache.camel.util.concurrent.ThreadPoolRejectedPolicy; import org.eclipse.milo.opcua.sdk.client.OpcUaClient; -import org.eclipse.milo.opcua.stack.core.Stack; import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue; import org.eclipse.milo.opcua.stack.core.types.builtin.ExpandedNodeId; import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode; @@ -39,10 +44,12 @@ import static java.util.Objects.requireNonNull; -public class MiloClientConnection implements AutoCloseable { +public class MiloClientConnection implements AutoCloseable, CamelContextAware { private final Lock lock = new ReentrantLock(); private final MiloClientConfiguration configuration; + private volatile CamelContext camelContext; + private ScheduledExecutorService executor; private SubscriptionManager manager; private volatile boolean initialized; @@ -57,6 +64,16 @@ public MiloClientConfiguration getConfiguration() { return configuration; } + @Override + public CamelContext getCamelContext() { + return camelContext; + } + + @Override + public void setCamelContext(final CamelContext camelContext) { + this.camelContext = camelContext; + } + /** * Returns the underlying Eclipse Milo {@link OpcUaClient} backing this connection. *

@@ -81,14 +98,37 @@ public OpcUaClient getOpcUaClient() { } protected void init() { - this.manager = new SubscriptionManager(this.configuration, Stack.sharedScheduledExecutor(), 10_000); + 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. + 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); + // 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.manager = new SubscriptionManager(this.configuration, this.executor, 10_000); } @Override public void close() throws Exception { - if (this.manager != null) { - this.manager.dispose(); - this.manager = null; + lock.lock(); + try { + if (this.manager != null) { + this.manager.dispose(); + this.manager = null; + } + } finally { + if (this.executor != null) { + this.camelContext.getExecutorServiceManager().shutdown(this.executor); + this.executor = null; + } + lock.unlock(); } } diff --git a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientEndpoint.java b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientEndpoint.java index afc7cc34ec2fe..254b612f2f6b6 100644 --- a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientEndpoint.java +++ b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientEndpoint.java @@ -145,7 +145,10 @@ public Consumer createConsumer(final Processor processor) throws Exception { } public MiloClientConnection createConnection() { - return this.connectionManager.createConnection(configuration, monitorFilterConfiguration); + final MiloClientConnection connection + = this.connectionManager.createConnection(configuration, monitorFilterConfiguration); + connection.setCamelContext(getCamelContext()); + return connection; } public void releaseConnection(MiloClientConnection connection) { 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 e8dab0b933369..314a55b4625db 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 @@ -834,8 +834,8 @@ private void recoverSubscription(final OpcUaSubscription subscription, final Sub * The executor that the client of a subscription delivers its values on, which handles the outcome of a check or a * delete. Creating the subscription again and rebuilding the connection block until the server answers, and a * request sent while the connection is lost waits for the channel until a reconnect attempt restores it or fails. - * The client schedules those attempts on the shared scheduled executor of Milo, which {@link #executor} is as well - * and which has a single thread on a machine with one processor, so the wait must not hold one of its threads. + * The client schedules those attempts on the shared scheduled executor of Milo, which has a single thread on a + * machine with one processor, so the wait must not hold one of its threads. *

* Milo asks that blocking calls are not made on the threads of this executor either, because they complete the * responses. The executor of the transport is not set here, so this is the shared executor of Milo, a cached pool 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 new file mode 100644 index 0000000000000..a4be2a00135be --- /dev/null +++ b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientExecutorTest.java @@ -0,0 +1,127 @@ +/* + * 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.milo; + +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +import org.apache.camel.EndpointInject; +import org.apache.camel.RoutesBuilder; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.milo.server.MiloServerComponent; +import org.apache.camel.component.mock.MockEndpoint; +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 static org.apache.camel.component.milo.NodeIds.nodeValue; +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests the executor that a client connection handles the outcome of its requests on. + */ +public class MiloClientExecutorTest extends AbstractMiloServerTest { + + private static final String MILO_CLIENT_ITEM + = "milo-client:opc.tcp://foo:bar@localhost:@@port@@?node=" + + nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "myitem1") + "&overrideHost=true"; + + @EndpointInject("mock:item") + protected MockEndpoint item; + + @Override + protected RoutesBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("milo-server:myitem1").to("mock:item"); + from("direct:write").routeId("write").to(resolve(MILO_CLIENT_ITEM)); + } + }; + } + + /** + * A failed request disconnects the client, and the disconnect can wait for a reconnect attempt that the client + * schedules on the shared scheduled executor of Milo, which has a single thread on a machine with one processor. + * The outcome of a request is therefore not handled on that executor: an awaited write completes while all of its + * threads are busy. + */ + @Test + void awaitedWriteCompletesWhileTheSharedScheduledExecutorOfMiloIsBusy() throws Exception { + item.expectedMessageCount(2); + testBody(item.message(0), assertGoodValue("Foo1")); + testBody(item.message(1), assertGoodValue("Foo2")); + + // connects + write("Foo1"); + + final ScheduledThreadPoolExecutor scheduler = (ScheduledThreadPoolExecutor) Stack.sharedScheduledExecutor(); + final CountDownLatch busy = new CountDownLatch(scheduler.getCorePoolSize()); + final CountDownLatch release = new CountDownLatch(1); + for (int i = 0; i < scheduler.getCorePoolSize(); i++) { + scheduler.execute(() -> { + busy.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + } + try { + assertTrue(busy.await(30, TimeUnit.SECONDS), "All the threads of the executor should be busy"); + + final CompletableFuture written = template.asyncRequestBodyAndHeader( + "direct:write", new Variant("Foo2"), MiloConstants.HEADER_AWAIT, true); + written.get(30, TimeUnit.SECONDS); + } finally { + release.countDown(); + } + + MockEndpoint.assertIsSatisfied(context); + } + + /** + * The executor of a connection is shut down when the last endpoint releases the connection. + */ + @Test + void executorIsShutDownWithTheConnection() throws Exception { + write("Foo1"); + assertFalse(executorThreads().isEmpty(), "The write should have been handled on the executor"); + + // the route is the only user of the connection + context.getRouteController().stopRoute("write"); + + await().atMost(30, TimeUnit.SECONDS).until(() -> executorThreads().isEmpty()); + } + + private void write(final String value) { + template.sendBodyAndHeader("direct:write", new Variant(value), MiloConstants.HEADER_AWAIT, true); + } + + private List executorThreads() { + final String prefix = "Camel (" + context.getName() + ") thread #"; + return Thread.getAllStackTraces().keySet().stream() + .filter(thread -> thread.getName().startsWith(prefix) && thread.getName().endsWith(" - MiloClient")) + .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 5b1e7e99e39d9..f495884d3c2bc 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 @@ -882,6 +882,17 @@ 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 + +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. + +A `MiloClientConnection` created outside an endpoint needs a `CamelContext`, set with `setCamelContext`, before it is +used. + === camel-lra - only an LRA hosted by the configured coordinator is joined The LRA saga service now joins, completes or compensates an LRA only when its URL is under the configured LRA From b0656303766b40c9fabbece315fb57e5926e3a0c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Sun, 18 Oct 2026 14:25:51 +0300 Subject: [PATCH 2/2] CAMEL-25713: camel-milo - release the connection lock even if its pool fails to shut down close() released the lock of the connection at the end of the finally block that shuts down its thread pool, so an exception from the shutdown left the lock held. The lock now has a finally block of its own, and the pool is still shut down when disposing the connection fails. Co-authored-by: Claude Opus 5.5 --- .../milo/client/MiloClientConnection.java | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 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 71c427c27fa42..02226f6aa2ba2 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 @@ -119,15 +119,18 @@ protected void init() { public void close() throws Exception { lock.lock(); try { - if (this.manager != null) { - this.manager.dispose(); - this.manager = null; + try { + if (this.manager != null) { + this.manager.dispose(); + this.manager = null; + } + } finally { + if (this.executor != null) { + this.camelContext.getExecutorServiceManager().shutdown(this.executor); + this.executor = null; + } } } finally { - if (this.executor != null) { - this.camelContext.getExecutorServiceManager().shutdown(this.executor); - this.executor = null; - } lock.unlock(); } }