CAMEL-25671: camel-milo - report a node with an unknown namespace URI instead of failing the connection

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/milo-client-component.adoc

@@ -149,6 +149,7 @@ need a consumer of their own.
 
 A node that does not exist on the server does not affect the other nodes. A node that cannot be monitored is
 reported once with its status code and is not retried; restarting the route or a rebuilt connection to the server creates the monitored items again.
+A node whose namespace URI (`nsu=`) is not in the namespace table of the server is reported in the same way.
 The consumer delivers one exchange with the bad status code for such a node, and the `CamelMiloNodeId` header names it.
 
 === Status codes and null values

components/camel-milo/src/main/docs/milo-client-component.adoc

@@ -149,6 +149,7 @@ need a consumer of their own.
 
 A node that does not exist on the server does not affect the other nodes. A node that cannot be monitored is
 reported once with its status code and is not retried; restarting the route or a rebuilt connection to the server creates the monitored items again.
+A node whose namespace URI (`nsu=`) is not in the namespace table of the server is reported in the same way.
 The consumer delivers one exchange with the bad status code for such a node, and the `CamelMiloNodeId` header names it.
 
 === Status codes and null values

components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java

@@ -177,8 +177,6 @@ private class Connected {
         private OpcUaClient client;
         private final OpcUaSubscription manager;
 
-        private final Map<UInteger, Subscription> badSubscriptions = new HashMap<>();
-
         private final Map<UInteger, OpcUaMonitoredItem> goodSubscriptions = new HashMap<>();
 
         private final Map<String, UShort> namespaceCache = new ConcurrentHashMap<>();
@@ -214,10 +212,14 @@ public List<PendingValue> putSubscriptions(final Map<UInteger, Subscription> sub
             for (final Map.Entry<UInteger, Subscription> entry : subscriptions.entrySet()) {
                 final Subscription s = entry.getValue();
 
-                final NodeId node = lookupNamespace(s.getNodeId()).get();
+                final NodeId node = lookupMonitoredNode(s.getNodeId());
 
                 if (node == null) {
-                    handleSubscriptionError(new StatusCode(StatusCodes.Bad_InvalidArgument), entry.getKey(), s, pendingValues);
+                    // reported once like a node the server rejects, the other nodes and the connection are not affected
+                    final StatusCode status = new StatusCode(StatusCodes.Bad_NodeIdUnknown);
+                    LOG.warn("Monitored item for node {} could not be created, the namespace URI is unknown to the server: {}",
+                            s.getNodeId().toParseableString(), status);
+                    pendingValues.add(new PendingValue(s.getValueConsumer(), new DataValue(status)));
                 } else {
                     final ReadValueId itemId = new ReadValueId(node, AttributeId.Value.uid(), null, QualifiedName.NULL_VALUE);
                     final OpcUaMonitoredItem item = new OpcUaMonitoredItem(itemId, MonitoringMode.Reporting);
@@ -283,29 +285,17 @@ public List<PendingValue> putSubscriptions(final Map<UInteger, Subscription> sub
                 }
             }
 
-            if (!this.badSubscriptions.isEmpty()) {
-                SubscriptionManager.this.executor.schedule(this::resubscribe, SubscriptionManager.this.reconnectTimeout,
-                        TimeUnit.MILLISECONDS);
-            }
-
             return pendingValues;
         }
 
-        private void handleSubscriptionError(
-                final StatusCode statusCode, final UInteger clientHandle, final Subscription s,
-                final List<PendingValue> pendingValues) {
-            this.badSubscriptions.put(clientHandle, s);
-            pendingValues.add(new PendingValue(s.getValueConsumer(), new DataValue(statusCode)));
-        }
-
-        private void resubscribe() {
-            final Map<UInteger, Subscription> subscriptions = new HashMap<>(this.badSubscriptions);
-            this.badSubscriptions.clear();
-            try {
-                putSubscriptions(subscriptions).forEach(PendingValue::deliver);
-            } catch (final Exception e) {
-                handleConnectionFailure(e);
-            }
+        /**
+         * @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);
         }
 
         public void deactivate(final List<UInteger> clientHandles) throws Exception {
@@ -314,8 +304,6 @@ public void deactivate(final List<UInteger> clientHandles) throws Exception {
                 final OpcUaMonitoredItem item = this.goodSubscriptions.remove(clientHandle);
                 if (item != null) {
                     items.add(item);
-                } else {
-                    this.badSubscriptions.remove(clientHandle);
                 }
             }
             if (!items.isEmpty()) {

components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorNodesTest.java

@@ -52,6 +52,9 @@ public class MonitorNodesTest extends AbstractMiloServerTest {
 
     private static final String UNKNOWN_NODE = "nsu=" + MiloServerComponent.DEFAULT_NAMESPACE_URI + ";s=unknownNode";
 
+    // a namespace URI that is not in the namespace table of the server
+    private static final String UNKNOWN_NAMESPACE_NODE = "nsu=urn:not:there;s=node1";
+
     // the same node as NODE_1 with the default server index spelled out
     private static final String NODE_1_WITH_SERVER_INDEX = "svr=0;" + NODE_1;
 
@@ -65,6 +68,9 @@ public class MonitorNodesTest extends AbstractMiloServerTest {
     private static final String MILO_CLIENT_UNKNOWN_NODE
             = MILO_CLIENT + "&nodes=RAW(" + NODE_1 + "," + UNKNOWN_NODE + "," + NODE_2 + ")";
 
+    private static final String MILO_CLIENT_UNKNOWN_NAMESPACE
+            = MILO_CLIENT + "&nodes=RAW(" + NODE_1 + "," + UNKNOWN_NAMESPACE_NODE + "," + NODE_2 + ")";
+
     private static final String MILO_CLIENT_DUPLICATES
             = MILO_CLIENT + "&nodes=RAW( " + NODE_1 + " ,, " + NODE_2 + " ," + NODE_1 + ")";
 
@@ -79,6 +85,9 @@ public class MonitorNodesTest extends AbstractMiloServerTest {
     @EndpointInject("mock:unknownNode")
     protected MockEndpoint unknownNodeEndpoint;
 
+    @EndpointInject("mock:unknownNamespace")
+    protected MockEndpoint unknownNamespaceEndpoint;
+
     @EndpointInject("mock:duplicates")
     protected MockEndpoint duplicatesEndpoint;
 
@@ -96,6 +105,8 @@ public void configure() {
                 // the consumers deliver the current values while they start, so each test starts its route
                 from(resolve(MILO_CLIENT_NODES)).routeId("nodes").autoStartup(false).to("mock:nodes");
                 from(resolve(MILO_CLIENT_UNKNOWN_NODE)).routeId("unknownNode").autoStartup(false).to("mock:unknownNode");
+                from(resolve(MILO_CLIENT_UNKNOWN_NAMESPACE)).routeId("unknownNamespace").autoStartup(false)
+                        .to("mock:unknownNamespace");
                 from(resolve(MILO_CLIENT_DUPLICATES)).routeId("duplicates").autoStartup(false).to("mock:duplicates");
                 from(resolve(MILO_CLIENT_NODE)).routeId("node").autoStartup(false).to("mock:node");
                 from(resolve(MILO_CLIENT_SAME_NODE)).routeId("sameNode").autoStartup(false).to("mock:sameNode");
@@ -171,6 +182,30 @@ void notGoodStatusWithNullValueIsDeliveredByDefault() throws Exception {
         unknownNodeEndpoint.assertIsSatisfied();
     }
 
+    @Test
+    void unknownNamespaceUriReceivesBadStatusAndTheOtherNodesKeepWorking() throws Exception {
+        writeServerValues();
+
+        // the bad status is reported once, the connection stays up for the other nodes
+        unknownNamespaceEndpoint.expectedMessageCount(3);
+        unknownNamespaceEndpoint.setAssertPeriod(1000);
+        context.getRouteController().startRoute("unknownNamespace");
+        unknownNamespaceEndpoint.assertIsSatisfied();
+
+        final Map<String, DataValue> values = valuesByNode(unknownNamespaceEndpoint);
+        assertEquals(Set.of(NODE_1, UNKNOWN_NAMESPACE_NODE, NODE_2), values.keySet());
+        assertEquals(new StatusCode(StatusCodes.Bad_NodeIdUnknown), values.get(UNKNOWN_NAMESPACE_NODE).getStatusCode());
+        assertGoodValue("Foo").accept(values.get(NODE_1));
+        assertGoodValue("Bar").accept(values.get(NODE_2));
+
+        unknownNamespaceEndpoint.reset();
+        unknownNamespaceEndpoint.expectedMessageCount(1);
+        unknownNamespaceEndpoint.expectedHeaderReceived(MiloConstants.HEADER_NODE_ID, NODE_1);
+        testBody(unknownNamespaceEndpoint.message(0), assertGoodValue("Baz"));
+        template.sendBody("milo-server:node1", "Baz");
+        unknownNamespaceEndpoint.assertIsSatisfied();
+    }
+
     @Test
     void duplicateNodesAreMonitoredOnce() throws Exception {
         writeServerValues();

docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc

@@ -1534,6 +1534,10 @@ exist, received nothing. Now it receives one `DataValue` with the bad status cod
 monitored item is created again when the route is restarted or when Camel rebuilds the connection to the server; unlike
 Camel 3.x, it is not retried every 10 seconds.
 
+A node ID with a namespace URI (`nsu=`) that is not in the namespace table of the server no longer takes the whole
+client connection down, including the consumers of other nodes. Its consumer receives one `DataValue` with
+`Bad_NodeIdUnknown`, like a node that does not exist.
+
 The consumer now sets the `CamelMiloNodeId` header on every exchange, including a consumer with a single `node`.
 
 Stopping a consumer now deletes its monitored items. Previously they stayed on the server while another endpoint kept