From 8a277280f5fda576db4d9e8557765f7e6d14422c Mon Sep 17 00:00:00 2001 From: gnodet Date: Fri, 16 Oct 2026 15:23:21 +0000 Subject: [PATCH 1/6] 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 --- .../client/internal/SubscriptionManager.java | 20 +- .../MiloClientProducerWriteFailureTest.java | 221 ++++++++++++++++++ ...iptionManagerExceptionPropagationTest.java | 203 ++++++++++++++++ 3 files changed, 432 insertions(+), 12 deletions(-) create mode 100644 components/camel-milo/src/test/java/org/apache/camel/component/milo/client/MiloClientProducerWriteFailureTest.java create mode 100644 components/camel-milo/src/test/java/org/apache/camel/component/milo/client/internal/SubscriptionManagerExceptionPropagationTest.java 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 88aae74851140..37281416f7b27 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 @@ -1542,13 +1542,12 @@ public CompletableFuture call( 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(); @@ -1562,13 +1561,12 @@ public CompletableFuture write(final ExpandedNodeId nodeId, final DataValue v 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(); @@ -1582,13 +1580,12 @@ public CompletableFuture readValues(final List nodeIds) { 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(); @@ -1606,13 +1603,12 @@ public CompletableFuture> browse( 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/client/MiloClientProducerWriteFailureTest.java b/components/camel-milo/src/test/java/org/apache/camel/component/milo/client/MiloClientProducerWriteFailureTest.java new file mode 100644 index 0000000000000..aec7677cad6d7 --- /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. + * + *

+ * 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 callResult; + + StubMiloClientConnection(CompletableFuture writeResult, CompletableFuture callResult) { + super(new MiloClientConfiguration()); + this.writeResult = writeResult; + this.callResult = callResult; + } + + @Override + public CompletableFuture writeValue(ExpandedNodeId nodeId, Object value) { + return writeResult; + } + + @Override + public CompletableFuture call(ExpandedNodeId nodeId, ExpandedNodeId methodId, Object value) { + return callResult; + } + + @Override + public CompletableFuture readValues(List nodeIds) { + return CompletableFuture.completedFuture(null); + } + + @Override + public CompletableFuture> browse( + List 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 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 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 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 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 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 0000000000000..7a456f27fbec3 --- /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(). + * + *

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

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

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

+ * 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 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, 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, 10_000); + try { + CompletableFuture 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, 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(); + } + } +} From 1ef8bd830758e828438f8418323457ef0f48a534 Mon Sep 17 00:00:00 2001 From: gnodet Date: Wed, 21 Oct 2026 16:13:34 +0000 Subject: [PATCH 2/6] 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). --- .../component/milo/ServerCertificateValidationTest.java | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/components/camel-milo/src/test/java/org/apache/camel/component/milo/ServerCertificateValidationTest.java b/components/camel-milo/src/test/java/org/apache/camel/component/milo/ServerCertificateValidationTest.java index ff2abef17cbd1..f816b99cddb98 100644 --- a/components/camel-milo/src/test/java/org/apache/camel/component/milo/ServerCertificateValidationTest.java +++ b/components/camel-milo/src/test/java/org/apache/camel/component/milo/ServerCertificateValidationTest.java @@ -126,7 +126,8 @@ protected RoutesBuilder createRouteBuilder() { public void configure() { from(DIRECT_START).to(MILO_SERVER_ITEM); - // the consumers deliver the current value while they start, so the test starts these routes + // consumers deliver the current server value when they subscribe, so they are started + // only after the initial write to avoid receiving both the initial value and the write as two messages from(resolve(trustListClient("trusted"))).autoStartup(false).to("mock:trustList"); from(resolve(trustListClient("empty"))).autoStartup(false).to("mock:emptyTrustList"); from(resolve(MILO_CLIENT)).autoStartup(false).to("mock:notConfigured"); @@ -151,8 +152,9 @@ void onlyClientsThatCanValidateTheServerCertificateConnect() throws Exception { testBody(this.notConfiguredEndpoint.message(0), assertNoCommunication()); this.notConfiguredEndpoint.setAssertPeriod(5_000); - // the value is written before the consumers start, so it is the first notification of each client that - // connects (and not the initial Bad value of the item) + // write the initial value before starting the consumer routes so that each consumer receives + // exactly one notification (the current value at subscription time) rather than two + // (the uninitialized value at subscription time, then the write) this.producer.sendBody("Foo"); context.getRouteController().startAllRoutes(); From a2a0eb8d800ec88f162aefc139064600ebc9d01d Mon Sep 17 00:00:00 2001 From: gnodet Date: Wed, 21 Oct 2026 20:27:43 +0000 Subject: [PATCH 3/6] fix: update SubscriptionManager test constructor calls to match new signature --- .../SubscriptionManagerExceptionPropagationTest.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) 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 7a456f27fbec3..061e857dd2fa8 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 @@ -152,7 +152,7 @@ void testHandleAsyncSwallowsExceptionalCompletion() throws Exception { */ @Test void testWriteWhenDisconnectedReturnsExceptionalFuture() { - SubscriptionManager manager = new SubscriptionManager(new MiloClientConfiguration(), executor, 10_000); + 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"))); @@ -170,7 +170,7 @@ void testWriteWhenDisconnectedReturnsExceptionalFuture() { */ @Test void testCallWhenDisconnectedReturnsExceptionalFuture() { - SubscriptionManager manager = new SubscriptionManager(new MiloClientConfiguration(), executor, 10_000); + SubscriptionManager manager = new SubscriptionManager(new MiloClientConfiguration(), executor, executor, 10_000); try { CompletableFuture result = manager.call( ExpandedNodeId.parse("ns=1;s=myNode"), @@ -190,7 +190,7 @@ void testCallWhenDisconnectedReturnsExceptionalFuture() { */ @Test void testReadValuesWhenDisconnectedReturnsExceptionalFuture() { - SubscriptionManager manager = new SubscriptionManager(new MiloClientConfiguration(), executor, 10_000); + SubscriptionManager manager = new SubscriptionManager(new MiloClientConfiguration(), executor, executor, 10_000); try { CompletableFuture result = manager.readValues(List.of(ExpandedNodeId.parse("ns=1;s=myNode"))); From 689e77abae33b113fb090c127c63b3a3b9f5aecb Mon Sep 17 00:00:00 2001 From: Guillaume Nodet Date: Fri, 23 Oct 2026 08:03:11 +0000 Subject: [PATCH 4/6] 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 --- .../milo/ServerCertificateValidationTest.java | 8 +++----- .../camel/component/milo/call/CallClientTest.java | 15 +++++++++++++++ .../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 10 ++++++++++ 3 files changed, 28 insertions(+), 5 deletions(-) diff --git a/components/camel-milo/src/test/java/org/apache/camel/component/milo/ServerCertificateValidationTest.java b/components/camel-milo/src/test/java/org/apache/camel/component/milo/ServerCertificateValidationTest.java index f816b99cddb98..ff2abef17cbd1 100644 --- a/components/camel-milo/src/test/java/org/apache/camel/component/milo/ServerCertificateValidationTest.java +++ b/components/camel-milo/src/test/java/org/apache/camel/component/milo/ServerCertificateValidationTest.java @@ -126,8 +126,7 @@ protected RoutesBuilder createRouteBuilder() { public void configure() { from(DIRECT_START).to(MILO_SERVER_ITEM); - // consumers deliver the current server value when they subscribe, so they are started - // only after the initial write to avoid receiving both the initial value and the write as two messages + // the consumers deliver the current value while they start, so the test starts these routes from(resolve(trustListClient("trusted"))).autoStartup(false).to("mock:trustList"); from(resolve(trustListClient("empty"))).autoStartup(false).to("mock:emptyTrustList"); from(resolve(MILO_CLIENT)).autoStartup(false).to("mock:notConfigured"); @@ -152,9 +151,8 @@ void onlyClientsThatCanValidateTheServerCertificateConnect() throws Exception { testBody(this.notConfiguredEndpoint.message(0), assertNoCommunication()); this.notConfiguredEndpoint.setAssertPeriod(5_000); - // write the initial value before starting the consumer routes so that each consumer receives - // exactly one notification (the current value at subscription time) rather than two - // (the uninitialized value at subscription time, then the write) + // the value is written before the consumers start, so it is the first notification of each client that + // connects (and not the initial Bad value of the item) this.producer.sendBody("Foo"); context.getRouteController().startAllRoutes(); 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 5e5ff94b3916b..8ecdeb794cbf1 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.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 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 @@ private EndpointConfig buildTcpEndpoint(EndpointConfig.Builder base) { .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( + "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/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 5587787d422e7..3ce0307a66f5a 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 @@ -2131,6 +2131,16 @@ 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 whose server-side operation fails, for example a write +of a wrong type or a call that times out, 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. + +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-protobuf - a bytes field in a Map When a protobuf message is converted to a `Map` (for example `getBody(Map.class)` after unmarshalling), the value of From 0b4c0e742c31a9d65520a8e93acee303e38af9f5 Mon Sep 17 00:00:00 2001 From: gnodet Date: Fri, 23 Oct 2026 08:43:03 +0000 Subject: [PATCH 5/6] fix: fix upgrade guide wording - Bad_TypeMismatch is a per-node status, not a transport failure --- .../modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) 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 fe62f5da42851..2a59360524ae7 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 @@ -2368,9 +2368,11 @@ 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 whose server-side operation fails, for example a write -of a wrong type or a call that times out, 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. +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 From f0992a73ef74ba2882fdb06bd24d2c7295594e0b Mon Sep 17 00:00:00 2001 From: gnodet Date: Fri, 23 Oct 2026 11:11:13 +0000 Subject: [PATCH 6/6] fix: use correct requestBodyAndHeader signature with endpoint URI in CallClientTest --- .../org/apache/camel/component/milo/call/CallClientTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 8ecdeb794cbf1..3255db6dc935d 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 @@ -202,7 +202,7 @@ 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( - "foo", "CamelMiloAwait", true, CallMethodResult.class); + DIRECT_START_1, "foo", "CamelMiloAwait", true, CallMethodResult.class); assertNotNull(result); assertTrue(result.getStatusCode().isGood());