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 b418ab6acd42 CAMEL-25912: camel-milo - client producers no longer wait
for the manager lock or rebuild the connection (#28371)
b418ab6acd42 is described below
commit b418ab6acd424d3d92750a9b42c60fb29fe75bef
Author: Serdar Gökay <[email protected]>
AuthorDate: Sat Oct 24 16:51:34 2026 +0300
CAMEL-25912: camel-milo - client producers no longer wait for the manager
lock or rebuild the connection (#28371)
Co-Authored-By: Claude Fable 5.1 <[email protected]>
Co-authored-by: Claude Fable 5.1 <[email protected]>
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../camel/catalog/docs/milo-client-component.adoc | 6 +-
.../src/main/docs/milo-client-component.adoc | 6 +-
.../milo/client/MiloClientConnection.java | 15 +-
.../milo/client/internal/SubscriptionManager.java | 173 ++++----
.../component/milo/MiloClientExecutorTest.java | 7 +-
.../milo/ProducerOnSharedConnectionTest.java | 433 +++++++++++++++++++++
.../org/apache/camel/component/milo/TcpRelay.java | 28 +-
.../client/MiloClientProducerWriteFailureTest.java | 2 +-
...ubscriptionManagerExceptionPropagationTest.java | 4 +-
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 25 +-
10 files changed, 580 insertions(+), 119 deletions(-)
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 e12de6d7a3b7..3acad3bb65d1 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 e12de6d7a3b7..3acad3bb65d1 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 3b276156dd81..0855eb458265 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 @@ public class MiloClientConnection implements
AutoCloseable, CamelContextAware {
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 @@ public class MiloClientConnection implements AutoCloseable,
CamelContextAware {
// 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 882d185c1069..78a2c8ba4b7a 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 @@ public class SubscriptionManager {
}
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 @@ public class SubscriptionManager {
/**
* 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 @@ public class SubscriptionManager {
}
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 @@ public class SubscriptionManager {
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 class SubscriptionManager {
* @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 @@ public class SubscriptionManager {
* subscriptions.
* <p>
* 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 @@ public class SubscriptionManager {
* failure is the status of the nodes until the next attempt.
* <p>
* 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 @@ public class SubscriptionManager {
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 class SubscriptionManager {
}
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 class SubscriptionManager {
public CompletableFuture<CallMethodResult> 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<ExpandedNodeId> 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<Map<ExpandedNodeId, BrowseResult>> browse(
List<ExpandedNodeId> 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.
+ * <p>
+ * 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 <T> CompletableFuture<T> outcomeOf(final CompletableFuture<T>
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 8326debb6fac..bb52813c9182 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 class MiloClientExecutorTest extends
AbstractMiloServerTest {
}
/**
- * 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 000000000000..d71c73cb9e98
--- /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<Void> 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<Exchange> 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<Exchange> 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();
+ }
+ }
+
+ /**
+ * 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<DataValue> read(
+ final AddressSpace.ReadContext context, final Double
maxAge, final TimestampsToReturn timestamps,
+ final List<ReadValueId> 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<DataValue> 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();
+ }
+ }
+
+ /**
+ * 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());
+ 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<StatusCode> write(final AddressSpace.WriteContext
context, final List<WriteValue> 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 b564fc117aef..032de545559c 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.ServerSocket;
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<Socket> sockets = new ArrayList<>();
+ // the sockets whose data is dropped instead of relayed
+ private final Set<Socket> 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 @@ final class TcpRelay {
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 @@ final class TcpRelay {
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 aec7677cad6d..314ac7612ade 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 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
* 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 061e857dd2fa..4b5fc73b775a 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 @@ class SubscriptionManagerExceptionPropagationTest {
CompletableFuture<Object> 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 4fef939267fc..437fa838df39 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
@@ -2426,15 +2426,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
@@ -2452,6 +2452,25 @@ 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. 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
+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,
