CAMEL-25713: camel-milo - give each client connection its own executor

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) {

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.
      * <p>
@@ -81,14 +98,40 @@ 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 {
+            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 {
+            lock.unlock();
         }
     }
 

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) {

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.
      * <p>
      * 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

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<Object> 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<Thread> executorThreads() {
+        final String prefix = "Camel (" + context.getName() + ") thread #";
+        return Thread.getAllStackTraces().keySet().stream()
+                .filter(thread -> thread.getName().startsWith(prefix) && thread.getName().endsWith(" - MiloClient"))
+                .toList();
+    }
+}

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