CAMEL-25812: camel-milo - end idle client threads and run reconnects on a thread of their own

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 {

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<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 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();

components/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientExecutorTest.java

@@ -27,9 +27,11 @@
 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 @@ 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: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 @@ 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 +124,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<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();
     }
 }

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.