From ec1e6a7039a9c8424d3ff021f9f1b4f1215384af Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Sun, 18 Oct 2026 10:34:42 +0300 Subject: [PATCH] CAMEL-25690: camel-milo - answer a producer node with an unknown namespace URI instead of reconnecting The producers resolved the namespace URI (nsu=) of a node ID through the namespace table of the server and failed with a NullPointerException when the URI was not in it. The write, method call, read or browse took that for a lost connection: the client reconnected, which also recreated the subscriptions of the consumers that share the connection. The consumer and the producers now share one node lookup, which completes with null for such a URI, as the lookup of the consumer has done since CAMEL-25671. A producer answers such a node like a node the server does not have, without sending it: Bad_NodeIdUnknown for a write and for that node of a read or browse, and Bad_NodeIdUnknown or Bad_MethodInvalid for a method call. The other nodes of the request, the connection and its consumers are not affected. The method call answer does not reach the exchange yet, because SubscriptionManager.call drops every call result (CAMEL-25694). Co-authored-by: Claude Opus 5.5 --- .../catalog/docs/milo-client-component.adoc | 5 + .../src/main/docs/milo-client-component.adoc | 5 + .../client/internal/SubscriptionManager.java | 93 ++++++-- .../milo/ProducerUnknownNamespaceTest.java | 211 ++++++++++++++++++ .../pages/camel-4x-upgrade-guide-4_24.adoc | 9 + 5 files changed, 301 insertions(+), 22 deletions(-) create mode 100644 components/camel-milo/src/test/java/org/apache/camel/component/milo/ProducerUnknownNamespaceTest.java diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/milo-client-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/milo-client-component.adoc index 6b9e7e4b13e56..fec24e18aa2b4 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 @@ -127,6 +127,11 @@ However Camel allows to wrap the actual value inside `RAW(…)`, which makes esc milo-client:opc.tcp://user:password@localhost:12345?node=RAW(nsu=http://foo.bar;s=foo/bar) ---- +A producer request for a node whose namespace URI (`nsu=`) is not in the namespace table of the server is answered +like a request for a node that the server does not have. A write returns the status code `Bad_NodeIdUnknown`. A read, +or a browse of the xref:milo-browse-component.adoc[Milo Browse] component, returns that status code for this node and +the results of the other nodes. The connection and the consumers that share it are not affected. + === Monitoring and writing a node The following routes monitor one node and write another, on the server of the 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 6b9e7e4b13e56..fec24e18aa2b4 100644 --- a/components/camel-milo/src/main/docs/milo-client-component.adoc +++ b/components/camel-milo/src/main/docs/milo-client-component.adoc @@ -127,6 +127,11 @@ However Camel allows to wrap the actual value inside `RAW(…)`, which makes esc milo-client:opc.tcp://user:password@localhost:12345?node=RAW(nsu=http://foo.bar;s=foo/bar) ---- +A producer request for a node whose namespace URI (`nsu=`) is not in the namespace table of the server is answered +like a request for a node that the server does not have. A write returns the status code `Bad_NodeIdUnknown`. A read, +or a browse of the xref:milo-browse-component.adoc[Milo Browse] component, returns that status code for this node and +the results of the other nodes. The connection and the consumers that share it are not affected. + === Monitoring and writing a node The following routes monitor one node and write another, on the server of the 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 e8dab0b933369..bda548e475af5 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 @@ -23,6 +23,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.Iterator; import java.util.LinkedHashMap; import java.util.LinkedList; import java.util.List; @@ -66,6 +67,7 @@ import org.eclipse.milo.opcua.stack.core.security.CertificateValidator; import org.eclipse.milo.opcua.stack.core.types.builtin.ByteString; import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue; +import org.eclipse.milo.opcua.stack.core.types.builtin.DiagnosticInfo; 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.QualifiedName; @@ -266,7 +268,7 @@ public List putSubscriptions(final Map sub for (final Map.Entry entry : subscriptions.entrySet()) { final Subscription s = entry.getValue(); - final NodeId node = lookupMonitoredNode(s.getNodeId()); + final NodeId node = lookupNamespace(s.getNodeId()).get(); if (node == null) { // reported once like a node the server rejects, the other nodes and the connection are not affected @@ -342,16 +344,6 @@ public List putSubscriptions(final Map sub return pendingValues; } - /** - * @return the node ID with the namespace index of the server, or null if its namespace URI is not in the - * namespace table of the server - */ - private NodeId lookupMonitoredNode(final ExpandedNodeId nodeId) throws Exception { - final String uri = nodeId.getNamespaceUri(); - final UShort index = uri == null ? nodeId.getNamespaceIndex() : lookupNamespaceIndex(uri).get(); - return index == null ? null : toNodeId(index, nodeId); - } - /** * Creates the subscription on the server again, with the monitored items it had, after the server lost it. * @@ -445,6 +437,10 @@ public void dispose() { } } + /** + * @return the node ID with the namespace index of the server, completed with null if its namespace URI is not + * in the namespace table of the server: such a node does not exist on the server + */ private CompletableFuture lookupNamespace(final ExpandedNodeId nodeId) { LOG.trace("Expanded Node Id: {}", nodeId); @@ -452,7 +448,13 @@ private CompletableFuture lookupNamespace(final ExpandedNodeId nodeId) { if (uri != null) { LOG.trace("Looking up namespace: {}", uri); - return lookupNamespaceIndex(uri).thenApply(index -> toNodeId(index, nodeId)); + return lookupNamespaceIndex(uri).thenApply(index -> { + if (index == null) { + LOG.debug("The namespace URI of node {} is unknown to the server", nodeId.toParseableString()); + return null; + } + return toNodeId(index, nodeId); + }); } else { final UShort index = nodeId.getNamespaceIndex(); LOG.trace("Using provided index: {}", index); @@ -479,6 +481,11 @@ public CompletableFuture write(final ExpandedNodeId nodeId, final Da LOG.debug("Node - expanded: {}, full: {}", nodeId, node); + if (node == null) { + // answered like a node the server does not have + return completedFuture(new StatusCode(StatusCodes.Bad_NodeIdUnknown)); + } + return this.client.writeValuesAsync(List.of(node), List.of(value)) .thenApply(r -> r.get(0)) .whenComplete((status, error) -> { @@ -502,6 +509,14 @@ public CompletableFuture call( LOG.debug("Method - expanded: {}, full: {}", methodId, method); + if (node == null || method == null) { + // answered like the server: an unknown object is Bad_NodeIdUnknown, an unknown method Bad_MethodInvalid + final StatusCode status + = new StatusCode(node == null ? StatusCodes.Bad_NodeIdUnknown : StatusCodes.Bad_MethodInvalid); + return completedFuture( + new CallMethodResult(status, new StatusCode[0], new DiagnosticInfo[0], new Variant[0])); + } + final CallMethodRequest cmr = new CallMethodRequest(node, method, inputArguments); /* @@ -535,7 +550,23 @@ public CompletableFuture> readValues(List expand return CompletableFuture.allOf(nodeIdFutures).thenCompose(param -> { List nodeIds = Stream.of(nodeIdFutures).map(CompletableFuture::join).collect(Collectors.toList()); - return this.client.readValuesAsync(0, TimestampsToReturn.Server, nodeIds); + final List known = nodeIds.stream().filter(Objects::nonNull).collect(Collectors.toList()); + if (known.size() == nodeIds.size()) { + return this.client.readValuesAsync(0, TimestampsToReturn.Server, nodeIds); + } + // a node the server does not have gets the value the server returns for such a node + final CompletableFuture> read = known.isEmpty() + ? completedFuture(List.of()) : this.client.readValuesAsync(0, TimestampsToReturn.Server, known); + return read.thenApply(values -> { + final Iterator knownValues = values.iterator(); + final List result = new ArrayList<>(nodeIds.size()); + for (final NodeId nodeId : nodeIds) { + result.add(nodeId == null + ? new DataValue(new StatusCode(StatusCodes.Bad_NodeIdUnknown)) + : knownValues.next()); + } + return result; + }); }); } @@ -704,15 +735,33 @@ public CompletableFuture> browse( .thenCompose(__ -> { - final List nodeIds = Stream.of(futures).map(CompletableFuture::join) - .collect(Collectors.toList()); - - return completedFuture(nodeIds.stream().map(nodeId -> new BrowseDescription( - nodeId, direction, NodeIds.References, includeSubTypes, uint(nodeClasses), - uint(BrowseResultMask.All.getValue()))).collect(Collectors.toList())); - }) - - .thenCompose(descriptions -> browse(descriptions, depth, maxDepth, pattern, maxNodesPerRequest)); + // a node the server does not have gets the result the server returns for such a node + final Map unknown = new HashMap<>(); + final List descriptions = new ArrayList<>(); + for (int i = 0; i < futures.length; i++) { + final NodeId nodeId = futures[i].join(); + if (nodeId == null) { + unknown.put(expandedNodeIds.get(i), new BrowseResult( + new StatusCode(StatusCodes.Bad_NodeIdUnknown), ByteString.NULL_VALUE, + new ReferenceDescription[0])); + } else { + descriptions.add(new BrowseDescription( + nodeId, direction, NodeIds.References, includeSubTypes, uint(nodeClasses), + uint(BrowseResultMask.All.getValue()))); + } + } + if (unknown.isEmpty()) { + return browse(descriptions, depth, maxDepth, pattern, maxNodesPerRequest); + } + final CompletableFuture> knownResults = descriptions.isEmpty() + ? completedFuture(Map.of()) + : browse(descriptions, depth, maxDepth, pattern, maxNodesPerRequest); + return knownResults.thenApply(results -> { + final Map all = new HashMap<>(results); + all.putAll(unknown); + return all; + }); + }); } } diff --git a/components/camel-milo/src/test/java/org/apache/camel/component/milo/ProducerUnknownNamespaceTest.java b/components/camel-milo/src/test/java/org/apache/camel/component/milo/ProducerUnknownNamespaceTest.java new file mode 100644 index 0000000000000..1fbf3b06a9573 --- /dev/null +++ b/components/camel-milo/src/test/java/org/apache/camel/component/milo/ProducerUnknownNamespaceTest.java @@ -0,0 +1,211 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.milo; + +import java.util.List; +import java.util.Map; + +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.browse.MiloBrowseEndpoint; +import org.apache.camel.component.milo.client.MiloClientConnection; +import org.apache.camel.component.milo.client.MiloClientEndpoint; +import org.apache.camel.component.milo.server.MiloServerComponent; +import org.apache.camel.component.mock.MockEndpoint; +import org.eclipse.milo.opcua.sdk.client.OpcUaClient; +import org.eclipse.milo.opcua.stack.core.StatusCodes; +import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue; +import org.eclipse.milo.opcua.stack.core.types.builtin.ExpandedNodeId; +import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode; +import org.eclipse.milo.opcua.stack.core.types.structured.BrowseResult; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Testing producer requests with a node ID whose namespace URI is not in the namespace table of the server, which are + * answered like requests for a node that the server does not have + */ +public class ProducerUnknownNamespaceTest extends AbstractMiloServerTest { + + private static final String NODE = "nsu=" + MiloServerComponent.DEFAULT_NAMESPACE_URI + ";s=node1"; + + // namespace URIs that are not in the namespace table of the server + private static final String UNKNOWN_NAMESPACE_NODE = "nsu=urn:not:there;s=node1"; + + private static final String UNKNOWN_NAMESPACE_METHOD = "nsu=urn:not:there;s=method1"; + + private static final String ROOT_FOLDER = "ns=0;i=84"; + + private static final String SERVER = "opc.tcp://foo:bar@localhost:@@port@@" + + "?allowedSecurityPolicies=None&overrideHost=true"; + + // the client endpoints only differ in the node and the producer options, so they share one connection + private static final String MILO_CLIENT = "milo-client:" + SERVER + "&requestedPublishingInterval=100"; + + private static final String MILO_CLIENT_NODE = MILO_CLIENT + "&node=RAW(" + NODE + ")&samplingInterval=50"; + + private static final String MILO_CLIENT_AWAIT = MILO_CLIENT + "&defaultAwaitWrites=true"; + + private static final String MILO_CLIENT_WRITE = MILO_CLIENT_AWAIT + "&node=RAW(" + UNKNOWN_NAMESPACE_NODE + ")"; + + private static final String MILO_CLIENT_CALL + = MILO_CLIENT_AWAIT + "&node=RAW(" + NODE + ")&method=RAW(" + UNKNOWN_NAMESPACE_METHOD + ")"; + + private static final String MILO_CLIENT_CALL_UNKNOWN_OBJECT + = MILO_CLIENT_AWAIT + "&node=RAW(" + UNKNOWN_NAMESPACE_NODE + ")&method=RAW(" + UNKNOWN_NAMESPACE_METHOD + ")"; + + private static final String MILO_BROWSE = "milo-browse:" + SERVER; + + @EndpointInject("mock:consumer") + protected MockEndpoint consumerEndpoint; + + @Override + protected RoutesBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + // the consumer delivers the current value while it starts, so each test starts the route + from(resolve(MILO_CLIENT_NODE)).routeId("consumer").autoStartup(false).to("mock:consumer"); + } + }; + } + + @Test + void writeIsAnsweredWithBadNodeIdUnknown() throws Exception { + final OpcUaClient client = startConsumer(); + + final Exchange exchange = template.send(resolve(MILO_CLIENT_WRITE), e -> e.getMessage().setBody("Baz")); + + assertNull(exchange.getException()); + assertEquals(new StatusCode(StatusCodes.Bad_NodeIdUnknown), exchange.getMessage().getBody(StatusCode.class)); + assertConnectionKept(client); + } + + @Test + void readAnswersOnlyThatNodeWithBadNodeIdUnknown() throws Exception { + final OpcUaClient client = startConsumer(); + + final Exchange exchange = template.send(resolve(MILO_CLIENT_AWAIT), + e -> e.getMessage().setHeader(MiloConstants.HEADER_NODE_IDS, List.of(NODE, UNKNOWN_NAMESPACE_NODE))); + + assertNull(exchange.getException()); + final List values = exchange.getMessage().getBody(List.class); + assertNotNull(values); + assertEquals(2, values.size()); + assertGoodValue("Foo").accept((DataValue) values.get(0)); + assertEquals(new StatusCode(StatusCodes.Bad_NodeIdUnknown), ((DataValue) values.get(1)).getStatusCode()); + + // a read of only such a node sends no request: the server would answer an empty one with Bad_NothingToDo + final Exchange onlyUnknown = template.send(resolve(MILO_CLIENT_AWAIT), + e -> e.getMessage().setHeader(MiloConstants.HEADER_NODE_IDS, List.of(UNKNOWN_NAMESPACE_NODE))); + + assertNull(onlyUnknown.getException()); + assertEquals(new StatusCode(StatusCodes.Bad_NodeIdUnknown), + ((DataValue) onlyUnknown.getMessage().getBody(List.class).get(0)).getStatusCode()); + assertConnectionKept(client); + } + + @Test + void methodCallKeepsTheConnection() throws Exception { + final OpcUaClient client = startConsumer(); + + // an unknown method and an unknown object: the result is not checked, because the call guard drops every call + // result (CAMEL-25694) + assertNull(template.send(resolve(MILO_CLIENT_CALL), e -> e.getMessage().setBody(null)).getException()); + assertNull(template.send(resolve(MILO_CLIENT_CALL_UNKNOWN_OBJECT), e -> e.getMessage().setBody(null)) + .getException()); + assertConnectionKept(client); + } + + @Test + void browseAnswersOnlyThatNodeWithBadNodeIdUnknown() throws Exception { + // the browse component has a connection of its own + final MiloBrowseEndpoint endpoint = context.getEndpoint(resolve(MILO_BROWSE), MiloBrowseEndpoint.class); + final MiloClientConnection connection = endpoint.createConnection(); + try { + final OpcUaClient client = connection.getOpcUaClient(); + assertNotNull(client); + + final Exchange exchange = template.send(endpoint, e -> e.getMessage() + .setHeader(MiloConstants.HEADER_NODE_IDS, List.of(ROOT_FOLDER, UNKNOWN_NAMESPACE_NODE))); + + assertNull(exchange.getException()); + final Map results = exchange.getMessage().getBody(Map.class); + assertEquals(2, results.size()); + final BrowseResult unknown = (BrowseResult) results.get(ExpandedNodeId.parse(UNKNOWN_NAMESPACE_NODE)); + assertEquals(new StatusCode(StatusCodes.Bad_NodeIdUnknown), unknown.getStatusCode()); + assertEquals(0, unknown.getReferences().length); + final BrowseResult rootFolder = (BrowseResult) results.get(ExpandedNodeId.parse(ROOT_FOLDER)); + assertTrue(rootFolder.getStatusCode().isGood()); + assertTrue(rootFolder.getReferences().length > 0); + + // a browse of only such a node sends no request: the server would answer an empty one with Bad_NothingToDo + final Exchange onlyUnknown = template.send(endpoint, + e -> e.getMessage().setHeader(MiloConstants.HEADER_NODE_IDS, List.of(UNKNOWN_NAMESPACE_NODE))); + + assertNull(onlyUnknown.getException()); + assertEquals(new StatusCode(StatusCodes.Bad_NodeIdUnknown), ((BrowseResult) onlyUnknown.getMessage() + .getBody(Map.class).get(ExpandedNodeId.parse(UNKNOWN_NAMESPACE_NODE))).getStatusCode()); + + // the connection was not rebuilt + assertSame(client, connection.getOpcUaClient()); + } finally { + endpoint.releaseConnection(connection); + } + } + + private OpcUaClient startConsumer() throws Exception { + template.sendBody("milo-server:node1", "Foo"); + + consumerEndpoint.expectedMessageCount(1); + testBody(consumerEndpoint.message(0), assertGoodValue("Foo")); + context.getRouteController().startRoute("consumer"); + consumerEndpoint.assertIsSatisfied(); + + final OpcUaClient client = currentClient(); + assertNotNull(client); + return client; + } + + private void assertConnectionKept(final OpcUaClient client) throws Exception { + // the connection was not rebuilt and the consumer that shares it receives the next value + assertSame(client, currentClient()); + + consumerEndpoint.reset(); + consumerEndpoint.expectedMessageCount(1); + testBody(consumerEndpoint.message(0), assertGoodValue("Bar")); + template.sendBody("milo-server:node1", "Bar"); + consumerEndpoint.assertIsSatisfied(); + } + + private OpcUaClient currentClient() { + final MiloClientEndpoint endpoint = context.getEndpoint(resolve(MILO_CLIENT_NODE), MiloClientEndpoint.class); + final MiloClientConnection connection = endpoint.createConnection(); + try { + return connection.getOpcUaClient(); + } finally { + endpoint.releaseConnection(connection); + } + } +} 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 e11786cc9c922..e2b7ac95fc1b1 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 @@ -1556,6 +1556,15 @@ server, it is now created again in the existing session; previously the consumer connection was rebuilt for another reason. The monitored items deliver their current values again after the subscription is recreated. +=== camel-milo - a producer request for a node with an unknown namespace URI + +A `milo-client` write, method call or read, or a `milo-browse` browse, for a node ID with a namespace URI (`nsu=`) that +is not in the namespace table of the server no longer makes the client reconnect, which also recreated the +subscriptions of the consumers that share the connection. A write, read or browse answers the node like a node that +the server does not have, with the status code `Bad_NodeIdUnknown`, and the other nodes of a read or browse are not +affected. Previously an awaited write or read completed without an error and with a `null` body, and a browse failed +with a `NullPointerException`. + === 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