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 3d6b26642f12 CAMEL-25812: camel-milo - end idle client threads and run
reconnects on a thread of their own (#28111)
3d6b26642f12 is described below
commit 3d6b26642f12aea48a65bbb49fa50ff350257240
Author: Serdar Gökay <[email protected]>
AuthorDate: Mon Oct 19 17:28:33 2026 +0300
CAMEL-25812: camel-milo - end idle client threads and run reconnects on a
thread of their own (#28111)
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../milo/client/MiloClientConnection.java | 44 ++++++++++++++++------
.../milo/client/internal/SubscriptionManager.java | 18 ++++++---
.../component/milo/MiloClientExecutorTest.java | 34 ++++++++++++++++-
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 13 ++++---
4 files changed, 85 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 02226f6aa2ba..f0837e2fe1aa 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.Collections;
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.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.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 @@ public class MiloClientConnection implements
AutoCloseable, CamelContextAware {
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 class MiloClientConnection implements
AutoCloseable, CamelContextAware {
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 314a55b4625d..d2a6efcbd417 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 @@ public class SubscriptionManager {
}
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 @@ public class SubscriptionManager {
private Future<?> reconnectJob;
private final Map<UInteger, Subscription> 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 class SubscriptionManager {
}
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 @@ public class SubscriptionManager {
}
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 a4be2a00135b..8326debb6fac 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,9 +27,11 @@ 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.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;
+import org.junit.jupiter.api.extension.RegisterExtension;
import static org.apache.camel.component.milo.NodeIds.nodeValue;
import static org.awaitility.Awaitility.await;
@@ -45,6 +47,10 @@ 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
+ @RegisterExtension
+ AvailablePortFinder.Port unreachablePort = AvailablePortFinder.find();
+
@EndpointInject("mock:item")
protected MockEndpoint item;
@@ -55,6 +61,9 @@ public class MiloClientExecutorTest extends
AbstractMiloServerTest {
public void configure() {
from("milo-server:myitem1").to("mock:item");
from("direct:write").routeId("write").to(resolve(MILO_CLIENT_ITEM));
+ from("milo-client:opc.tcp://localhost:" +
unreachablePort.getPort() + "?node="
+ + nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI,
"myitem1") + "&overrideHost=true")
+
.routeId("unreachable").autoStartup(false).to("mock:unreachable");
}
};
}
@@ -101,7 +110,8 @@ public class MiloClientExecutorTest extends
AbstractMiloServerTest {
}
/**
- * 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 +124,34 @@ public class MiloClientExecutorTest extends
AbstractMiloServerTest {
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<Thread> executorThreads() {
+ return threads(" - MiloClient");
+ }
+
+ private List<Thread> 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 e4d72ed5a57c..7abafb4d7275 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
@@ -923,13 +923,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.
