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 330754002072 CAMEL-25694: camel-milo - fix write/call service failures
being reported as success (#27956)
330754002072 is described below
commit 33075400207277fe6cf9ab3cb2c1cf514e50af3d
Author: Guillaume Nodet <[email protected]>
AuthorDate: Fri Oct 23 19:21:13 2026 +0200
CAMEL-25694: camel-milo - fix write/call service failures being reported as
success (#27956)
* CAMEL-25694: camel-milo - fix write/call service failures being reported
as success
In SubscriptionManager.write(), call(), readValues() and browse(), the
futures
returned by the underlying Milo client were chained with handleAsync().
When the
upstream future completed exceptionally (e.g. Bad_Timeout service fault),
the
handleAsync() callback ran with a non-null 'e' parameter but returned the
result
value (null), converting the exceptional completion into a normal
completion with
a null value.
MiloClientProducer awaits the future with whenComplete() and only sets an
exchange
exception when the throwable is non-null. Since handleAsync() had already
discarded
the exception, the producer saw a null-completed future, reported success,
and set
the exchange body to null.
The fix replaces handleAsync() with whenCompleteAsync() in all four methods.
whenCompleteAsync() runs the side-effect (handleConnectionFailure) but
propagates
the original completion unchanged: exceptional stays exceptional, success
stays
success.
Co-authored-by: Claude Sonnet 4.6 <[email protected]>
* fix: ServerCertificateValidationTest - start consumer routes after
initial write
OPC UA consumers deliver the current server node value when they subscribe.
In the original test, all routes started automatically before the write, so
each consumer received two messages: the uninitialised value at subscription
time, and the 'Foo' write. The mock expected exactly 1 message, causing a
deterministic failure ('Expected: <1> but was: <2>').
Follow the pattern established in MonitorItemTest: mark consumer routes with
autoStartup(false), write the initial value first via the server producer,
then start all consumer routes. Each consumer then sees only one
notification
(the value written before it subscribed).
* fix: update SubscriptionManager test constructor calls to match new
signature
* fix: address review - add awaited call test, upgrade guide, revert
redundant test changes
- Add testAwaitedCallReturnsResult() to CallClientTest to verify that an
awaited
call returns a non-null CallMethodResult with a good status and the
expected
output argument (CAMEL-25694 regression test on the real server path)
- Add upgrade guide entry for the two observable behaviour changes:
* an awaited write/call/read that fails now fails the exchange
* an awaited call that succeeds now returns CallMethodResult as the body
- Revert comment-only changes to ServerCertificateValidationTest: main
already
has the autoStartup(false) fix from CAMEL-25645 (#27999), so these
rewrites
only add noise to the commit history
* fix: fix upgrade guide wording - Bad_TypeMismatch is a per-node status,
not a transport failure
* fix: use correct requestBodyAndHeader signature with endpoint URI in
CallClientTest
---------
Co-authored-by: Claude Sonnet 4.6 <[email protected]>
---
.../milo/client/internal/SubscriptionManager.java | 20 +-
.../camel/component/milo/call/CallClientTest.java | 15 ++
.../client/MiloClientProducerWriteFailureTest.java | 221 +++++++++++++++++++++
...ubscriptionManagerExceptionPropagationTest.java | 203 +++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 12 ++
5 files changed, 459 insertions(+), 12 deletions(-)
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 7ff69a4ec4bc..882d185c1069 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
@@ -1708,13 +1708,12 @@ public class SubscriptionManager {
return newNotConnectedResult();
}
- return this.connected.call(nodeId, methodId,
inputArguments).handleAsync((status, e) -> {
+ return this.connected.call(nodeId, methodId,
inputArguments).whenCompleteAsync((status, e) -> {
// handle outside the lock, running using
- // handleAsync
+ // whenCompleteAsync (preserves the original exceptional
completion)
if (e != null) {
handleConnectionFailure(e);
}
- return null;
}, this.executor);
} finally {
lock.unlock();
@@ -1728,13 +1727,12 @@ public class SubscriptionManager {
return newNotConnectedResult();
}
- return this.connected.write(nodeId, value).handleAsync((status, e)
-> {
+ return this.connected.write(nodeId,
value).whenCompleteAsync((status, e) -> {
// handle outside the lock, running using
- // handleAsync
+ // whenCompleteAsync (preserves the original exceptional
completion)
if (e != null) {
handleConnectionFailure(e);
}
- return status;
}, this.executor);
} finally {
lock.unlock();
@@ -1748,13 +1746,12 @@ public class SubscriptionManager {
return newNotConnectedResult();
}
- return this.connected.readValues(nodeIds).handleAsync((nodes, e)
-> {
+ return
this.connected.readValues(nodeIds).whenCompleteAsync((nodes, e) -> {
// handle outside the lock, running using
- // handleAsync
+ // whenCompleteAsync (preserves the original exceptional
completion)
if (e != null) {
handleConnectionFailure(e);
}
- return nodes;
}, this.executor);
} finally {
lock.unlock();
@@ -1772,13 +1769,12 @@ public class SubscriptionManager {
return this.connected.browse(expandedNodeIds, direction,
nodeClasses, 1, maxDepth,
null != filter ? Pattern.compile(filter) : null,
includeSubTypes, maxNodesPerRequest)
- .handleAsync((browseResult, e) -> {
+ .whenCompleteAsync((browseResult, e) -> {
// handle outside the lock, running using
- // handleAsync
+ // whenCompleteAsync (preserves the original
exceptional completion)
if (e != null) {
handleConnectionFailure(e);
}
- return browseResult;
}, this.executor);
} finally {
lock.unlock();
diff --git
a/components/camel-milo/src/test/java/org/apache/camel/component/milo/call/CallClientTest.java
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/call/CallClientTest.java
index 5e5ff94b3916..3255db6dc935 100644
---
a/components/camel-milo/src/test/java/org/apache/camel/component/milo/call/CallClientTest.java
+++
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/call/CallClientTest.java
@@ -37,6 +37,7 @@ import org.eclipse.milo.opcua.stack.core.security.*;
import org.eclipse.milo.opcua.stack.core.transport.TransportProfile;
import org.eclipse.milo.opcua.stack.core.types.builtin.LocalizedText;
import org.eclipse.milo.opcua.stack.core.types.enumerated.MessageSecurityMode;
+import org.eclipse.milo.opcua.stack.core.types.structured.CallMethodResult;
import org.eclipse.milo.opcua.stack.core.types.structured.UserTokenPolicy;
import org.eclipse.milo.opcua.stack.transport.server.tcp.OpcTcpServerTransport;
import
org.eclipse.milo.opcua.stack.transport.server.tcp.OpcTcpServerTransportConfig;
@@ -48,7 +49,9 @@ import org.slf4j.LoggerFactory;
import static org.apache.camel.component.milo.NodeIds.nodeValue;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* Unit tests for calling from the client side
@@ -194,6 +197,18 @@ public class CallClientTest extends AbstractMiloServerTest
{
.build();
}
+ @Test
+ public void testAwaitedCallReturnsResult() {
+ // Before CAMEL-25694, the call() handleAsync unconditionally returned
null even on success,
+ // so an awaited call always left the exchange body null. Now the body
is the CallMethodResult.
+ final CallMethodResult result = this.producer1.requestBodyAndHeader(
+ DIRECT_START_1, "foo", "CamelMiloAwait", true,
CallMethodResult.class);
+
+ assertNotNull(result);
+ assertTrue(result.getStatusCode().isGood());
+ assertEquals("out-foo", result.getOutputArguments()[0].getValue());
+ }
+
@Test
public void testCall1() {
// call
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
new file mode 100644
index 000000000000..aec7677cad6d
--- /dev/null
+++
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/client/MiloClientProducerWriteFailureTest.java
@@ -0,0 +1,221 @@
+/*
+ * 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.client;
+
+import java.lang.reflect.Field;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.component.milo.MiloConstants;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultExchange;
+import org.eclipse.milo.opcua.stack.core.types.builtin.ExpandedNodeId;
+import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
+import org.eclipse.milo.opcua.stack.core.types.enumerated.BrowseDirection;
+import org.eclipse.milo.opcua.stack.core.types.structured.BrowseResult;
+import org.eclipse.milo.opcua.stack.core.types.structured.CallMethodResult;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Unit tests verifying that OPC-UA service failures from awaited write and
call operations are propagated as exchange
+ * exceptions rather than being silently swallowed and reported as success.
+ *
+ * <p>
+ * Regression test for CAMEL-25694: when {@code SubscriptionManager.write()}
or {@code call()} received an exceptional
+ * 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.
+ */
+class MiloClientProducerWriteFailureTest {
+
+ /**
+ * Stub connection that returns a pre-arranged future from {@code
writeValue()} and {@code call()}, allowing tests
+ * to simulate service failures without a real OPC UA server.
+ */
+ static class StubMiloClientConnection extends MiloClientConnection {
+
+ private final CompletableFuture<?> writeResult;
+ private final CompletableFuture<CallMethodResult> callResult;
+
+ StubMiloClientConnection(CompletableFuture<?> writeResult,
CompletableFuture<CallMethodResult> callResult) {
+ super(new MiloClientConfiguration());
+ this.writeResult = writeResult;
+ this.callResult = callResult;
+ }
+
+ @Override
+ public CompletableFuture<?> writeValue(ExpandedNodeId nodeId, Object
value) {
+ return writeResult;
+ }
+
+ @Override
+ public CompletableFuture<CallMethodResult> call(ExpandedNodeId nodeId,
ExpandedNodeId methodId, Object value) {
+ return callResult;
+ }
+
+ @Override
+ public CompletableFuture<?> readValues(List<ExpandedNodeId> nodeIds) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<Map<ExpandedNodeId, BrowseResult>> browse(
+ List<ExpandedNodeId> expandedNodeIds, BrowseDirection
direction, int nodeClasses, int maxDepth,
+ String filter, boolean includeSubTypes, int
maxNodesPerRequest) {
+ return CompletableFuture.completedFuture(null);
+ }
+ }
+
+ private DefaultCamelContext camelContext;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ camelContext = new DefaultCamelContext();
+ camelContext.start();
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ camelContext.stop();
+ }
+
+ /**
+ * Verifies that when the awaited write future completes exceptionally,
the exchange receives the exception rather
+ * than reporting success with a null body.
+ */
+ @Test
+ void testAwaitedWriteServiceFailurePropagatesAsException() throws
Exception {
+ RuntimeException serviceError = new RuntimeException("Bad_Timeout
OPC-UA service fault");
+ CompletableFuture<StatusCode> failingWrite = new CompletableFuture<>();
+ failingWrite.completeExceptionally(serviceError);
+
+ Exchange exchange = runProducerWithStubConnection(
+ new StubMiloClientConnection(failingWrite,
CompletableFuture.completedFuture(null)),
+ null, true);
+
+ assertNotNull(exchange.getException(), "Exchange must carry the
service exception on failed awaited write");
+ assertInstanceOf(RuntimeException.class, exchange.getException());
+ }
+
+ /**
+ * Verifies that when the awaited call future completes exceptionally, the
exchange receives the exception rather
+ * than reporting success with a null body.
+ */
+ @Test
+ void testAwaitedCallServiceFailurePropagatesAsException() throws Exception
{
+ RuntimeException serviceError = new RuntimeException("Bad_Timeout
OPC-UA service fault");
+ CompletableFuture<CallMethodResult> failingCall = new
CompletableFuture<>();
+ failingCall.completeExceptionally(serviceError);
+
+ Exchange exchange = runProducerWithStubConnection(
+ new
StubMiloClientConnection(CompletableFuture.completedFuture(null), failingCall),
+ "ns=1;s=myMethod", true);
+
+ assertNotNull(exchange.getException(), "Exchange must carry the
service exception on failed awaited call");
+ assertInstanceOf(RuntimeException.class, exchange.getException());
+ }
+
+ /**
+ * Verifies that a successful awaited write does not set an exception on
the exchange.
+ */
+ @Test
+ void testAwaitedWriteSuccessDoesNotSetException() throws Exception {
+ CompletableFuture<StatusCode> successWrite =
CompletableFuture.completedFuture(StatusCode.GOOD);
+
+ Exchange exchange = runProducerWithStubConnection(
+ new StubMiloClientConnection(successWrite,
CompletableFuture.completedFuture(null)),
+ null, true);
+
+ assertNull(exchange.getException(), "Exchange must not carry an
exception on successful awaited write");
+ }
+
+ /**
+ * Verifies that non-awaited writes (fire-and-forget mode) do not
propagate the exception to the exchange.
+ */
+ @Test
+ void testNonAwaitedWriteServiceFailureDoesNotSetException() throws
Exception {
+ RuntimeException serviceError = new RuntimeException("Bad_Timeout
OPC-UA service fault");
+ CompletableFuture<StatusCode> failingWrite = new CompletableFuture<>();
+ failingWrite.completeExceptionally(serviceError);
+
+ Exchange exchange = runProducerWithStubConnection(
+ new StubMiloClientConnection(failingWrite,
CompletableFuture.completedFuture(null)),
+ null, false);
+
+ assertNull(exchange.getException(), "Non-awaited write must not
propagate exceptions to the exchange");
+ }
+
+ /**
+ * Creates a {@link MiloClientProducer} backed by the given stub
connection, runs process() synchronously, and
+ * returns the resulting exchange.
+ *
+ * @param stubConnection the stub connection returning pre-arranged futures
+ * @param methodIdStr if non-null, treated as a method-call producer;
otherwise a write producer
+ * @param await whether to set {@code CamelMiloAwait=true} on the
exchange
+ */
+ private Exchange runProducerWithStubConnection(
+ StubMiloClientConnection stubConnection,
+ String methodIdStr,
+ boolean await)
+ throws Exception {
+
+ MiloClientEndpoint endpoint = new MiloClientEndpoint(
+ "milo-client:opc.tcp://localhost:4840", new
MiloClientComponent(), "opc.tcp://localhost:4840",
+ new MiloClientCachingConnectionManager());
+ endpoint.setCamelContext(camelContext);
+ endpoint.setNode("ns=1;s=myNode");
+ if (methodIdStr != null) {
+ endpoint.setMethod(methodIdStr);
+ }
+
+ MiloClientProducer producer = new MiloClientProducer(endpoint, await);
+
+ // Inject the stub connection via reflection to bypass doStart() which
requires a real server
+ Field connectionField =
MiloClientProducer.class.getDeclaredField("connection");
+ connectionField.setAccessible(true);
+ connectionField.set(producer, stubConnection);
+
+ DefaultExchange exchange = new DefaultExchange(camelContext);
+ exchange.getIn().setBody("testValue");
+ exchange.getIn().setHeader(MiloConstants.HEADER_AWAIT, await);
+
+ CountDownLatch latch = new CountDownLatch(1);
+ AtomicReference<Exchange> resultRef = new AtomicReference<>();
+
+ producer.process(exchange, done -> {
+ resultRef.set(exchange);
+ latch.countDown();
+ });
+
+ latch.await(5, TimeUnit.SECONDS);
+ assertTrue(latch.getCount() == 0, "Producer callback must complete
within 5 seconds");
+ return resultRef.get();
+ }
+}
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
new file mode 100644
index 000000000000..061e857dd2fa
--- /dev/null
+++
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/client/internal/SubscriptionManagerExceptionPropagationTest.java
@@ -0,0 +1,203 @@
+/*
+ * 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.client.internal;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.component.milo.client.MiloClientConfiguration;
+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.Variant;
+import org.eclipse.milo.opcua.stack.core.types.structured.CallMethodResult;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * Unit tests for {@link SubscriptionManager} exception propagation through
write() and call().
+ *
+ * <p>
+ * Regression tests for CAMEL-25694: {@code SubscriptionManager.write()} and
{@code call()} previously used
+ * {@code handleAsync()} which converted exceptional completions into normal
completions with null values. The fix
+ * replaces {@code handleAsync()} with {@code whenCompleteAsync()} to preserve
the original exceptional completion.
+ *
+ * <p>
+ * The {@code whenCompleteAsync}-vs-{@code handleAsync} contract is tested via
standalone helper methods (no real OPC UA
+ * connection needed). The disconnected-manager tests use a real {@link
SubscriptionManager} to verify that
+ * {@code write()}, {@code call()}, and {@code readValues()} all return
exceptional futures before any connection is
+ * established.
+ */
+class SubscriptionManagerExceptionPropagationTest {
+
+ private ScheduledExecutorService executor;
+
+ @BeforeEach
+ void setUp() {
+ executor = Executors.newSingleThreadScheduledExecutor();
+ }
+
+ @AfterEach
+ void tearDown() throws InterruptedException {
+ executor.shutdownNow();
+ assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS),
+ "Executor must shut down cleanly within 5 seconds");
+ }
+
+ //
-------------------------------------------------------------------------
+ // Contract tests: whenCompleteAsync vs handleAsync
+ //
+ // These do NOT instantiate SubscriptionManager (which would trigger a
+ // connection attempt in its constructor). Instead they exercise the same
+ // CompletableFuture chain used by the production fix.
+ //
-------------------------------------------------------------------------
+
+ /**
+ * Verifies that {@code whenCompleteAsync} (the fix) preserves an
exceptional completion.
+ *
+ * <p>
+ * Before CAMEL-25694 was fixed, {@code SubscriptionManager.write()} used
{@code handleAsync()}, which converted the
+ * exceptional completion into a normal null completion. This test
verifies the correct behavior: a
+ * {@code whenCompleteAsync} side-effect chain over a failed future must
stay failed.
+ */
+ @Test
+ void testWhenCompleteAsyncPreservesExceptionalCompletion() throws
Exception {
+ RuntimeException serviceError = new RuntimeException("Bad_Timeout
OPC-UA service fault");
+ CompletableFuture<Object> upstream = new CompletableFuture<>();
+ upstream.completeExceptionally(serviceError);
+
+ // Same chain as the fixed SubscriptionManager.write(): side-effect
only, completion preserved
+ CompletableFuture<?> result = upstream.whenCompleteAsync((r, e) -> {
+ // side-effect (e.g. handleConnectionFailure); exception
propagated by whenCompleteAsync
+ }, executor);
+
+ ExecutionException ex = null;
+ try {
+ result.get(5, TimeUnit.SECONDS);
+ } catch (ExecutionException e) {
+ ex = e;
+ } catch (Exception e) {
+ fail("Unexpected exception type: " + e);
+ }
+ assertNotNull(ex, "Get on a failed whenCompleteAsync future must throw
ExecutionException");
+ assertInstanceOf(RuntimeException.class, ex.getCause());
+ assertTrue(result.isCompletedExceptionally(),
+ "whenCompleteAsync must preserve the exceptional completion");
+ }
+
+ /**
+ * Demonstrates why the old code was wrong: {@code handleAsync} converts
an exceptional completion into a normal
+ * null completion, silently swallowing the service failure.
+ *
+ * <p>
+ * This test documents the buggy behavior. If the fix is regressed by
reintroducing {@code handleAsync},
+ * {@link #testWhenCompleteAsyncPreservesExceptionalCompletion()} will
fail, and this test shows the exact semantic
+ * difference.
+ */
+ @Test
+ void testHandleAsyncSwallowsExceptionalCompletion() throws Exception {
+ RuntimeException serviceError = new RuntimeException("Bad_Timeout
OPC-UA service fault");
+ CompletableFuture<Object> upstream = new CompletableFuture<>();
+ upstream.completeExceptionally(serviceError);
+
+ // Same chain as the old buggy SubscriptionManager.write(): returns
result (null) on exception
+ CompletableFuture<?> result = upstream.handleAsync((r, e) -> r,
executor);
+
+ // Wait for the handleAsync chain to complete normally (it swallows
the exception)
+ result.join();
+
+ assertFalse(result.isCompletedExceptionally(),
+ "handleAsync incorrectly completes normally with null when
upstream fails");
+ assertNull(result.get(),
+ "handleAsync returns null when upstream failed — this is the
swallowed failure");
+ }
+
+ //
-------------------------------------------------------------------------
+ // SubscriptionManager disconnected-path tests
+ //
+ // These instantiate a real SubscriptionManager with a default (no-server)
+ // configuration. The constructor triggers a connect() call that fails
+ // immediately (null endpoint URI), leaving the manager in the disconnected
+ // state. write(), call(), and readValues() must return exceptional
futures.
+ //
-------------------------------------------------------------------------
+
+ /**
+ * Verifies that a disconnected manager returns an exceptional future from
write(), ensuring producers fail fast
+ * when there is no connection.
+ */
+ @Test
+ void testWriteWhenDisconnectedReturnsExceptionalFuture() {
+ SubscriptionManager manager = new SubscriptionManager(new
MiloClientConfiguration(), executor, executor, 10_000);
+ try {
+ CompletableFuture<?> result =
manager.write(ExpandedNodeId.parse("ns=1;s=myNode"),
+ new DataValue(new Variant("testValue")));
+
+ assertTrue(result.isCompletedExceptionally(),
+ "write() must return an exceptional future when not
connected");
+ } finally {
+ manager.dispose();
+ }
+ }
+
+ /**
+ * Verifies that a disconnected manager returns an exceptional future from
call(), ensuring producers fail fast when
+ * there is no connection.
+ */
+ @Test
+ void testCallWhenDisconnectedReturnsExceptionalFuture() {
+ SubscriptionManager manager = new SubscriptionManager(new
MiloClientConfiguration(), executor, executor, 10_000);
+ try {
+ CompletableFuture<CallMethodResult> result = manager.call(
+ ExpandedNodeId.parse("ns=1;s=myNode"),
+ ExpandedNodeId.parse("ns=1;s=myMethod"),
+ new Variant[0]);
+
+ assertTrue(result.isCompletedExceptionally(),
+ "call() must return an exceptional future when not
connected");
+ } finally {
+ manager.dispose();
+ }
+ }
+
+ /**
+ * Verifies that a disconnected manager returns an exceptional future from
readValues(), ensuring consumers fail
+ * fast when there is no connection.
+ */
+ @Test
+ void testReadValuesWhenDisconnectedReturnsExceptionalFuture() {
+ SubscriptionManager manager = new SubscriptionManager(new
MiloClientConfiguration(), executor, executor, 10_000);
+ try {
+ CompletableFuture<?> result =
manager.readValues(List.of(ExpandedNodeId.parse("ns=1;s=myNode")));
+
+ assertTrue(result.isCompletedExceptionally(),
+ "readValues() must return an exceptional future when not
connected");
+ } finally {
+ manager.dispose();
+ }
+ }
+}
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 c513494b38f6..b213826e71a7 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
@@ -2416,6 +2416,18 @@ To keep the previous behaviour, filter the exchanges
with a bad status out right
`filter(simple("${body.statusCode.bad} == false"))`. This also drops the other
bad statuses, such as the refusal of a
node, and the current values delivered after an outage still arrive.
+=== camel-milo - an awaited write, call or read that fails now fails the
exchange
+
+An awaited `milo-client` write, method call or read that fails at the service
or transport level, for example a
+request that times out (`Bad_Timeout`) or a lost connection, now fails the
exchange and lets the route's error
+handler run. Before, the exchange completed without an error and with a `null`
body, silently discarding the failure.
+A bad status for the node itself, such as `Bad_TypeMismatch` on a write, is
not a failure of the operation and is
+still returned as the body.
+
+An awaited `call` that succeeds now returns the `CallMethodResult` as the
exchange body. Before, the body was
+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 - 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,
