From fbd934e2798d0ba4e31f51c22c3748c700328680 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Fri, 23 Oct 2026 21:27:49 +0300 Subject: [PATCH 1/3] CAMEL-25912: camel-milo - client producers no longer wait for the manager lock or rebuild the connection Problem: SubscriptionManager used one lock for the subscription work and the producer requests. Registering and deleting monitored items, recreating a subscription and the periodic retry of refused nodes hold the lock across a blocking OPC UA service call. write, call, readValues, browse and getOpcUaClient() took the same lock, although under it they only read the current connection and start an asynchronous request. A producer therefore waited for unrelated subscription work on the same connection: one round trip normally, about 10 s while the link was dropped silently (until the keep-alive declared the session lost), up to requestTimeout when the server answered keep-alives but not the call. Even without the lock, the outcome of a producer request waited for the refused-node retry: the retry ran its blocking CreateMonitoredItems call on the MiloClient executor, which also completes the producer requests and has a single thread on a machine with one processor. On a real server that held CreateMonitoredItems for 30 s and 70 s with --cpus=1, writes it answered in 10-25 ms completed only after 29.0 s and 59.1 s, when the retry returned. Cause: the lock guarded the read of Connected.client, which dispose() nulled. A failed producer request also called handleConnectionFailure, which tore down whatever connection was current when the failure arrived (possibly a replacement) and held the lock across client.disconnect(). Fix: - The producers and getOpcUaClient() read the volatile connected without the lock. Connected.client is final and no longer nulled by dispose(), so a request on a connection disposed meanwhile fails with Milo's own UaException (Bad_SessionClosed after the disconnect, Bad_ConnectionClosed for one in flight) instead of a NullPointerException. dispose() is idempotent through listener.closed. - SubscriptionManager.dispose() nulls connected under the lock, and handleConnectionFailure() nulls it before it disconnects, so getOpcUaClient() keeps returning null while there is no usable connection, as documented. handleConnectionFailure() sets lastConnectFailure first, so a producer that finds no connection during the teardown gets that failure as the cause of its not-connected result. - A failed producer request no longer calls handleConnectionFailure: the exchange gets the failure (CAMEL-25694) and the connection is kept. A lost session is recovered by Milo, a session the server no longer has fails the subscription transfer, which rebuilds the connection, and the watchdog handles a subscription that stops publishing. The outcome is still completed on the MiloClient executor with whenCompleteAsync, so the route never continues on a Milo thread. - The refused-node retry runs on the MiloClientReconnect thread, which already ran its timer, instead of handing the attempt to the MiloClient executor. Trade-off: a stalled attempt keeps that thread busy for up to requestTimeout, and a reconnect queued behind it waits. That adds no wait: the attempt is only sent while the session is active and the connection is current, so no reconnect is pending, and a rebuild needs the lock, which the attempt holds until the server answers. A dedicated retry thread would only add a thread per connection. Nothing else occupies a MiloClient thread across a service call: the executor only completes the producer requests and delivers the loss report of a consumer that starts without a connection; the checks, deletes and resends of the watchdog and of the session recovery complete on the executor of the Milo client. - Unchanged: the not-connected result is still synchronous with lastConnectFailure as its cause; consumer-side failures in whenConnected, recreateSubscription, recoverSubscription and onTransferFailed still rebuild the connection; the subscription work keeps the lock and its ordering. Behaviour change: a failed write, call, read or browse no longer disconnects the client and rebuilds the subscriptions of the consumers that share the connection (no Bad_NoCommunication report, same session). Producer requests and getOpcUaClient() no longer wait for a consumer that starts or stops or for the refused-node retry, whose attempts now run on the MiloClientReconnect thread. While Camel rebuilds the connection, producers now fail at once with the not-connected result instead of waiting for the rebuild to finish. Documented in milo-client-component.adoc (and its catalog copy) and in the 4.24 upgrade guide; two sentences of the CAMEL-25835 note there that said a failed request rebuilds the connection are corrected. Not done (follow-ups): - handleConnectionFailure still disconnects the old client under the lock; during a silent drop its CloseSession can block for seconds. Producers and getOpcUaClient() no longer take the lock, so they no longer wait for it, and producer failures no longer reach it; only the subscription work after a consumer-side failure still does. Moving the disconnect out is not small: whenConnected (under registerItems and unregisterItems), recreateSubscription and recoverSubscription call it while they already hold the lock, so they would have to hand the old connection out of their own lock scope with an identity check, and an asynchronous disconnect would hold the single MiloClient or reconnect thread on one processor instead. - A failed producer request used to re-run endpoint discovery through the rebuild. Milo's own reconnect reuses the cached endpoint and server certificate, so a server that renews its certificate or changes its endpoints while it is unreachable is now recovered only by a Camel rebuild, for example a context restart. Consumer-only connections already behaved this way since CAMEL-25691 and CAMEL-25692. Documented in the component adoc; no mechanism in this commit. - onTransferFailed still tears down whichever connection is current, without checking that it is the one whose transfer failed. Pre-existing, the same class of defect as the producer teardown removed here. Tests: - ProducerOnSharedConnectionTest (new): - producerDoesNotWaitForAConsumerThatStartsOnASilentLink: the relay goes silent, a consumer starts; once its thread waits in putSubscriptions for its CreateMonitoredItems call under the lock, getOpcUaClient() and a fire-and-forget write return within 2 s while the start is still waiting. On the unmodified code it fails: "execution timed out after 2000 ms", the getOpcUaClient() thread parked in ReentrantLock.lock from SubscriptionManager.getOpcUaClient. - writeThatTimesOutKeepsTheConnection: the server holds a write past requestTimeout=1000; the exchange fails with UaException(Bad_Timeout), the consumer gets no Bad_NoCommunication and only the next value, same client and session. On the unmodified code it fails: the consumer received the Bad_NoCommunication report of the rebuild ("expected: but was: "). - writeOnAReplacedConnectionFailsAndKeepsItsReplacement: a held write is in flight while a consumer-side failure replaces the connection; the exchange fails with UaException(Bad_ConnectionClosed) and the replacement is kept (same client and session, next value only). Passes on the unmodified code too: there the write fails during the dispose, before the replacement exists. - writeDoesNotWaitForTheAttemptToCreateARefusedNodeAgain: a manager built with a single-thread executor, as with one processor; the server refuses a node, then holds the next attempt to create it (its attribute read). A write must complete within 2 s while the attempt waits. It fails with "execution timed out after 2000 ms" on the unmodified code (the write parked in ReentrantLock.lock from SubscriptionManager.write) and with only the lock change, the retry still on the MiloClient executor (the write waiting in CompletableFuture.get for its outcome). - TcpRelay gains silence(): the current connections drop their data without closing, until dropped. - No existing assertion pinned the producer teardown; only prose did and is corrected without changing an assertion: MiloClientExecutorTest (javadoc said a failed request disconnects the client), MiloClientProducerWriteFailureTest (class javadoc: "while still triggering the reconnection side-effect"), SubscriptionManagerExceptionPropagationTest (comment naming handleConnectionFailure as the side-effect), MiloClientConnection.init() comments. - CommunicationLossTest.lostSessionReportsTheNodeAndResendsItsValue fails under host load with a duplicate resent value on the unmodified code as well (2 of 5 runs with -Dsurefire.rerunFailingTestsCount=0); unrelated. Co-authored-by: Claude Opus 5.5 --- .../catalog/docs/milo-client-component.adoc | 6 +- .../src/main/docs/milo-client-component.adoc | 6 +- .../milo/client/MiloClientConnection.java | 15 +- .../client/internal/SubscriptionManager.java | 173 ++++--- .../milo/MiloClientExecutorTest.java | 7 +- .../milo/ProducerOnSharedConnectionTest.java | 433 ++++++++++++++++++ .../apache/camel/component/milo/TcpRelay.java | 28 +- .../MiloClientProducerWriteFailureTest.java | 2 +- ...iptionManagerExceptionPropagationTest.java | 4 +- .../pages/camel-4x-upgrade-guide-4_24.adoc | 22 +- 10 files changed, 577 insertions(+), 119 deletions(-) create mode 100644 components/camel-milo/src/test/java/org/apache/camel/component/milo/ProducerOnSharedConnectionTest.java diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/milo-client-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/milo-client-component.adoc index e12de6d7a3b75..3acad3bb65d1a 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/milo-client-component.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/milo-client-component.adoc @@ -239,7 +239,11 @@ or detects changes must tolerate this. A value that did not change during the lo Routes run while communication is lost. An OPC UA request that a route makes while the client reconnects, for example a write, waits for the next reconnect attempt and is sent if that attempt succeeds, so it can take effect after -communication is restored. A request that fails rebuilds the connection. +communication is restored. A request that fails completes with its failure; the connection is kept, and a lost +session is recovered by the client as described above. The client recovers it with the endpoint and the server +certificate it found when it connected, so a server that changes either while it cannot be reached is reached again +only when Camel rebuilds the connection, for example when the Camel context restarts. A request does not wait while a +consumer of the same connection creates or deletes its monitored items. Camel keeps no last value. A route that needs one keeps the last good value of each node itself. To recognise the status, check the status code of the `DataValue` body, for example with `${body.statusCode.bad}`. diff --git a/components/camel-milo/src/main/docs/milo-client-component.adoc b/components/camel-milo/src/main/docs/milo-client-component.adoc index e12de6d7a3b75..3acad3bb65d1a 100644 --- a/components/camel-milo/src/main/docs/milo-client-component.adoc +++ b/components/camel-milo/src/main/docs/milo-client-component.adoc @@ -239,7 +239,11 @@ or detects changes must tolerate this. A value that did not change during the lo Routes run while communication is lost. An OPC UA request that a route makes while the client reconnects, for example a write, waits for the next reconnect attempt and is sent if that attempt succeeds, so it can take effect after -communication is restored. A request that fails rebuilds the connection. +communication is restored. A request that fails completes with its failure; the connection is kept, and a lost +session is recovered by the client as described above. The client recovers it with the endpoint and the server +certificate it found when it connected, so a server that changes either while it cannot be reached is reached again +only when Camel rebuilds the connection, for example when the Camel context restarts. A request does not wait while a +consumer of the same connection creates or deletes its monitored items. Camel keeps no last value. A route that needs one keeps the last good value of each node itself. To recognise the status, check the status code of the `DataValue` body, for example with `${body.statusCode.bad}`. 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 3b276156dd81c..0855eb458265b 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 @@ -138,15 +138,14 @@ MonitoringState getMonitoringState() { 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 executors of its own. + // The outcome of a request continues its route, and a reconnect waits for the server; neither may hold a thread of + // Milo, whose shared scheduled executor runs the reconnect attempts of the clients. 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, 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. + // processor, a virtual thread for each task. 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); @@ -156,8 +155,8 @@ protected void init() { // a request that completes after the connection is closed fails instead of being dropped profile.setRejectedPolicy(ThreadPoolRejectedPolicy.Abort); 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 + // the reconnects of the connection and its attempts to create the refused monitored items again 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( 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 882d185c10693..78a2c8ba4b7ac 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 @@ -343,7 +343,12 @@ void deliver() { } private class Connected { - private OpcUaClient client; + /** + * The client, which is kept after the connection is disposed: a producer reads the connection without + * {@code lock}, so it can still send a request on a connection that is disposed, which the client then fails + * with the status of its closed session + */ + private final OpcUaClient client; private final OpcUaSubscription manager; private final SubscriptionListenerImpl listener; private final long generation; @@ -517,16 +522,18 @@ private void refuse( /** * Schedules the next attempt to create the refused nodes again, unless there is none or one is scheduled. The - * timer runs on the reconnect thread, the attempt, which blocks until the server answers, on the executor of - * the connection, so that a slow attempt does not hold up a reconnect. + * attempt blocks until the server answers, so it runs on the reconnect thread, not on the executor of the + * connection, whose threads complete the requests of the producers: with one processor that executor has a + * single thread, and the outcome of every request would wait for the attempt. A rebuild of the connection does + * not wait for the reconnect thread longer than it waits for the attempt anyway: it needs {@code lock}, which + * the attempt holds until the server answers. */ private void scheduleRetry() { if (this.refused.isEmpty() || this.retryJob != null || SubscriptionManager.this.disposed) { return; } this.retryJob = SubscriptionManager.this.reconnectExecutor.schedule( - () -> SubscriptionManager.this.executor.execute(() -> retryRefused(this)), - SubscriptionManager.this.reconnectTimeout, TimeUnit.MILLISECONDS); + () -> retryRefused(this), SubscriptionManager.this.reconnectTimeout, TimeUnit.MILLISECONDS); } /** @@ -623,16 +630,15 @@ private CompletableFuture lookupNamespaceIndex(final String namespaceUri } public void dispose() { + if (this.listener.closed) { + return; + } // the session closed by the disconnect is no loss of communication this.listener.closed = true; - if (this.client != null) { - try { - this.client.disconnect(); - this.client = null; - } catch (final UaException e) { - LOG.warn("Failed to disconnect client", e); - } - + try { + this.client.disconnect(); + } catch (final UaException e) { + LOG.warn("Failed to disconnect client", e); } } @@ -972,7 +978,8 @@ nodeId, direction, NodeIds.References, includeSubTypes, uint(nodeClasses), private final Lock lock = new ReentrantLock(); /** - * The current connection, written under {@code lock} and read by the monitoring state without it + * The current connection, written under {@code lock} and read by the producers, {@link #getOpcUaClient()} and the + * monitoring state without it */ private volatile Connected connected; private volatile Throwable lastConnectFailure; @@ -1003,12 +1010,9 @@ public SubscriptionManager(final MiloClientConfiguration configuration, final Sc * @return the active client, or {@code null} if there is no current connection */ public OpcUaClient getOpcUaClient() { - lock.lock(); - try { - return this.connected != null ? this.connected.client : null; - } finally { - lock.unlock(); - } + // read without lock, which a service call holds until the server answers + final Connected connected = this.connected; + return connected != null ? connected.client : null; } /** @@ -1079,8 +1083,8 @@ private void checkSubscription(final OpcUaSubscription subscription, final Subsc * subscriptions. *

* The delete neither blocks nor holds {@code lock}: a request sent before the connection was lost only fails once - * the client notices the loss, and holding {@code lock} until then would hold up the producers and - * {@link #getOpcUaClient()} as well. + * the client notices the loss, and holding {@code lock} until then would hold up the consumers that start or stop + * as well. */ private void recoverSubscription(final OpcUaSubscription subscription, final SubscriptionListenerImpl listener) { subscription.deleteAsync().whenCompleteAsync((unit, e) -> { @@ -1156,10 +1160,11 @@ private void recreateSubscription(final OpcUaSubscription subscription) { * failure is the status of the nodes until the next attempt. *

* Like the registration of a consumer, the attempt holds {@code lock} until the server answers, so that a consumer - * that stops or a subscription that is recreated in the meantime cannot interleave with it, and the producers and - * {@link #getOpcUaClient()} wait for it. The wait is bounded: the attempt is only sent while the session is active, - * a call in flight when the connection drops silently fails once the keep-alive of the client declares the session - * lost, and a call that the server does not answer fails after the request timeout. + * that stops or a subscription that is recreated in the meantime cannot interleave with it, and a consumer that + * starts or stops waits for it; the producers and {@link #getOpcUaClient()} do not. The wait is bounded: the + * attempt is only sent while the session is active, a call in flight when the connection drops silently fails once + * the keep-alive of the client declares the session lost, and a call that the server does not answer fails after + * the request timeout. */ private void retryRefused(final Connected connected) { lock.lock(); @@ -1270,14 +1275,17 @@ private void deliverLoss(final Map nodes, final long gen private void handleConnectionFailure(final Throwable e) { lock.lock(); try { - if (this.connected != null) { - // the nodes lose their communication with the connection, after the values it still delivers - reportLoss(this.connected.manager, this.connected.generation, this.subscriptions); - this.connected.cancelRetry(); - this.connected.dispose(); + // set first: a producer that finds no connection while it is torn down fails with this cause + this.lastConnectFailure = e; + final Connected failed = this.connected; + if (failed != null) { + // the producers and getOpcUaClient() no longer reach the connection that is torn down this.connected = null; + // the nodes lose their communication with the connection, after the values it still delivers + reportLoss(failed.manager, failed.generation, this.subscriptions); + failed.cancelRetry(); + failed.dispose(); } - this.lastConnectFailure = e; // log @@ -1498,6 +1506,8 @@ public void dispose() { } this.disposed = true; connected = this.connected; + // the producers and getOpcUaClient() no longer reach the connection that is disposed + this.connected = null; if (connected != null) { // the next attempt is cancelled under the lock that guards it, the connection is disposed outside of it connected.cancelRetry(); @@ -1702,82 +1712,57 @@ public void unregisterItems(final List clientHandles) { public CompletableFuture call( final ExpandedNodeId nodeId, final ExpandedNodeId methodId, final Variant[] inputArguments) { - lock.lock(); - try { - if (this.connected == null) { - return newNotConnectedResult(); - } - - return this.connected.call(nodeId, methodId, inputArguments).whenCompleteAsync((status, e) -> { - // handle outside the lock, running using - // whenCompleteAsync (preserves the original exceptional completion) - if (e != null) { - handleConnectionFailure(e); - } - }, this.executor); - } finally { - lock.unlock(); + final Connected connected = this.connected; + if (connected == null) { + return newNotConnectedResult(); } + + return outcomeOf(connected.call(nodeId, methodId, inputArguments)); } public CompletableFuture write(final ExpandedNodeId nodeId, final DataValue value) { - lock.lock(); - try { - if (this.connected == null) { - return newNotConnectedResult(); - } - - return this.connected.write(nodeId, value).whenCompleteAsync((status, e) -> { - // handle outside the lock, running using - // whenCompleteAsync (preserves the original exceptional completion) - if (e != null) { - handleConnectionFailure(e); - } - }, this.executor); - } finally { - lock.unlock(); + final Connected connected = this.connected; + if (connected == null) { + return newNotConnectedResult(); } + + return outcomeOf(connected.write(nodeId, value)); } public CompletableFuture readValues(final List nodeIds) { - lock.lock(); - try { - if (this.connected == null) { - return newNotConnectedResult(); - } - - return this.connected.readValues(nodeIds).whenCompleteAsync((nodes, e) -> { - // handle outside the lock, running using - // whenCompleteAsync (preserves the original exceptional completion) - if (e != null) { - handleConnectionFailure(e); - } - }, this.executor); - } finally { - lock.unlock(); + final Connected connected = this.connected; + if (connected == null) { + return newNotConnectedResult(); } + + return outcomeOf(connected.readValues(nodeIds)); } public CompletableFuture> browse( List expandedNodeIds, BrowseDirection direction, int nodeClasses, int maxDepth, String filter, boolean includeSubTypes, int maxNodesPerRequest) { - lock.lock(); - try { - if (this.connected == null) { - return newNotConnectedResult(); - } - - return this.connected.browse(expandedNodeIds, direction, nodeClasses, 1, maxDepth, - null != filter ? Pattern.compile(filter) : null, includeSubTypes, maxNodesPerRequest) - .whenCompleteAsync((browseResult, e) -> { - // handle outside the lock, running using - // whenCompleteAsync (preserves the original exceptional completion) - if (e != null) { - handleConnectionFailure(e); - } - }, this.executor); - } finally { - lock.unlock(); + final Connected connected = this.connected; + if (connected == null) { + return newNotConnectedResult(); } + + return outcomeOf(connected.browse(expandedNodeIds, direction, nodeClasses, 1, maxDepth, + null != filter ? Pattern.compile(filter) : null, includeSubTypes, maxNodesPerRequest)); + } + + /** + * The outcome of a producer request, completed on the executor of the connection, so that the route continues there + * and not on a thread of Milo, which completes the responses; unlike thenApplyAsync, whenCompleteAsync hands a + * failure to the executor too. A request that fails completes with its failure and does not rebuild the connection: + * the client recovers a lost session itself, and a session that the server no longer has fails the transfer of the + * subscription, which rebuilds the connection. + *

+ * The producers read the current connection without {@code lock}, which a service call of the consumers holds until + * the server answers, so they do not wait for it. A request on a connection that is disposed meanwhile fails with + * the status of its closed session. + */ + private CompletableFuture outcomeOf(final CompletableFuture request) { + return request.whenCompleteAsync((result, e) -> { + }, this.executor); } } 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 8326debb6face..bb52813c91824 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 @@ -69,10 +69,9 @@ public void configure() { } /** - * 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. + * An awaited write completes while every thread of the shared scheduled executor of Milo is busy: neither the + * request nor the handling of its outcome needs a thread of that executor, which has a single thread on a machine + * with one processor. */ @Test void awaitedWriteCompletesWhileTheSharedScheduledExecutorOfMiloIsBusy() throws Exception { diff --git a/components/camel-milo/src/test/java/org/apache/camel/component/milo/ProducerOnSharedConnectionTest.java b/components/camel-milo/src/test/java/org/apache/camel/component/milo/ProducerOnSharedConnectionTest.java new file mode 100644 index 0000000000000..dedb7cfcc70ad --- /dev/null +++ b/components/camel-milo/src/test/java/org/apache/camel/component/milo/ProducerOnSharedConnectionTest.java @@ -0,0 +1,433 @@ +/* + * 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.time.Duration; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.camel.EndpointInject; +import org.apache.camel.Exchange; +import org.apache.camel.RoutesBuilder; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.milo.client.MiloClientConnection; +import org.apache.camel.component.milo.client.MiloClientEndpoint; +import org.apache.camel.component.milo.client.MonitorFilterConfiguration; +import org.apache.camel.component.milo.client.internal.SubscriptionManager; +import org.apache.camel.component.milo.server.MiloServerComponent; +import org.apache.camel.component.milo.server.internal.CamelNamespace; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.test.AvailablePortFinder; +import org.apache.camel.util.ObjectHelper; +import org.eclipse.milo.opcua.sdk.client.OpcUaClient; +import org.eclipse.milo.opcua.sdk.server.AddressSpace; +import org.eclipse.milo.opcua.stack.core.StatusCodes; +import org.eclipse.milo.opcua.stack.core.UaException; +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.NodeId; +import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode; +import org.eclipse.milo.opcua.stack.core.types.builtin.Variant; +import org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.UInteger; +import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn; +import org.eclipse.milo.opcua.stack.core.types.structured.ReadValueId; +import org.eclipse.milo.opcua.stack.core.types.structured.WriteValue; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.extension.RegisterExtension; + +import static org.apache.camel.component.milo.NodeIds.nodeValue; +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * The producers of a connection that consumers share do not wait while a consumer creates its monitored items, and a + * request that fails sets its failure on the exchange without rebuilding the connection: the consumers keep their + * monitored items and their session. + */ +public class ProducerOnSharedConnectionTest extends AbstractMiloServerTest { + + // a namespace that the server does not have until a test adds it + private static final String HELD_NAMESPACE_URI = "urn:org:apache:camel:held"; + + private static final String HELD_NODE = "nsu=" + HELD_NAMESPACE_URI + ";s=heldItem"; + + // a namespace whose node the server does not have, so it refuses to monitor it + private static final String REFUSED_NAMESPACE_URI = "urn:org:apache:camel:refused"; + + private static final String REFUSED_NODE = "refusedItem"; + + // the client endpoints only differ in their nodes, so they share one connection + private static final String MILO_CLIENT_BASE = "milo-client:opc.tcp://foo:bar@localhost:@@port@@" + + "?allowedSecurityPolicies=None&overrideHost=true" + + "&samplingInterval=50&requestedPublishingInterval=100"; + + private static final String ITEM_1 = "&node=" + nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "myitem1"); + + private static final String ITEM_2 = "&node=" + nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "myitem2"); + + private static final String HELD = "&node=RAW(" + HELD_NODE + ")"; + + // a connection of its own, whose requests time out after a second + private static final String MILO_CLIENT_TIMEOUT_BASE = MILO_CLIENT_BASE + "&clientId=timeout&requestTimeout=1000"; + + // a connection of its own, through the relay: the client uses the port of the relay, not the one of the server + private static final String MILO_CLIENT_RELAYED_BASE = MILO_CLIENT_BASE + "&overridePort=true"; + + @RegisterExtension + AvailablePortFinder.Port relayPortHolder = AvailablePortFinder.find(); + + private final TcpRelay relay = new TcpRelay(); + + @EndpointInject("mock:item1") + protected MockEndpoint item1Endpoint; + + @EndpointInject("mock:item2") + protected MockEndpoint item2Endpoint; + + @EndpointInject("mock:timeoutItem1") + protected MockEndpoint timeoutItem1Endpoint; + + @EndpointInject("mock:relayedItem1") + protected MockEndpoint relayedItem1Endpoint; + + @Override + protected RoutesBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + // the consumers deliver the current values while they start, so the tests start these routes + from(resolve(MILO_CLIENT_BASE + ITEM_1)).routeId("item1").autoStartup(false).to("mock:item1"); + from(resolve(MILO_CLIENT_BASE + ITEM_2)).routeId("item2").autoStartup(false).to("mock:item2"); + from(resolve(MILO_CLIENT_TIMEOUT_BASE + ITEM_1)).routeId("timeoutItem1").autoStartup(false) + .to("mock:timeoutItem1"); + from(relayed(MILO_CLIENT_RELAYED_BASE + ITEM_1)).routeId("relayedItem1").autoStartup(false) + .to("mock:relayedItem1"); + from(relayed(MILO_CLIENT_RELAYED_BASE + ITEM_2)).routeId("relayedItem2").autoStartup(false) + .to("mock:relayedItem2"); + + // a write that the route does not wait for + from("direct:relayedWrite").to(relayed(MILO_CLIENT_RELAYED_BASE + ITEM_1)); + } + }; + } + + private String relayed(final String uri) { + return uri.replace("@@port@@", Integer.toString(relayPortHolder.getPort())); + } + + @Override + protected void doPostSetup() throws Exception { + relay.start(relayPortHolder.getPort(), getServerPort()); + } + + @Override + protected void doPostTearDown() throws Exception { + // after the context, so that its client disconnects through the relay + relay.stop(); + } + + /** + * A consumer that starts while the link to the server is silent waits for the server until the keep-alive of the + * client declares the session lost. A producer and {@link MiloClientConnection#getOpcUaClient()} of the same + * connection do not wait for it. + */ + @Test + @Timeout(value = 60, unit = TimeUnit.SECONDS) + void producerDoesNotWaitForAConsumerThatStartsOnASilentLink() throws Exception { + template.sendBody("milo-server:myitem1", "Foo"); + relayedItem1Endpoint.expectedMessageCount(1); + testBody(relayedItem1Endpoint.message(0), assertGoodValue("Foo")); + context.getRouteController().startRoute("relayedItem1"); + relayedItem1Endpoint.assertIsSatisfied(); + + final MiloClientEndpoint endpoint + = context.getEndpoint(relayed(MILO_CLIENT_RELAYED_BASE + ITEM_1), MiloClientEndpoint.class); + final MiloClientConnection connection = endpoint.createConnection(); + final CompletableFuture start = new CompletableFuture<>(); + try { + final OpcUaClient client = connection.getOpcUaClient(); + assertNotNull(client); + + // the link drops silently, then the next consumer starts: it waits for the answer to its + // CreateMonitoredItems call, which does not come + relay.silence(); + final Thread starter = new Thread(() -> { + try { + context.getRouteController().startRoute("relayedItem2"); + start.complete(null); + } catch (final Exception e) { + start.completeExceptionally(e); + } + }, "start-relayedItem2"); + starter.start(); + // the consumer holds the lock of the manager while it waits for the server + await().atMost(10, TimeUnit.SECONDS).until(() -> waitsInPutSubscriptions(starter)); + + assertSame(client, assertTimeoutPreemptively(Duration.ofSeconds(2), connection::getOpcUaClient)); + assertTimeoutPreemptively(Duration.ofSeconds(2), () -> template.sendBody("direct:relayedWrite", "Bar")); + // the consumer was still waiting for the server + assertFalse(start.isDone()); + } finally { + // the connections are dropped, which fails the call of the consumer + relay.dropConnections(); + start.handle((result, e) -> null).get(30, TimeUnit.SECONDS); + endpoint.releaseConnection(connection); + } + } + + private static boolean waitsInPutSubscriptions(final Thread thread) { + final Thread.State state = thread.getState(); + return (state == Thread.State.WAITING || state == Thread.State.TIMED_WAITING) + && Arrays.stream(thread.getStackTrace()) + .anyMatch(frame -> frame.getClassName().startsWith(SubscriptionManager.class.getName()) + && frame.getMethodName().equals("putSubscriptions")); + } + + /** + * A write that times out because the server does not answer it in time fails the exchange. The session is still + * active, so the connection is kept: its consumer reports no loss and keeps receiving values in the same session. + */ + @Test + @Timeout(value = 60, unit = TimeUnit.SECONDS) + void writeThatTimesOutKeepsTheConnection() throws Exception { + template.sendBody("milo-server:myitem1", "Foo"); + timeoutItem1Endpoint.expectedMessageCount(1); + testBody(timeoutItem1Endpoint.message(0), assertGoodValue("Foo")); + context.getRouteController().startRoute("timeoutItem1"); + timeoutItem1Endpoint.assertIsSatisfied(); + + final CountDownLatch writing = new CountDownLatch(1); + final CountDownLatch release = new CountDownLatch(1); + final CamelNamespace namespace = holdingNamespace(writing, release); + final MiloClientEndpoint endpoint + = context.getEndpoint(resolve(MILO_CLIENT_TIMEOUT_BASE + ITEM_1), MiloClientEndpoint.class); + final MiloClientConnection connection = endpoint.createConnection(); + try { + namespace.startup(); + namespace.getOrAddItem("heldItem").update("Foo"); + final OpcUaClient client = connection.getOpcUaClient(); + final NodeId sessionId = client.getSession().getSessionId(); + + // the consumer receives the next value only, no loss report and not its value again + timeoutItem1Endpoint.reset(); + timeoutItem1Endpoint.expectedMessageCount(1); + timeoutItem1Endpoint.setAssertPeriod(2_000); + testBody(timeoutItem1Endpoint.message(0), assertGoodValue("Bar")); + + // the server holds the write + final Exchange written = template.send(resolve(MILO_CLIENT_TIMEOUT_BASE + HELD), + exchange -> { + exchange.getMessage().setBody("Baz"); + exchange.getMessage().setHeader(MiloConstants.HEADER_AWAIT, true); + }); + assertTrue(writing.await(10, TimeUnit.SECONDS)); + assertStatus(StatusCodes.Bad_Timeout, written); + + template.sendBody("milo-server:myitem1", "Bar"); + timeoutItem1Endpoint.assertIsSatisfied(); + + assertSame(client, connection.getOpcUaClient()); + assertEquals(sessionId, client.getSession().getSessionId()); + } finally { + release.countDown(); + endpoint.releaseConnection(connection); + namespace.shutdown(); + } + } + + /** + * A write in flight on a connection that Camel replaces fails with the status of its closed connection. Its failure + * does not rebuild the connection that replaced it. + */ + @Test + @Timeout(value = 60, unit = TimeUnit.SECONDS) + void writeOnAReplacedConnectionFailsAndKeepsItsReplacement() throws Exception { + template.sendBody("milo-server:myitem1", "Foo"); + template.sendBody("milo-server:myitem2", "Foo"); + item1Endpoint.expectedMessageCount(1); + testBody(item1Endpoint.message(0), assertGoodValue("Foo")); + context.getRouteController().startRoute("item1"); + item1Endpoint.assertIsSatisfied(); + + final CountDownLatch writing = new CountDownLatch(1); + final CountDownLatch release = new CountDownLatch(1); + final CamelNamespace namespace = holdingNamespace(writing, release); + final MiloClientEndpoint endpoint = context.getEndpoint(resolve(MILO_CLIENT_BASE + ITEM_1), MiloClientEndpoint.class); + final MiloClientConnection connection = endpoint.createConnection(); + try { + namespace.startup(); + namespace.getOrAddItem("heldItem").update("Foo"); + final OpcUaClient client = connection.getOpcUaClient(); + item1Endpoint.reset(); + + // the server holds the write + final CompletableFuture written = template.asyncSend(resolve(MILO_CLIENT_BASE + HELD), + exchange -> { + exchange.getMessage().setBody("Baz"); + exchange.getMessage().setHeader(MiloConstants.HEADER_AWAIT, true); + }); + assertTrue(writing.await(10, TimeUnit.SECONDS)); + + // the server deletes the subscription, so the CreateMonitoredItems call of the next consumer fails as a + // whole, which rebuilds the connection + final UInteger subscriptionId = client.getSubscriptions().get(0).getSubscriptionId().orElseThrow(); + assertTrue(client.deleteSubscriptions(List.of(subscriptionId)).getResults()[0].isGood()); + context.getRouteController().startRoute("item2"); + + assertStatus(StatusCodes.Bad_ConnectionClosed, written.get(30, TimeUnit.SECONDS)); + final OpcUaClient replacement = await().atMost(30, TimeUnit.SECONDS) + .until(connection::getOpcUaClient, current -> current != null && current != client); + final NodeId sessionId = replacement.getSession().getSessionId(); + // the consumer receives its current value from the replacement + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> { + final List received = item1Endpoint.getReceivedExchanges(); + assertFalse(received.isEmpty()); + assertGoodValue("Foo").accept(received.get(received.size() - 1).getMessage().getBody(DataValue.class)); + }); + + // the replacement is kept: the consumer receives the next value only + item1Endpoint.reset(); + item1Endpoint.expectedMessageCount(1); + item1Endpoint.setAssertPeriod(2_000); + testBody(item1Endpoint.message(0), assertGoodValue("Bar")); + template.sendBody("milo-server:myitem1", "Bar"); + item1Endpoint.assertIsSatisfied(); + + assertSame(replacement, connection.getOpcUaClient()); + assertNotSame(client, replacement); + assertEquals(sessionId, replacement.getSession().getSessionId()); + } finally { + release.countDown(); + endpoint.releaseConnection(connection); + namespace.shutdown(); + } + } + + /** + * Asserts that an awaited request failed with the given status code of the client + */ + /** + * A connection creates a node that the server refused again while its producer sends a request. With one processor + * the executor that completes the requests has a single thread, as here: a write completes while the server holds + * the attempt, which waits for its answer. + */ + @Test + @Timeout(value = 60, unit = TimeUnit.SECONDS) + void writeDoesNotWaitForTheAttemptToCreateARefusedNodeAgain() throws Exception { + template.sendBody("milo-server:myitem1", "Foo"); + + final AtomicBoolean holding = new AtomicBoolean(); + final CountDownLatch creating = new CountDownLatch(1); + final CountDownLatch release = new CountDownLatch(1); + final CamelNamespace namespace = new CamelNamespace(REFUSED_NAMESPACE_URI, getServer().getServer()) { + @Override + public List read( + final AddressSpace.ReadContext context, final Double maxAge, final TimestampsToReturn timestamps, + final List readValueIds) { + // the server reads the attributes of a node before it creates its monitored item + if (holding.get() && readValueIds.stream() + .anyMatch(id -> REFUSED_NODE.equals(id.getNodeId().getIdentifier()))) { + creating.countDown(); + try { + release.await(30, TimeUnit.SECONDS); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + return super.read(context, maxAge, timestamps, readValueIds); + } + }; + namespace.startup(); + + final MiloClientEndpoint endpoint + = context.getEndpoint(resolve(MILO_CLIENT_BASE + ITEM_1), MiloClientEndpoint.class); + // a connection of its own that completes its requests on a single thread, as with one processor, and tries to + // create the refused node again every 500 ms + final ScheduledExecutorService reconnectExecutor = Executors.newSingleThreadScheduledExecutor(); + final ExecutorService executor = Executors.newSingleThreadExecutor(); + final SubscriptionManager manager = new SubscriptionManager( + endpoint.getConfiguration(), reconnectExecutor, executor, 500); + try { + final List refusals = new CopyOnWriteArrayList<>(); + manager.registerItems( + Map.of(ExpandedNodeId.parse("nsu=" + REFUSED_NAMESPACE_URI + ";s=" + REFUSED_NODE), refusals::add), + 50.0, null, new MonitorFilterConfiguration()); + await().atMost(10, TimeUnit.SECONDS).until(() -> !refusals.isEmpty()); + assertEquals(new StatusCode(StatusCodes.Bad_NodeIdUnknown), refusals.get(0).getStatusCode()); + + // the server holds the next attempt to create the node + holding.set(true); + assertTrue(creating.await(10, TimeUnit.SECONDS)); + + final ExpandedNodeId item = ExpandedNodeId.parse(endpoint.getNode()); + assertTimeoutPreemptively(Duration.ofSeconds(2), + () -> manager.write(item, new DataValue(new Variant("Bar"))).get()); + // the attempt was still waiting for the server + assertEquals(1, release.getCount()); + } finally { + release.countDown(); + manager.dispose(); + reconnectExecutor.shutdownNow(); + executor.shutdownNow(); + namespace.shutdown(); + } + } + + private static void assertStatus(final long statusCode, final Exchange exchange) { + final UaException failure = ObjectHelper.getException(UaException.class, exchange.getException()); + assertNotNull(failure, () -> "Expected a UaException, but the exchange failed with " + exchange.getException()); + assertEquals(new StatusCode(statusCode), failure.getStatusCode()); + } + + /** + * A namespace of the server that holds the writes to its nodes until it is released + */ + private CamelNamespace holdingNamespace(final CountDownLatch writing, final CountDownLatch release) { + return new CamelNamespace(HELD_NAMESPACE_URI, getServer().getServer()) { + @Override + public List write(final AddressSpace.WriteContext context, final List writeValues) { + writing.countDown(); + try { + release.await(30, TimeUnit.SECONDS); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + return super.write(context, writeValues); + } + }; + } + + private MiloServerComponent getServer() { + return context.getComponent("milo-server", MiloServerComponent.class); + } +} diff --git a/components/camel-milo/src/test/java/org/apache/camel/component/milo/TcpRelay.java b/components/camel-milo/src/test/java/org/apache/camel/component/milo/TcpRelay.java index b564fc117aef1..032de545559c3 100644 --- a/components/camel-milo/src/test/java/org/apache/camel/component/milo/TcpRelay.java +++ b/components/camel-milo/src/test/java/org/apache/camel/component/milo/TcpRelay.java @@ -23,10 +23,12 @@ import java.net.Socket; import java.util.ArrayList; import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; /** - * A TCP relay to the server, whose connections a test can drop while it keeps listening, and which can be down while - * the client tries to connect. + * A TCP relay to the server, whose connections a test can drop or silence while it keeps listening, and which can be + * down while the client tries to connect. */ final class TcpRelay { @@ -34,6 +36,9 @@ final class TcpRelay { private final List sockets = new ArrayList<>(); + // the sockets whose data is dropped instead of relayed + private final Set silenced = ConcurrentHashMap.newKeySet(); + void start(final int port, final int serverPort) throws IOException { final ServerSocket listener = new ServerSocket(port); this.listener = listener; @@ -61,14 +66,16 @@ void start(final int port, final int serverPort) throws IOException { acceptor.start(); } - private static void pump(final Socket from, final Socket to) { + private void pump(final Socket from, final Socket to) { final Thread thread = new Thread(() -> { final byte[] buffer = new byte[8192]; try (InputStream in = from.getInputStream(); OutputStream out = to.getOutputStream()) { int read; while ((read = in.read(buffer)) >= 0) { - out.write(buffer, 0, read); - out.flush(); + if (!silenced.contains(from)) { + out.write(buffer, 0, read); + out.flush(); + } } } catch (final IOException e) { // the connection was dropped @@ -81,10 +88,21 @@ private static void pump(final Socket from, final Socket to) { thread.start(); } + /** + * Drops what the current connections carry in both directions, without closing them: the link goes silent, as when + * the network drops it without a reset, until the connections are dropped. New connections are relayed. + */ + void silence() { + synchronized (sockets) { + silenced.addAll(sockets); + } + } + void dropConnections() { synchronized (sockets) { sockets.forEach(TcpRelay::close); sockets.clear(); + silenced.clear(); } } diff --git a/components/camel-milo/src/test/java/org/apache/camel/component/milo/client/MiloClientProducerWriteFailureTest.java b/components/camel-milo/src/test/java/org/apache/camel/component/milo/client/MiloClientProducerWriteFailureTest.java index aec7677cad6d7..314ac7612ade5 100644 --- a/components/camel-milo/src/test/java/org/apache/camel/component/milo/client/MiloClientProducerWriteFailureTest.java +++ b/components/camel-milo/src/test/java/org/apache/camel/component/milo/client/MiloClientProducerWriteFailureTest.java @@ -51,7 +51,7 @@ * completion from the underlying future (e.g. a Bad_Timeout service fault) and used {@code handleAsync()}, it converted * the failure into a normal completion with a null value. {@code MiloClientProducer} then reported success with a null * body. The fix replaces {@code handleAsync()} with {@code whenCompleteAsync()} to preserve the original exceptional - * completion while still triggering the reconnection side-effect. + * completion. */ class MiloClientProducerWriteFailureTest { diff --git a/components/camel-milo/src/test/java/org/apache/camel/component/milo/client/internal/SubscriptionManagerExceptionPropagationTest.java b/components/camel-milo/src/test/java/org/apache/camel/component/milo/client/internal/SubscriptionManagerExceptionPropagationTest.java index 061e857dd2fa8..4b5fc73b775ae 100644 --- a/components/camel-milo/src/test/java/org/apache/camel/component/milo/client/internal/SubscriptionManagerExceptionPropagationTest.java +++ b/components/camel-milo/src/test/java/org/apache/camel/component/milo/client/internal/SubscriptionManagerExceptionPropagationTest.java @@ -91,9 +91,9 @@ void testWhenCompleteAsyncPreservesExceptionalCompletion() throws Exception { CompletableFuture upstream = new CompletableFuture<>(); upstream.completeExceptionally(serviceError); - // Same chain as the fixed SubscriptionManager.write(): side-effect only, completion preserved + // Same chain as the fixed SubscriptionManager.write(): no side-effect, completion preserved CompletableFuture result = upstream.whenCompleteAsync((r, e) -> { - // side-effect (e.g. handleConnectionFailure); exception propagated by whenCompleteAsync + // no side-effect; exception propagated by whenCompleteAsync }, executor); ExecutionException ex = null; 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 ceb34e9b59065..f2c614238c376 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 @@ -2406,15 +2406,15 @@ the consumer delivered nothing during an outage, and values that did not change A consumer that starts while the server cannot be reached, or whose client cannot connect at all, for example because of wrong credentials or an untrusted server certificate, delivers the status for its nodes too; the cause is in the -log. A connection that Camel rebuilds after a failed request delivers it for every node of the connection. Each loss and -each recovery delivers one exchange per node. +log. A connection that Camel rebuilds delivers it for every node of the connection. Each loss and each recovery +delivers one exchange per node. A route that uses only the value of the `DataValue` body receives a null for these exchanges, which the type converters turn into `0` or `false`; a route that stores values checks the status code of the body first, for example with `${body.statusCode.bad}`. A value can now arrive again after an outage, so a route that counts values, computes differences or detects changes must tolerate repeats. The consumer's route also runs during an outage: an OPC UA write that it makes while the client reconnects waits for the next reconnect attempt and can take effect after communication -is restored, and a request that fails rebuilds the connection. +is restored. To keep the previous behaviour, filter the exchanges with a bad status out right after the consumer, for example with `filter(simple("${body.statusCode.bad} == false"))`. This also drops the other bad statuses, such as the refusal of a @@ -2432,6 +2432,22 @@ An awaited `call` that succeeds now returns the `CallMethodResult` as the exchan always `null` even on success. A route that reads the result of an awaited call must be updated to read the `CallMethodResult` from the body instead of expecting `null`. +=== camel-milo - a failed producer request keeps the connection + +A `milo-client` write, method call or read, or a `milo-browse` browse, that fails (for example with `Bad_Timeout`, or +because the session was lost) no longer disconnects the client connection and rebuilds the subscriptions of the +consumers that share it. The failure is still set on the exchange of an awaited request. A lost session is recovered by +the client itself, and a session that the server no longer has rebuilds the connection, as before. + +Producer requests and `MiloClientConnection.getOpcUaClient()` no longer wait for the subscription work on the same +connection to finish, such as a consumer that starts or stops, or the periodic attempt to create the monitored items +that the server refused. Before, they waited until the server answered that work, and while the network was down +silently, until the client noticed the lost session. While Camel rebuilds the connection, a producer request now fails +at once with an `IllegalStateException` ("No connected", caused by the failure that ended the connection) instead of +waiting for the rebuild to finish. The attempts to create the refused monitored items again now run on the +`MiloClientReconnect` thread instead of the `MiloClient` pool, so that a slow attempt no longer holds up the outcome of +the requests, which that pool completes; on a machine with one processor the pool has a single thread. + === camel-milo - the reconnectTimeout of the components sets the delay between connection attempts `setReconnectTimeout` on the `milo-client` and `milo-browse` components set the request timeout of every service call, From 1b5493aa77b8f8905f6b75d103b7c1e991ed874b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Fri, 23 Oct 2026 22:45:22 +0300 Subject: [PATCH 2/3] CAMEL-25912: camel-milo - move the javadoc of assertStatus to the method Review note on #28371: the javadoc of assertStatus() was left in front of the fourth test when the test was added. Co-authored-by: Claude Fable 5.1 --- .../component/milo/ProducerOnSharedConnectionTest.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/components/camel-milo/src/test/java/org/apache/camel/component/milo/ProducerOnSharedConnectionTest.java b/components/camel-milo/src/test/java/org/apache/camel/component/milo/ProducerOnSharedConnectionTest.java index dedb7cfcc70ad..d71c73cb9e989 100644 --- a/components/camel-milo/src/test/java/org/apache/camel/component/milo/ProducerOnSharedConnectionTest.java +++ b/components/camel-milo/src/test/java/org/apache/camel/component/milo/ProducerOnSharedConnectionTest.java @@ -333,9 +333,6 @@ void writeOnAReplacedConnectionFailsAndKeepsItsReplacement() throws Exception { } } - /** - * Asserts that an awaited request failed with the given status code of the client - */ /** * A connection creates a node that the server refused again while its producer sends a request. With one processor * the executor that completes the requests has a single thread, as here: a write completes while the server holds @@ -403,6 +400,9 @@ public List read( } } + /** + * Asserts that an awaited request failed with the given status code of the client + */ private static void assertStatus(final long statusCode, final Exchange exchange) { final UaException failure = ObjectHelper.getException(UaException.class, exchange.getException()); assertNotNull(failure, () -> "Expected a UaException, but the exchange failed with " + exchange.getException()); From 9871e1b677905fbb37a1b8e5878550f6d6e7dcbd Mon Sep 17 00:00:00 2001 From: etzel-mes Date: Sat, 24 Oct 2026 00:26:05 +0300 Subject: [PATCH 3/3] CAMEL-25912: camel-milo - the upgrade guide names the endpoint and certificate consequence The component page already says that the client recovers a lost session with the endpoint and the server certificate it found when it connected, so a server that changes either while it cannot be reached is reached again only when Camel rebuilds the connection. The 4.24 upgrade guide section of this change now says the same, since a failed producer request used to trigger that rebuild. Co-Authored-By: Claude Fable 5.1 --- .../modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) 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 f2c614238c376..c2e25fc23252a 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 @@ -2437,7 +2437,10 @@ always `null` even on success. A route that reads the result of an awaited call A `milo-client` write, method call or read, or a `milo-browse` browse, that fails (for example with `Bad_Timeout`, or because the session was lost) no longer disconnects the client connection and rebuilds the subscriptions of the consumers that share it. The failure is still set on the exchange of an awaited request. A lost session is recovered by -the client itself, and a session that the server no longer has rebuilds the connection, as before. +the client itself, and a session that the server no longer has rebuilds the connection, as before. The client recovers +a lost session with the endpoint and the server certificate it found when it connected, so a server that changes either +while it cannot be reached is reached again only when Camel rebuilds the connection, for example when the Camel context +restarts. Producer requests and `MiloClientConnection.getOpcUaClient()` no longer wait for the subscription work on the same connection to finish, such as a consumer that starts or stops, or the periodic attempt to create the monitored items