From 622709232975ce06cd6561f29ce3816d5d450969 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Tue, 20 Oct 2026 11:03:14 +0300 Subject: [PATCH 1/3] CAMEL-25692: camel-milo - create a refused monitored item again instead of dropping it A node whose monitored item the server refuses is still reported once with the status of the refusal. While the connection lasts, Camel now creates the refused items of the connection again every 10 seconds in one request, quietly, until the server accepts them; the first value of a created item is the current value of the node. This covers nodes and namespace URIs that the server gets later; the namespace table is read once per attempt. The attempt is skipped while the session is not active, a failed attempt is made again with the next one and never rebuilds the connection, and a stopped consumer is not retried. Co-authored-by: Claude Opus 5.5 Co-authored-by: Claude Fable 5.1 --- .../catalog/docs/milo-client-component.adoc | 9 +- .../src/main/docs/milo-client-component.adoc | 9 +- .../client/internal/SubscriptionManager.java | 138 +++++++++-- .../component/milo/CommunicationLossTest.java | 59 +++++ .../milo/MonitorItemLifecycleTest.java | 16 +- .../milo/MonitoredItemRetryTest.java | 218 ++++++++++++++++++ .../pages/camel-4x-upgrade-guide-4_24.adoc | 5 +- 7 files changed, 426 insertions(+), 28 deletions(-) create mode 100644 components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitoredItemRetryTest.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 3ba0ced7902eb..4f08ac6de8d6d 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 @@ -192,8 +192,13 @@ monitor filter (`monitorFilterType`, `dataChangeFilter*`). Tags that need a diff 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. +reported once with its status code, and Camel tries to create its monitored item again every 10 seconds while the +connection lasts, for all such nodes of the connection in one request. A failed attempt delivers nothing; once the item +is created, the node delivers its current value as its first value. This covers a node that is added to the server +later, for example with a new PLC program, and a server that refused the item because it had too many. A rebuilt +connection creates the monitored items of all nodes again, and a stopped consumer is not retried. +A node whose namespace URI (`nsu=`) is not in the namespace table of the server is reported and retried in the same +way, so a namespace that the server adds later is found. 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 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 3ba0ced7902eb..4f08ac6de8d6d 100644 --- a/components/camel-milo/src/main/docs/milo-client-component.adoc +++ b/components/camel-milo/src/main/docs/milo-client-component.adoc @@ -192,8 +192,13 @@ monitor filter (`monitorFilterType`, `dataChangeFilter*`). Tags that need a diff 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. +reported once with its status code, and Camel tries to create its monitored item again every 10 seconds while the +connection lasts, for all such nodes of the connection in one request. A failed attempt delivers nothing; once the item +is created, the node delivers its current value as its first value. This covers a node that is added to the server +later, for example with a new PLC program, and a server that refused the item because it had too many. A rebuilt +connection creates the monitored items of all nodes again, and a stopped consumer is not retried. +A node whose namespace URI (`nsu=`) is not in the namespace table of the server is reported and retried in the same +way, so a namespace that the server adds later is found. 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 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..75efa205dfaf5 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,8 +23,10 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.Iterator; import java.util.LinkedHashMap; +import java.util.LinkedHashSet; import java.util.LinkedList; import java.util.List; import java.util.Map; @@ -344,6 +346,17 @@ private class Connected { private final Map namespaceCache = new ConcurrentHashMap<>(); + /** + * The client handles of the nodes that the server refused to monitor on this connection, which are created + * again every {@code reconnectTimeout} milliseconds, guarded by {@code lock} + */ + private final Set refused = new LinkedHashSet<>(); + + /** + * The next attempt to create the refused nodes again, if one is scheduled, guarded by {@code lock} + */ + private Future retryJob; + Connected(final OpcUaClient client, final OpcUaSubscription manager, final SubscriptionListenerImpl listener) { this.client = client; this.manager = manager; @@ -352,15 +365,20 @@ private class Connected { } /** - * Creates the monitored items of the subscriptions on the server. + * Creates the monitored items of the subscriptions on the server. A node that the server refuses to monitor is + * created again on this connection every {@code reconnectTimeout} milliseconds, until its item is created or it + * is unregistered. * * @param subscriptions the subscriptions to create, by client handle + * @param retry whether this is an attempt to create refused nodes again: their refusal has been + * delivered already, so a node refused again is neither logged above DEBUG nor delivered * @return the bad status values of the nodes that could not be monitored, to be delivered to * their consumers after the lock of the manager has been released - * @throws UaException if the server rejected the CreateMonitoredItems service call, the connection has to be - * rebuilt + * @throws UaException if the server rejected the CreateMonitoredItems service call; unless {@code retry} is + * set, the connection has to be rebuilt */ - public List putSubscriptions(final Map subscriptions) throws Exception { + public List putSubscriptions(final Map subscriptions, final boolean retry) + throws Exception { final List pendingValues = new ArrayList<>(); @@ -373,18 +391,24 @@ public List putSubscriptions(final Map sub // Milo 1.0.5 API: create monitored items and set up listeners final List items = new ArrayList<>(subscriptions.size()); final Map itemToClientHandle = new HashMap<>(); + // the namespace URIs that the server does not have, read once per call: all nodes of a missing namespace + final Set unknownNamespaces = new HashSet<>(); for (final Map.Entry entry : subscriptions.entrySet()) { final Subscription s = entry.getValue(); - final NodeId node = lookupNamespace(s.getNodeId()).get(); + final String namespaceUri = s.getNodeId().getNamespaceUri(); + final NodeId node = namespaceUri != null && unknownNamespaces.contains(namespaceUri) + ? null : lookupNamespace(s.getNodeId()).get(); if (node == null) { - // 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, this.generation, new DataValue(status))); + if (namespaceUri != null) { + unknownNamespaces.add(namespaceUri); + } + // refused like a node the server rejects, the other nodes and the connection are not affected; the + // namespace table is read again on the next attempt, so a namespace added later is found + refuse(entry.getKey(), s, new StatusCode(StatusCodes.Bad_NodeIdUnknown), + "the namespace URI is unknown to the server", retry, pendingValues); } else { final ReadValueId itemId = new ReadValueId(node, AttributeId.Value.uid(), null, QualifiedName.NULL_VALUE); final OpcUaMonitoredItem item = new OpcUaMonitoredItem(itemId, MonitoringMode.Reporting); @@ -425,17 +449,18 @@ public List putSubscriptions(final Map sub if (result.isGood()) { this.goodSubscriptions.put(clientHandle, item); subscriptions.get(clientHandle).refused = false; + if (this.refused.remove(clientHandle) && retry) { + LOG.info("Monitored item for node {} created on retry", + subscriptions.get(clientHandle).getNodeId().toParseableString()); + } } else if (result.serviceResult().isBad()) { failedItems.add(item); rejectedServiceResult = result.serviceResult(); } else { - // reported once, the item is created again when the consumer restarts or the connection is rebuilt + // reported once, the item is created again later on this connection failedItems.add(item); - final Subscription s = subscriptions.get(clientHandle); - final StatusCode status = result.operationResult().orElse(result.serviceResult()); - LOG.warn("Monitored item for node {} could not be created: {}", s.getNodeId().toParseableString(), - status); - pendingValues.add(new PendingValue(s, this.generation, new DataValue(status))); + refuse(clientHandle, subscriptions.get(clientHandle), + result.operationResult().orElse(result.serviceResult()), null, retry, pendingValues); } } @@ -451,9 +476,42 @@ public List putSubscriptions(final Map sub } } + scheduleRetry(); return pendingValues; } + /** + * Records that the server refused to monitor a node. The first refusal is logged at WARN and delivered to the + * consumer, a refusal on a later attempt only logged at DEBUG. + */ + private void refuse( + final UInteger clientHandle, final Subscription s, final StatusCode status, final String reason, + final boolean retry, final List pendingValues) { + this.refused.add(clientHandle); + final String node = s.getNodeId().toParseableString(); + final String because = reason == null ? "" : ", " + reason; + if (retry) { + LOG.debug("Monitored item for node {} could not be created on retry{}: {}", node, because, status); + } else { + LOG.warn("Monitored item for node {} could not be created{}: {}", node, because, status); + pendingValues.add(new PendingValue(s, this.generation, new DataValue(status))); + } + } + + /** + * Schedules the next attempt to create the refused nodes again, unless there is none or one is scheduled. The + * timer runs on the reconnect thread, the attempt, which blocks until the server answers, on the executor of + * the connection, so that a slow attempt does not hold up a reconnect. + */ + private void scheduleRetry() { + if (this.refused.isEmpty() || this.retryJob != null || SubscriptionManager.this.disposed) { + return; + } + this.retryJob = SubscriptionManager.this.reconnectExecutor.schedule( + () -> SubscriptionManager.this.executor.execute(() -> retryRefused(this)), + SubscriptionManager.this.reconnectTimeout, TimeUnit.MILLISECONDS); + } + /** * Creates the subscription on the server again, with the monitored items it had, after the server lost it. * @@ -474,12 +532,13 @@ public List recreate(final Map subscriptio this.manager.deleteMonitoredItems(); this.manager.create(); - return putSubscriptions(monitored); + return putSubscriptions(monitored, false); } public void deactivate(final List clientHandles) throws Exception { final List items = new ArrayList<>(clientHandles.size()); for (final UInteger clientHandle : clientHandles) { + this.refused.remove(clientHandle); final OpcUaMonitoredItem item = this.goodSubscriptions.remove(clientHandle); if (item != null) { items.add(item); @@ -538,6 +597,10 @@ private CompletableFuture lookupNamespaceIndex(final String namespaceUri public void dispose() { // the session closed by the disconnect is no loss of communication this.listener.closed = true; + if (this.retryJob != null) { + this.retryJob.cancel(false); + this.retryJob = null; + } if (this.client != null) { try { this.client.disconnect(); @@ -1040,6 +1103,43 @@ private void recreateSubscription(final OpcUaSubscription subscription) { pendingValues.forEach(PendingValue::deliver); } + /** + * Creates the nodes that the server refused to monitor on a connection again, in one CreateMonitoredItems call, + * while the connection is the current one. Nothing is delivered to the consumers: a node created now delivers its + * first value from the server. While the session is not active, the attempt is skipped. A call that fails is made + * again with the next attempt and never rebuilds the connection, which would interrupt the nodes that are + * monitored: a lost session or subscription is recovered without it. + */ + private void retryRefused(final Connected connected) { + lock.lock(); + try { + connected.retryJob = null; + if (this.disposed || this.connected != connected || connected.refused.isEmpty()) { + return; + } + if (!connected.listener.sessionActive) { + LOG.debug("Session not active, the refused monitored items are created again later"); + connected.scheduleRetry(); + return; + } + final Map refused = new LinkedHashMap<>(); + for (final UInteger clientHandle : connected.refused) { + final Subscription subscription = this.subscriptions.get(clientHandle); + if (subscription != null) { + refused.put(clientHandle, subscription); + } + } + try { + connected.putSubscriptions(refused, true); + } catch (final Exception e) { + LOG.debug("Failed to create the refused monitored items again, trying again later", e); + connected.scheduleRetry(); + } + } finally { + lock.unlock(); + } + } + private boolean isCurrent(final OpcUaSubscription subscription) { return !this.disposed && this.connected != null && this.connected.manager == subscription; } @@ -1163,7 +1263,7 @@ private void performAndEvalConnect() { try { LOG.debug("Setting subscriptions: {}", this.subscriptions.size()); - pendingValues.addAll(connected.putSubscriptions(this.subscriptions)); + pendingValues.addAll(connected.putSubscriptions(this.subscriptions, false)); LOG.debug("Update state : {} -> {}", this.connected, connected); final Connected oldConnected = this.connected; @@ -1504,7 +1604,7 @@ public List registerItems( reportLoss(this.connected.manager, this.connected.generation, subscriptions); } - whenConnected(connected -> pendingValues.addAll(connected.putSubscriptions(subscriptions))); + whenConnected(connected -> pendingValues.addAll(connected.putSubscriptions(subscriptions, false))); } finally { lock.unlock(); } diff --git a/components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java b/components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java index 958faad13dd47..0b928e6b6e9e7 100644 --- a/components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java +++ b/components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java @@ -23,6 +23,8 @@ import java.net.Socket; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; import org.apache.camel.EndpointInject; import org.apache.camel.Produce; @@ -40,6 +42,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; +import static org.awaitility.Awaitility.await; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertSame; @@ -195,6 +198,62 @@ void lostSessionReportsTheNodeAndResendsItsValue() throws Exception { } } + @Test + void refusedNodeIsCreatedOnceTheSessionIsActiveAgain() throws Exception { + producer1.sendBody("Foo"); + + item1Endpoint.expectedMessageCount(1); + testBody(item1Endpoint.message(0), assertGoodValue("Foo")); + context.getRouteController().startRoute("item1"); + item1Endpoint.assertIsSatisfied(); + + unknownEndpoint.expectedMessageCount(1); + testBody(unknownEndpoint.message(0), + value -> assertEquals(StatusCodes.Bad_NodeIdUnknown, value.getStatusCode().getValue())); + context.getRouteController().startRoute("unknown"); + unknownEndpoint.assertIsSatisfied(); + + final MiloClientEndpoint endpoint = context.getEndpoint( + resolve(MILO_CLIENT_ITEM_1).replace("@@relay@@", Integer.toString(relayPort.getPort())), + MiloClientEndpoint.class); + final MiloClientConnection connection = endpoint.createConnection(); + try { + final OpcUaClient client = connection.getOpcUaClient(); + final NodeId sessionId = client.getSession().getSessionId(); + + // the refused node is not reported, and delivers its value once it is created after the session is back + unknownEndpoint.reset(); + unknownEndpoint.expectedMessageCount(1); + unknownEndpoint.setResultWaitTime(60_000); + unknownEndpoint.setAssertPeriod(2_000); + testBody(unknownEndpoint.message(0), assertGoodValue("Baz")); + item1Endpoint.reset(); + item1Endpoint.expectedMessageCount(2); + item1Endpoint.setResultWaitTime(60_000); + item1Endpoint.setAssertPeriod(2_000); + testBody(item1Endpoint.message(0), assertNoCommunication()); + testBody(item1Endpoint.message(1), assertGoodValue("Foo")); + + // the server stays unreachable for longer than the retry interval of the refused node + relay.stop(); + await().atMost(10, TimeUnit.SECONDS).until(() -> item1Endpoint.getReceivedCounter() >= 1); + await().pollDelay(12, TimeUnit.SECONDS).atMost(13, TimeUnit.SECONDS).until(() -> true); + // an attempt while the session is lost is skipped: it does not hold the connection until the session is back + assertSame(client, CompletableFuture.supplyAsync(connection::getOpcUaClient).get(2, TimeUnit.SECONDS)); + template.sendBody("milo-server:unknownItem", "Baz"); + relay.start(relayPort.getPort(), getServerPort()); + + unknownEndpoint.assertIsSatisfied(); + item1Endpoint.assertIsSatisfied(); + + // the attempts while the session was lost did not rebuild the connection + assertSame(client, connection.getOpcUaClient()); + assertEquals(sessionId, client.getSession().getSessionId()); + } finally { + endpoint.releaseConnection(connection); + } + } + @Test void consumerStartedWithoutTheServerReportsTheNodeThenReceivesItsValue() throws Exception { producer1.sendBody("Foo"); diff --git a/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorItemLifecycleTest.java b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorItemLifecycleTest.java index a770a93c20d87..c67ec3365ac30 100644 --- a/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorItemLifecycleTest.java +++ b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorItemLifecycleTest.java @@ -35,6 +35,7 @@ import org.eclipse.milo.opcua.sdk.client.subscriptions.OpcUaSubscription; import org.eclipse.milo.opcua.sdk.server.OpcUaServer; import org.eclipse.milo.opcua.sdk.server.Session; +import org.eclipse.milo.opcua.sdk.server.nodes.UaNode; 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; @@ -402,8 +403,8 @@ void subscriptionThatASlowRouteHeldUpIsKept() throws Exception { } @Test - @Timeout(value = 60, unit = TimeUnit.SECONDS) - void nodeThatCannotBeMonitoredAgainIsReportedWithoutRebuildingTheConnection() throws Exception { + @Timeout(value = 90, unit = TimeUnit.SECONDS) + void nodeThatCannotBeMonitoredAgainIsReportedAndCreatedOnceItIsBackWithoutRebuildingTheConnection() throws Exception { producer1.sendBody("Foo"); producer2.sendBody("Bar"); context.getRouteController().startRoute("item1"); @@ -420,7 +421,9 @@ void nodeThatCannotBeMonitoredAgainIsReportedWithoutRebuildingTheConnection() th final UInteger subscriptionId = subscription.getSubscriptionId().orElseThrow(); // the second node is removed on the server before the subscription times out there - getServer().getAddressSpaceManager().getManagedNode(ExpandedNodeId.parse(NODE_ITEM_2)).orElseThrow().delete(); + final UaNode node2 = getServer().getAddressSpaceManager().getManagedNode(ExpandedNodeId.parse(NODE_ITEM_2)) + .orElseThrow(); + node2.delete(); timeOutSubscription(subscriptionId); // the subscription is recreated in the same session, with the monitored items of the first node only @@ -441,6 +444,13 @@ && new StatusCode(StatusCodes.Bad_NodeIdUnknown) .untilAsserted(() -> assertGoodValue("Baz").accept(lastExchange(item1Endpoint).getMessage() .getBody(DataValue.class))); assertSame(client, connection.getOpcUaClient()); + + // the node is added again on the server: its item is created again and delivers its current value + node2.getNodeManager().addNode(node2); + await().atMost(30, TimeUnit.SECONDS) + .untilAsserted(() -> assertGoodValue("Bar").accept(lastExchange(items12Endpoint, NODE_ITEM_2).getMessage() + .getBody(DataValue.class))); + assertSame(client, connection.getOpcUaClient()); } finally { endpoint.releaseConnection(connection); } diff --git a/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitoredItemRetryTest.java b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitoredItemRetryTest.java new file mode 100644 index 0000000000000..2aaa23a2f327b --- /dev/null +++ b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitoredItemRetryTest.java @@ -0,0 +1,218 @@ +/* + * 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.concurrent.TimeUnit; +import java.util.function.Consumer; + +import org.apache.camel.EndpointInject; +import org.apache.camel.Produce; +import org.apache.camel.ProducerTemplate; +import org.apache.camel.RoutesBuilder; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.milo.client.MiloClientConnection; +import org.apache.camel.component.milo.client.MiloClientEndpoint; +import org.apache.camel.component.milo.server.MiloServerComponent; +import org.apache.camel.component.milo.server.internal.CamelNamespace; +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.StatusCode; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; + +/** + * A node that the server refuses to monitor is reported once and created again on the same connection, every 10 + * seconds, until the server has it or its consumer stops + */ +public class MonitoredItemRetryTest extends AbstractMiloServerTest { + + private static final String DIRECT_START_1 = "direct:start1"; + + private static final String MILO_SERVER_ITEM_1 = "milo-server:myitem1"; + + // a namespace that the server does not have until a test adds it + private static final String LATE_NAMESPACE_URI = "urn:org:apache:camel:late"; + + // the client endpoints only differ in the node, so they share one connection + private static final String MILO_CLIENT = "milo-client:opc.tcp://foo:bar@localhost:@@port@@" + + "?allowedSecurityPolicies=None&overrideHost=true" + + "&samplingInterval=50&requestedPublishingInterval=100&node="; + + private static final String MILO_CLIENT_ITEM_1 + = MILO_CLIENT + NodeIds.nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "myitem1"); + + private static final String MILO_CLIENT_LATE_ITEM + = MILO_CLIENT + NodeIds.nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "lateItem"); + + private static final String MILO_CLIENT_LATE_NAMESPACE_ITEM + = MILO_CLIENT + NodeIds.nodeValue(LATE_NAMESPACE_URI, "lateNamespaceItem"); + + private static final String MILO_CLIENT_STOPPED_ITEM + = MILO_CLIENT + NodeIds.nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "stoppedItem"); + + @EndpointInject("mock:item1") + protected MockEndpoint item1Endpoint; + + @EndpointInject("mock:lateItem") + protected MockEndpoint lateItemEndpoint; + + @EndpointInject("mock:lateNamespaceItem") + protected MockEndpoint lateNamespaceItemEndpoint; + + @EndpointInject("mock:stoppedItem") + protected MockEndpoint stoppedItemEndpoint; + + @Produce(DIRECT_START_1) + protected ProducerTemplate producer1; + + @Override + protected RoutesBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from(DIRECT_START_1).to(MILO_SERVER_ITEM_1); + + // the consumers deliver the current values while they start, so the tests start these routes + from(resolve(MILO_CLIENT_ITEM_1)).routeId("item1").autoStartup(false).to("mock:item1"); + from(resolve(MILO_CLIENT_LATE_ITEM)).routeId("lateItem").autoStartup(false).to("mock:lateItem"); + from(resolve(MILO_CLIENT_LATE_NAMESPACE_ITEM)).routeId("lateNamespaceItem").autoStartup(false) + .to("mock:lateNamespaceItem"); + from(resolve(MILO_CLIENT_STOPPED_ITEM)).routeId("stoppedItem").autoStartup(false) + .to("mock:stoppedItem"); + } + }; + } + + @Test + @Timeout(value = 60, unit = TimeUnit.SECONDS) + void nodeCreatedOnTheServerLaterDeliversItsValue() throws Exception { + lateItemEndpoint.expectedMessageCount(1); + testBody(lateItemEndpoint.message(0), assertStatus(StatusCodes.Bad_NodeIdUnknown)); + context.getRouteController().startRoute("lateItem"); + lateItemEndpoint.assertIsSatisfied(); + + final MiloClientEndpoint endpoint = context.getEndpoint(resolve(MILO_CLIENT_LATE_ITEM), MiloClientEndpoint.class); + final MiloClientConnection connection = endpoint.createConnection(); + try { + final OpcUaClient client = connection.getOpcUaClient(); + + // the node is created on the server: its first value follows the refusal, without another bad status + lateItemEndpoint.reset(); + lateItemEndpoint.expectedMessageCount(1); + lateItemEndpoint.setResultWaitTime(30_000); + lateItemEndpoint.setAssertPeriod(2_000); + testBody(lateItemEndpoint.message(0), assertGoodValue("Foo")); + template.sendBody("milo-server:lateItem", "Foo"); + lateItemEndpoint.assertIsSatisfied(); + + // on the same connection + assertSame(client, connection.getOpcUaClient()); + } finally { + endpoint.releaseConnection(connection); + } + } + + @Test + @Timeout(value = 60, unit = TimeUnit.SECONDS) + void nodeOfANamespaceAddedLaterDeliversItsValue() throws Exception { + lateNamespaceItemEndpoint.expectedMessageCount(1); + testBody(lateNamespaceItemEndpoint.message(0), assertStatus(StatusCodes.Bad_NodeIdUnknown)); + context.getRouteController().startRoute("lateNamespaceItem"); + lateNamespaceItemEndpoint.assertIsSatisfied(); + + final MiloClientEndpoint endpoint + = context.getEndpoint(resolve(MILO_CLIENT_LATE_NAMESPACE_ITEM), MiloClientEndpoint.class); + final MiloClientConnection connection = endpoint.createConnection(); + // the server adds the namespace URI to its namespace table and the node to the namespace + final CamelNamespace namespace = new CamelNamespace(LATE_NAMESPACE_URI, getServer().getServer()); + try { + final OpcUaClient client = connection.getOpcUaClient(); + + lateNamespaceItemEndpoint.reset(); + lateNamespaceItemEndpoint.expectedMessageCount(1); + lateNamespaceItemEndpoint.setResultWaitTime(30_000); + lateNamespaceItemEndpoint.setAssertPeriod(2_000); + testBody(lateNamespaceItemEndpoint.message(0), assertGoodValue("Foo")); + namespace.startup(); + namespace.getOrAddItem("lateNamespaceItem").update("Foo"); + lateNamespaceItemEndpoint.assertIsSatisfied(); + + assertSame(client, connection.getOpcUaClient()); + } finally { + endpoint.releaseConnection(connection); + namespace.shutdown(); + } + } + + @Test + @Timeout(value = 60, unit = TimeUnit.SECONDS) + void stoppedConsumerIsNotCreatedAgain() throws Exception { + // the consumer of another node keeps the connection open + producer1.sendBody("Foo"); + item1Endpoint.expectedMessageCount(1); + testBody(item1Endpoint.message(0), assertGoodValue("Foo")); + context.getRouteController().startRoute("item1"); + item1Endpoint.assertIsSatisfied(); + + stoppedItemEndpoint.expectedMessageCount(1); + testBody(stoppedItemEndpoint.message(0), assertStatus(StatusCodes.Bad_NodeIdUnknown)); + context.getRouteController().startRoute("stoppedItem"); + stoppedItemEndpoint.assertIsSatisfied(); + + final MiloClientEndpoint endpoint = context.getEndpoint(resolve(MILO_CLIENT_ITEM_1), MiloClientEndpoint.class); + final MiloClientConnection connection = endpoint.createConnection(); + try { + final OpcUaClient client = connection.getOpcUaClient(); + + context.getRouteController().stopRoute("stoppedItem"); + stoppedItemEndpoint.reset(); + stoppedItemEndpoint.expectedMessageCount(0); + template.sendBody("milo-server:stoppedItem", "Foo"); + + // the next attempt would have created it by now + await().during(15, TimeUnit.SECONDS).atMost(20, TimeUnit.SECONDS) + .until(() -> client.getSubscriptions().get(0).getMonitoredItems().stream() + .noneMatch(item -> "stoppedItem" + .equals(item.getReadValueId().getNodeId().getIdentifier()))); + stoppedItemEndpoint.assertIsSatisfied(); + assertSame(client, connection.getOpcUaClient()); + + // the other consumer keeps receiving + item1Endpoint.reset(); + item1Endpoint.expectedMessageCount(1); + testBody(item1Endpoint.message(0), assertGoodValue("Bar")); + producer1.sendBody("Bar"); + item1Endpoint.assertIsSatisfied(); + } finally { + endpoint.releaseConnection(connection); + } + } + + private MiloServerComponent getServer() { + return context.getComponent("milo-server", MiloServerComponent.class); + } + + private static Consumer assertStatus(final long statusCode) { + return value -> assertEquals(new StatusCode(statusCode), value.getStatusCode()); + } +} 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 3fda251c39ed4..c188339981e4c 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 @@ -1970,8 +1970,9 @@ Bad status of a failed sensor. Since 4.16, a `milo-client` consumer whose monitored item could not be created, for example because the node does not exist, received nothing. Now it receives one `DataValue` with the bad status code. The -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. +monitored item is then created again every 10 seconds on the same connection, without delivering anything for a failed +attempt, until the server accepts it; its first value is the current value of the node. A route restart or a +connection that Camel rebuilds creates it again too. 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 From c92824359d3207e6e7761db78f7c6a2f57acfd9a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Wed, 21 Oct 2026 10:40:43 +0300 Subject: [PATCH 2/3] CAMEL-25692: camel-milo - cancel a retry under the lock and name the retry interval in the warning Review of the retry of refused monitored items. The next attempt of a connection is cancelled under the lock that guards it wherever the connection is replaced or disposed, also in the dispose of the manager, which disposes the connection itself outside of the lock. The warning for a refused node says that Camel tries again and how often. The javadoc of retryRefused says that the producers wait for an attempt and why that wait is bounded. CommunicationLossTest waits with during() and checks that the refused node is not reported while the server is unreachable, instead of sleeping. Co-authored-by: Claude Opus 5.5 --- .../client/internal/SubscriptionManager.java | 31 ++++++++++++++++--- .../component/milo/CommunicationLossTest.java | 3 +- 2 files changed, 28 insertions(+), 6 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 75efa205dfaf5..f9a8bd32676aa 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 @@ -493,7 +493,8 @@ private void refuse( if (retry) { LOG.debug("Monitored item for node {} could not be created on retry{}: {}", node, because, status); } else { - LOG.warn("Monitored item for node {} could not be created{}: {}", node, because, status); + LOG.warn("Monitored item for node {} could not be created{}, trying again every {} ms: {}", node, because, + SubscriptionManager.this.reconnectTimeout, status); pendingValues.add(new PendingValue(s, this.generation, new DataValue(status))); } } @@ -512,6 +513,17 @@ private void scheduleRetry() { SubscriptionManager.this.reconnectTimeout, TimeUnit.MILLISECONDS); } + /** + * Cancels the next attempt to create the refused nodes again, if one is scheduled: a connection that is + * replaced or disposed is not retried + */ + private void cancelRetry() { + if (this.retryJob != null) { + this.retryJob.cancel(false); + this.retryJob = null; + } + } + /** * Creates the subscription on the server again, with the monitored items it had, after the server lost it. * @@ -597,10 +609,6 @@ private CompletableFuture lookupNamespaceIndex(final String namespaceUri public void dispose() { // the session closed by the disconnect is no loss of communication this.listener.closed = true; - if (this.retryJob != null) { - this.retryJob.cancel(false); - this.retryJob = null; - } if (this.client != null) { try { this.client.disconnect(); @@ -1109,6 +1117,12 @@ private void recreateSubscription(final OpcUaSubscription subscription) { * first value from the server. While the session is not active, the attempt is skipped. A call that fails is made * again with the next attempt and never rebuilds the connection, which would interrupt the nodes that are * monitored: a lost session or subscription is recovered without it. + *

+ * Like the registration of a consumer, the attempt holds {@code lock} until the server answers, so that a consumer + * that stops or a subscription that is recreated in the meantime cannot interleave with it, and the producers and + * {@link #getOpcUaClient()} wait for it. The wait is bounded: the attempt is only sent while the session is active, + * a call in flight when the connection drops silently fails once the keep-alive of the client declares the session + * lost, and a call that the server does not answer fails after the request timeout. */ private void retryRefused(final Connected connected) { lock.lock(); @@ -1214,6 +1228,7 @@ private void handleConnectionFailure(final Throwable e) { if (this.connected != null) { // the nodes lose their communication with the connection, after the values it still delivers reportLoss(this.connected.manager, this.connected.generation, this.subscriptions); + this.connected.cancelRetry(); this.connected.dispose(); this.connected = null; } @@ -1272,11 +1287,13 @@ private void performAndEvalConnect() { if (oldConnected != null) { LOG.debug("Dispose old state"); + oldConnected.cancelRetry(); oldConnected.dispose(); } } catch (final Exception e) { LOG.info("Failed to set subscriptions", e); + connected.cancelRetry(); connected.dispose(); throw e; } @@ -1436,6 +1453,10 @@ public void dispose() { } this.disposed = true; connected = this.connected; + if (connected != null) { + // the next attempt is cancelled under the lock that guards it, the connection is disposed outside of it + connected.cancelRetry(); + } if (this.reconnectJob != null) { // a waiting reconnect would keep the thread of the reconnect executor until it is due this.reconnectJob.cancel(false); diff --git a/components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java b/components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java index 0b928e6b6e9e7..d89133882a330 100644 --- a/components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java +++ b/components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java @@ -237,7 +237,8 @@ void refusedNodeIsCreatedOnceTheSessionIsActiveAgain() throws Exception { // the server stays unreachable for longer than the retry interval of the refused node relay.stop(); await().atMost(10, TimeUnit.SECONDS).until(() -> item1Endpoint.getReceivedCounter() >= 1); - await().pollDelay(12, TimeUnit.SECONDS).atMost(13, TimeUnit.SECONDS).until(() -> true); + await().during(12, TimeUnit.SECONDS).atMost(15, TimeUnit.SECONDS) + .until(() -> unknownEndpoint.getReceivedCounter() == 0); // an attempt while the session is lost is skipped: it does not hold the connection until the session is back assertSame(client, CompletableFuture.supplyAsync(connection::getOpcUaClient).get(2, TimeUnit.SECONDS)); template.sendBody("milo-server:unknownItem", "Baz"); From c68d3a1c9f7d7b3feea3e48651e5e156998e666c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Wed, 21 Oct 2026 13:50:05 +0300 Subject: [PATCH 3/3] CAMEL-25692: camel-milo - deliver a refusal only while the node is still refused The refusals of a registration or of a recreated subscription are delivered outside the lock, one node after another. When the next attempt, or a new connection, created a node before its refusal was delivered, the route received the first value of the node and then its refusal, and the node stayed marked as refused, so a later loss of communication was no longer reported for it. The node is now marked as refused under the lock once the call that refused it has succeeded, and its refusal is delivered only while that still holds. A call that fails leaves the node unmarked, so that it still receives the loss report of the connection. Co-authored-by: Claude Opus 5.5 --- .../client/internal/SubscriptionManager.java | 35 +++++++---- .../milo/MonitoredItemRetryTest.java | 60 +++++++++++++++++-- 2 files changed, 78 insertions(+), 17 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 f9a8bd32676aa..1ea704e01576c 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 @@ -257,7 +257,8 @@ private static class Subscription { /** * Whether the server refused to monitor the node: its status is that refusal, whatever the communication, until - * its item is created + * its item is created. Set under the lock of the manager when a call hands the refusal out for delivery, and + * cleared under it when a call creates the item. */ private volatile boolean refused; @@ -295,11 +296,13 @@ synchronized void deliver(final long generation, final DataValue value) { } /** - * Delivers the status with which the server refused to monitor the node. + * Delivers the status with which the server refused to monitor the node, unless its item has been created + * since: the node then delivers its values instead. */ - void refuse(final long generation, final DataValue status) { - this.refused = true; - deliver(generation, status); + synchronized void refuse(final long generation, final DataValue status) { + if (this.refused) { + deliver(generation, status); + } } /** @@ -328,7 +331,8 @@ public MonitoringFilter createMonitoringFilter() { } /** - * A status value for a node that could not be monitored, delivered to its consumer outside of the lock + * A status value for a node that could not be monitored, delivered to its consumer outside of the lock while the + * node is still refused */ private record PendingValue(Subscription subscription, long generation, DataValue value) { void deliver() { @@ -370,8 +374,9 @@ private class Connected { * is unregistered. * * @param subscriptions the subscriptions to create, by client handle - * @param retry whether this is an attempt to create refused nodes again: their refusal has been - * delivered already, so a node refused again is neither logged above DEBUG nor delivered + * @param retry whether this is an attempt to create refused nodes again: their refusal has been queued + * for delivery already, so a node refused again is neither logged above DEBUG nor + * delivered * @return the bad status values of the nodes that could not be monitored, to be delivered to * their consumers after the lock of the manager has been released * @throws UaException if the server rejected the CreateMonitoredItems service call; unless {@code retry} is @@ -476,13 +481,17 @@ public List putSubscriptions(final Map sub } } + // marked under the lock once the call has succeeded: a refusal that the creation of its item overtakes, by + // an attempt or a new connection, is then dropped, and a call that fails leaves the nodes to the loss report + // of the connection + pendingValues.forEach(pending -> pending.subscription().refused = true); scheduleRetry(); return pendingValues; } /** - * Records that the server refused to monitor a node. The first refusal is logged at WARN and delivered to the - * consumer, a refusal on a later attempt only logged at DEBUG. + * Records that the server refused to monitor a node. The first refusal is logged at WARN and queued for + * delivery to the consumer, a refusal on a later attempt only logged at DEBUG. */ private void refuse( final UInteger clientHandle, final Subscription s, final StatusCode status, final String reason, @@ -1114,9 +1123,9 @@ private void recreateSubscription(final OpcUaSubscription subscription) { /** * Creates the nodes that the server refused to monitor on a connection again, in one CreateMonitoredItems call, * while the connection is the current one. Nothing is delivered to the consumers: a node created now delivers its - * first value from the server. While the session is not active, the attempt is skipped. A call that fails is made - * again with the next attempt and never rebuilds the connection, which would interrupt the nodes that are - * monitored: a lost session or subscription is recovered without it. + * first value from the server, and its refusal, if not delivered yet, is dropped. While the session is not active, + * the attempt is skipped. A call that fails is made again with the next attempt and never rebuilds the connection, + * which would interrupt the nodes that are monitored: a lost session or subscription is recovered without it. *

* Like the registration of a consumer, the attempt holds {@code lock} until the server answers, so that a consumer * that stops or a subscription that is recreated in the meantime cannot interleave with it, and the producers and diff --git a/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitoredItemRetryTest.java b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitoredItemRetryTest.java index 2aaa23a2f327b..468d4b918042e 100644 --- a/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitoredItemRetryTest.java +++ b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitoredItemRetryTest.java @@ -16,10 +16,12 @@ */ package org.apache.camel.component.milo; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import org.apache.camel.EndpointInject; +import org.apache.camel.Exchange; import org.apache.camel.Produce; import org.apache.camel.ProducerTemplate; import org.apache.camel.RoutesBuilder; @@ -53,10 +55,20 @@ public class MonitoredItemRetryTest extends AbstractMiloServerTest { // a namespace that the server does not have until a test adds it private static final String LATE_NAMESPACE_URI = "urn:org:apache:camel:late"; - // the client endpoints only differ in the node, so they share one connection - private static final String MILO_CLIENT = "milo-client:opc.tcp://foo:bar@localhost:@@port@@" - + "?allowedSecurityPolicies=None&overrideHost=true" - + "&samplingInterval=50&requestedPublishingInterval=100&node="; + // the client endpoints only differ in their nodes, so they share one connection + private static final String MILO_CLIENT_BASE = "milo-client:opc.tcp://foo:bar@localhost:@@port@@" + + "?allowedSecurityPolicies=None&overrideHost=true" + + "&samplingInterval=50&requestedPublishingInterval=100"; + + private static final String MILO_CLIENT = MILO_CLIENT_BASE + "&node="; + + // a node that the server never has, and one that it gets while the refusal of the first is being delivered + private static final String MISSING_NODE = "nsu=" + MiloServerComponent.DEFAULT_NAMESPACE_URI + ";s=missingItem"; + + private static final String OVERTAKEN_NODE = "nsu=" + MiloServerComponent.DEFAULT_NAMESPACE_URI + ";s=overtakenItem"; + + private static final String MILO_CLIENT_OVERTAKEN + = MILO_CLIENT_BASE + "&nodes=RAW(" + MISSING_NODE + "," + OVERTAKEN_NODE + ")"; private static final String MILO_CLIENT_ITEM_1 = MILO_CLIENT + NodeIds.nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "myitem1"); @@ -82,6 +94,11 @@ public class MonitoredItemRetryTest extends AbstractMiloServerTest { @EndpointInject("mock:stoppedItem") protected MockEndpoint stoppedItemEndpoint; + @EndpointInject("mock:overtaken") + protected MockEndpoint overtakenEndpoint; + + private final CountDownLatch overtakenCreated = new CountDownLatch(1); + @Produce(DIRECT_START_1) protected ProducerTemplate producer1; @@ -99,6 +116,8 @@ public void configure() { .to("mock:lateNamespaceItem"); from(resolve(MILO_CLIENT_STOPPED_ITEM)).routeId("stoppedItem").autoStartup(false) .to("mock:stoppedItem"); + from(resolve(MILO_CLIENT_OVERTAKEN)).routeId("overtaken").autoStartup(false) + .process(exchange -> holdRefusalOfMissingNode(exchange)).to("mock:overtaken"); } }; } @@ -208,6 +227,39 @@ void stoppedConsumerIsNotCreatedAgain() throws Exception { } } + @Test + @Timeout(value = 60, unit = TimeUnit.SECONDS) + void refusalOvertakenByTheNextAttemptIsNotDelivered() throws Exception { + // the missing node delivers its refusal, the other node its first value and no refusal after it + overtakenEndpoint.expectedHeaderValuesReceivedInAnyOrder(MiloConstants.HEADER_NODE_ID, MISSING_NODE, + OVERTAKEN_NODE); + overtakenEndpoint.setAssertPeriod(2_000); + context.getRouteController().startRoute("overtaken"); + overtakenEndpoint.assertIsSatisfied(); + + for (final Exchange exchange : overtakenEndpoint.getReceivedExchanges()) { + final DataValue value = exchange.getMessage().getBody(DataValue.class); + if (OVERTAKEN_NODE.equals(exchange.getMessage().getHeader(MiloConstants.HEADER_NODE_ID))) { + assertGoodValue("Foo").accept(value); + } else { + assertStatus(StatusCodes.Bad_NodeIdUnknown).accept(value); + } + } + } + + /** + * Holds the refusal of the missing node, which the consumer delivers first while it starts, until the server has + * the other node and the next attempt has created it, so that the first value of that node overtakes its refusal + */ + private void holdRefusalOfMissingNode(final Exchange exchange) throws InterruptedException { + if (MISSING_NODE.equals(exchange.getMessage().getHeader(MiloConstants.HEADER_NODE_ID))) { + template.sendBody("milo-server:overtakenItem", "Foo"); + overtakenCreated.await(30, TimeUnit.SECONDS); + } else if (exchange.getMessage().getBody(DataValue.class).getStatusCode().isGood()) { + overtakenCreated.countDown(); + } + } + private MiloServerComponent getServer() { return context.getComponent("milo-server", MiloServerComponent.class); }