From 675dec46fa2f11211bc539946bcae724e690d994 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Thu, 15 Oct 2026 20:00:17 +0300 Subject: [PATCH 1/5] CAMEL-25691: camel-milo - recreate the subscription when it stops publishing When the server stops publishing for the subscription (watchdog) or reports Bad_Timeout, the consumer deletes and creates the subscription again inside the existing session and creates its monitored items again, on the manager's executor under the connection lock. If that fails, the existing connection rebuild takes over. Previously the watchdog callback was ignored and Bad_Timeout was only logged, so the consumer stopped receiving values until the connection was rebuilt for another reason. Co-authored-by: Claude Fable 5.1 --- .../catalog/docs/milo-client-component.adoc | 8 ++ .../src/main/docs/milo-client-component.adoc | 8 ++ .../client/internal/SubscriptionManager.java | 52 +++++++++++- .../milo/MonitorItemLifecycleTest.java | 85 ++++++++++++++++++- .../pages/camel-4x-upgrade-guide-4_24.adoc | 4 + 5 files changed, 154 insertions(+), 3 deletions(-) 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 bf6f7bd1252b6..1d338dcb8efc3 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 @@ -165,6 +165,14 @@ The default `omitNullValues=true` also omits a null value with a Good status tha does not learn that the node recovered. Set `omitNullValues=false` for a node whose value can be null, such as a `String`. +=== Subscriptions that stop publishing + +The consumers of a connection share one subscription on the server. When the server stops publishing for it (no +notification or keep-alive within about one and a half keep-alive intervals) or reports that it timed out +(`Bad_Timeout`), Camel deletes and creates the subscription again in the existing session; if that fails, it rebuilds +the connection to the server. After the subscription is recreated, every monitored item delivers its current value +again, and value changes in between are not delivered. + === Method ID It is possible to perform methods calls on OPC UA nodes. If the parameter `method` is set to the Node ID of a method call (the node ID must be set to the parent object in this case), 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 bf6f7bd1252b6..1d338dcb8efc3 100644 --- a/components/camel-milo/src/main/docs/milo-client-component.adoc +++ b/components/camel-milo/src/main/docs/milo-client-component.adoc @@ -165,6 +165,14 @@ The default `omitNullValues=true` also omits a null value with a Good status tha does not learn that the node recovered. Set `omitNullValues=false` for a node whose value can be null, such as a `String`. +=== Subscriptions that stop publishing + +The consumers of a connection share one subscription on the server. When the server stops publishing for it (no +notification or keep-alive within about one and a half keep-alive intervals) or reports that it timed out +(`Bad_Timeout`), Camel deletes and creates the subscription again in the existing session; if that fails, it rebuilds +the connection to the server. After the subscription is recreated, every monitored item delivers its current value +again, and value changes in between are not delivered. + === Method ID It is possible to perform methods calls on OPC UA nodes. If the parameter `method` is set to the Node ID of a method call (the node ID must be set to the parent object in this case), 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 ed096e646d49b..2f298b7037d32 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 @@ -28,6 +28,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; @@ -103,7 +104,20 @@ private final class SubscriptionListenerImpl implements OpcUaSubscription.Subscr @Override public void onStatusChanged(final OpcUaSubscription subscription, final StatusCode status) { - LOG.info("Subscription status changed {} : {}", subscription.getSubscriptionId(), status); + if (status.getValue() == StatusCodes.Bad_Timeout) { + final Optional id = subscription.getSubscriptionId(); + LOG.warn("Subscription timed out on the server, recreating it"); + executor.execute(() -> recreateSubscription(subscription, id)); + } else { + LOG.info("Subscription status changed {} : {}", subscription.getSubscriptionId(), status); + } + } + + @Override + public void onWatchdogTimerElapsed(final OpcUaSubscription subscription) { + final Optional id = subscription.getSubscriptionId(); + LOG.warn("Subscription {} stopped publishing, recreating it", id.orElse(null)); + executor.execute(() -> recreateSubscription(subscription, id)); } @Override @@ -676,6 +690,42 @@ public OpcUaClient getOpcUaClient() { } } + /** + * Deletes and creates the subscription of the current connection again, inside the existing session, and creates + * its monitored items again. Does nothing if the subscription no longer belongs to the current connection or has + * already been recreated since the given id was observed. Falls back to rebuilding the connection if this fails. + *

+ * The service calls run while holding {@code lock}, like {@link #registerItems} and {@link #unregisterItems} do, + * because the {@link OpcUaSubscription} is shared with them and is not thread-safe. A request the server does not + * answer fails after the client's request timeout ({@code requestTimeout}, 60 s by default), so the lock is held + * for at most one timeout per call before the fallback runs. + */ + private void recreateSubscription(final OpcUaSubscription subscription, final Optional id) { + lock.lock(); + try { + if (this.disposed || this.connected == null || this.connected.manager != subscription + || !subscription.getSubscriptionId().equals(id)) { + LOG.debug("Subscription {} is stale or already recreated", id.orElse(null)); + return; + } + try { + subscription.delete(); + } catch (final UaException e) { + LOG.debug("Failed to delete subscription {}", id.orElse(null), e); + subscription.reset(); + } + subscription.create(); + if (!subscription.createMonitoredItems().stream().allMatch(MonitoredItemServiceOperationResult::isGood)) { + throw new UaException(StatusCodes.Bad_UnexpectedError, "Failed to create the monitored items again"); + } + LOG.info("Subscription recreated as {}", subscription.getSubscriptionId().orElse(null)); + } catch (final Exception e) { + handleConnectionFailure(e); + } finally { + lock.unlock(); + } + } + private void handleConnectionFailure(final Throwable e) { lock.lock(); try { 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 cec8822d6eb0f..f84f8fa9579d4 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 @@ -31,6 +31,9 @@ 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.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.stack.core.StatusCodes; import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue; import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode; @@ -39,13 +42,16 @@ import org.junit.jupiter.api.Timeout; import static org.awaitility.Awaitility.await; +import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; 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 the monitored items of consumers sharing a connection, when a consumer is restarted, its node does not exist - * or the server rejects the CreateMonitoredItems service call + * Testing the monitored items of consumers sharing a connection, when a consumer is restarted, its node does not exist, + * the server rejects the CreateMonitoredItems service call or the subscription stops publishing */ public class MonitorItemLifecycleTest extends AbstractMiloServerTest { @@ -264,6 +270,81 @@ void failedCreateMonitoredItemsCallRebuildsTheConnection() throws Exception { .getBody(DataValue.class))); } + @Test + @Timeout(value = 60, unit = TimeUnit.SECONDS) + void subscriptionThatStopsPublishingIsRecreatedInTheSameSession() throws Exception { + producer1.sendBody("Foo"); + + item1Endpoint.expectedMinimumMessageCount(1); + testBody(item1Endpoint.message(0), assertGoodValue("Foo")); + context.getRouteController().startRoute("item1"); + item1Endpoint.assertIsSatisfied(); + + final MiloClientEndpoint endpoint = context.getEndpoint(resolve(MILO_CLIENT_ITEM_1), MiloClientEndpoint.class); + final MiloClientConnection connection = endpoint.createConnection(); + try { + final OpcUaClient client = connection.getOpcUaClient(); + final OpcUaSubscription subscription = client.getSubscriptions().get(0); + // shortens the watchdog of the subscription to a few publishing intervals + subscription.setMaxKeepAliveCount(uint(1)); + subscription.setLifetimeCount(uint(3)); + subscription.modify(); + + // the server deletes the subscription and answers the pending publish requests without one + final UInteger subscriptionId = subscription.getSubscriptionId().orElseThrow(); + assertTrue(client.deleteSubscriptions(List.of(subscriptionId)).getResults()[0].isGood()); + + producer1.sendBody("Bar"); + await().atMost(30, TimeUnit.SECONDS) + .untilAsserted(() -> assertGoodValue("Bar").accept(lastExchange(item1Endpoint).getMessage() + .getBody(DataValue.class))); + + assertSame(client, connection.getOpcUaClient()); + assertNotEquals(subscriptionId, subscription.getSubscriptionId().orElseThrow()); + } finally { + endpoint.releaseConnection(connection); + } + } + + @Test + @Timeout(value = 60, unit = TimeUnit.SECONDS) + void subscriptionThatTimesOutOnTheServerIsRecreatedInTheSameSession() throws Exception { + producer1.sendBody("Foo"); + + item1Endpoint.expectedMinimumMessageCount(1); + testBody(item1Endpoint.message(0), assertGoodValue("Foo")); + context.getRouteController().startRoute("item1"); + item1Endpoint.assertIsSatisfied(); + + final MiloClientEndpoint endpoint = context.getEndpoint(resolve(MILO_CLIENT_ITEM_1), MiloClientEndpoint.class); + final MiloClientConnection connection = endpoint.createConnection(); + try { + final OpcUaClient client = connection.getOpcUaClient(); + final UInteger subscriptionId = client.getSubscriptions().get(0).getSubscriptionId().orElseThrow(); + + // the server reports the lifetime of the subscription as expired and removes it + final OpcUaServer server = context.getComponent("milo-server", MiloServerComponent.class).getServer(); + for (final Session session : server.getSessionManager().getAllSessions()) { + final var subscriptions = session.getSubscriptionManager(); + final var subscription = subscriptions.getSubscription(subscriptionId); + if (subscription != null) { + subscriptions.sendStatusChangeNotification(subscription, new StatusCode(StatusCodes.Bad_Timeout)); + subscriptions.removeSubscription(subscriptionId); + } + } + + producer1.sendBody("Bar"); + await().atMost(30, TimeUnit.SECONDS) + .untilAsserted(() -> assertGoodValue("Bar").accept(lastExchange(item1Endpoint).getMessage() + .getBody(DataValue.class))); + + assertSame(client, connection.getOpcUaClient()); + assertNotEquals(subscriptionId, client.getSubscriptions().get(0).getSubscriptionId().orElseThrow()); + } finally { + endpoint.releaseConnection(connection); + } + } + private static Exchange lastExchange(final MockEndpoint endpoint) { final List exchanges = endpoint.getReceivedExchanges(); return exchanges.get(exchanges.size() - 1); 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 1e5463becddd1..e552385b1f23b 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 @@ -1544,6 +1544,10 @@ Stopping a consumer now deletes its monitored items. Previously they stayed on t the shared connection open, so a restarted consumer received each value twice. While the server is unreachable, the stop waits until the client's next attempt to reconnect has failed. +A consumer whose subscription stops publishing, or times out on the server, now recreates the subscription in the +existing session; previously it stopped receiving values until the connection was rebuilt for another reason. The +monitored items deliver their current values again after the subscription is recreated. + === 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 26050d4780a0245c979446ca4c24c396b2707af7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Thu, 15 Oct 2026 22:56:02 +0300 Subject: [PATCH 2/5] CAMEL-25691: camel-milo - recreate the subscription after a confirmed delete The watchdog and Bad_Timeout now delete the subscription asynchronously, without holding the connection lock, and create it again in the session only once the server has confirmed that it is gone. If the delete fails because the connection or the session was lost, Milo's session recovery takes over and arms the watchdog again when publishing resumes; any other failure rebuilds the connection, which closes the session together with the subscription. The monitored items are created again through putSubscriptions, so a node that can no longer be monitored is reported to its consumer instead of rebuilding the connection. Co-Authored-By: Claude Opus 5.5 --- .../catalog/docs/milo-client-component.adoc | 16 ++- .../src/main/docs/milo-client-component.adoc | 16 ++- .../client/internal/SubscriptionManager.java | 112 +++++++++++++----- .../milo/MonitorItemLifecycleTest.java | 75 ++++++++++-- 4 files changed, 175 insertions(+), 44 deletions(-) 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 1d338dcb8efc3..167ae0ad43c70 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 @@ -169,9 +169,19 @@ does not learn that the node recovered. Set `omitNullValues=false` for a node wh The consumers of a connection share one subscription on the server. When the server stops publishing for it (no notification or keep-alive within about one and a half keep-alive intervals) or reports that it timed out -(`Bad_Timeout`), Camel deletes and creates the subscription again in the existing session; if that fails, it rebuilds -the connection to the server. After the subscription is recreated, every monitored item delivers its current value -again, and value changes in between are not delivered. +(`Bad_Timeout`), Camel deletes the subscription and, once the server has confirmed that it is gone, creates it again +in the existing session. A node that can no longer be monitored is reported as described above and does not affect +the other nodes. + +A network that drops silently can stop the publishing before the client notices that the connection is lost. The delete +then fails with the connection, and the client restores the session once the server can be reached again: the +subscription resumes publishing, or it is created again when the server reports that it timed out in the meantime. If +the network is back before the client has noticed the loss, the delete reaches the server and the subscription is +created again. + +Any other failure of the delete, or a subscription that cannot be created again, rebuilds the connection to the server. +After the subscription is recreated, every monitored item delivers its current value again, and value changes in +between are not delivered. === Method ID 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 1d338dcb8efc3..167ae0ad43c70 100644 --- a/components/camel-milo/src/main/docs/milo-client-component.adoc +++ b/components/camel-milo/src/main/docs/milo-client-component.adoc @@ -169,9 +169,19 @@ does not learn that the node recovered. Set `omitNullValues=false` for a node wh The consumers of a connection share one subscription on the server. When the server stops publishing for it (no notification or keep-alive within about one and a half keep-alive intervals) or reports that it timed out -(`Bad_Timeout`), Camel deletes and creates the subscription again in the existing session; if that fails, it rebuilds -the connection to the server. After the subscription is recreated, every monitored item delivers its current value -again, and value changes in between are not delivered. +(`Bad_Timeout`), Camel deletes the subscription and, once the server has confirmed that it is gone, creates it again +in the existing session. A node that can no longer be monitored is reported as described above and does not affect +the other nodes. + +A network that drops silently can stop the publishing before the client notices that the connection is lost. The delete +then fails with the connection, and the client restores the session once the server can be reached again: the +subscription resumes publishing, or it is created again when the server reports that it timed out in the meantime. If +the network is back before the client has noticed the loss, the delete reaches the server and the subscription is +created again. + +Any other failure of the delete, or a subscription that cannot be created again, rebuilds the connection to the server. +After the subscription is recreated, every monitored item delivers its current value again, and value changes in +between are not delivered. === Method ID 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 2f298b7037d32..39b3c0f4bc83c 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 @@ -28,7 +28,6 @@ import java.util.List; import java.util.Map; import java.util.Objects; -import java.util.Optional; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; @@ -98,6 +97,14 @@ public class SubscriptionManager { private static final Logger LOG = LoggerFactory.getLogger(SubscriptionManager.class); + /** + * The status codes of a request that failed because the connection or the session was lost, which the session + * recovery of the client deals with + */ + private static final Set SESSION_LOST = Set.of( + StatusCodes.Bad_ConnectionClosed, StatusCodes.Bad_SessionClosed, StatusCodes.Bad_SessionIdInvalid, + StatusCodes.Bad_SessionNotActivated); + private final AtomicLong clientHandleCounter = new AtomicLong(); private final class SubscriptionListenerImpl implements OpcUaSubscription.SubscriptionListener { @@ -105,9 +112,8 @@ private final class SubscriptionListenerImpl implements OpcUaSubscription.Subscr @Override public void onStatusChanged(final OpcUaSubscription subscription, final StatusCode status) { if (status.getValue() == StatusCodes.Bad_Timeout) { - final Optional id = subscription.getSubscriptionId(); LOG.warn("Subscription timed out on the server, recreating it"); - executor.execute(() -> recreateSubscription(subscription, id)); + recoverSubscription(subscription); } else { LOG.info("Subscription status changed {} : {}", subscription.getSubscriptionId(), status); } @@ -115,9 +121,8 @@ public void onStatusChanged(final OpcUaSubscription subscription, final StatusCo @Override public void onWatchdogTimerElapsed(final OpcUaSubscription subscription) { - final Optional id = subscription.getSubscriptionId(); - LOG.warn("Subscription {} stopped publishing, recreating it", id.orElse(null)); - executor.execute(() -> recreateSubscription(subscription, id)); + LOG.warn("Subscription {} stopped publishing, recreating it", subscription.getSubscriptionId().orElse(null)); + recoverSubscription(subscription); } @Override @@ -312,6 +317,29 @@ private NodeId lookupMonitoredNode(final ExpandedNodeId nodeId) throws Exception 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. + * + * @param subscriptions the registered subscriptions, by client handle + * @return the bad status values of the nodes that could not be monitored again, to be delivered + * to their consumers after the lock of the manager has been released + * @throws Exception if the subscription could not be created or the server rejected the + * CreateMonitoredItems service call, the connection has to be rebuilt + */ + public List recreate(final Map subscriptions) throws Exception { + final Map monitored = new HashMap<>(); + this.goodSubscriptions.keySet() + .forEach(clientHandle -> monitored.put(clientHandle, subscriptions.get(clientHandle))); + this.goodSubscriptions.clear(); + + // the monitored items were reset together with the subscription, so they are dropped without a service call + this.manager.removeMonitoredItems(this.manager.getMonitoredItems()); + this.manager.deleteMonitoredItems(); + + this.manager.create(); + return putSubscriptions(monitored); + } + public void deactivate(final List clientHandles) throws Exception { final List items = new ArrayList<>(clientHandles.size()); for (final UInteger clientHandle : clientHandles) { @@ -691,39 +719,69 @@ public OpcUaClient getOpcUaClient() { } /** - * Deletes and creates the subscription of the current connection again, inside the existing session, and creates - * its monitored items again. Does nothing if the subscription no longer belongs to the current connection or has - * already been recreated since the given id was observed. Falls back to rebuilding the connection if this fails. + * Recovers the subscription of a connection after it stopped publishing or timed out on the server. + *

+ * The subscription is deleted first, and it is only created again, inside the existing session, once the server has + * confirmed that it no longer exists; one that the client has already reset because it timed out is deleted without + * a service call. If the delete fails because the connection or the session was lost, the session recovery of the + * client takes over: it resumes the subscription, which arms its watchdog again, or reports that the subscription + * could not be transferred. Any other failure rebuilds the connection, which closes the session together with its + * subscriptions. *

- * The service calls run while holding {@code lock}, like {@link #registerItems} and {@link #unregisterItems} do, - * because the {@link OpcUaSubscription} is shared with them and is not thread-safe. A request the server does not - * answer fails after the client's request timeout ({@code requestTimeout}, 60 s by default), so the lock is held - * for at most one timeout per call before the fallback runs. + * The delete neither blocks nor holds {@code lock}: a request sent before the connection was lost only fails once + * the client notices the loss, which its keep-alive does on the threads of {@link #executor}. + */ + private void recoverSubscription(final OpcUaSubscription subscription) { + subscription.deleteAsync().whenCompleteAsync((unit, e) -> { + final StatusCode status = e == null + ? StatusCode.GOOD + : UaException.extractStatusCode(e).orElse(new StatusCode(StatusCodes.Bad_UnexpectedError)); + if (status.isGood() || status.getValue() == StatusCodes.Bad_SubscriptionIdInvalid) { + recreateSubscription(subscription); + } else if (SESSION_LOST.contains(status.getValue())) { + LOG.info("Session lost while deleting subscription {}, leaving it to the session recovery: {}", + subscription.getSubscriptionId().orElse(null), status); + } else { + lock.lock(); + try { + if (isCurrent(subscription)) { + handleConnectionFailure(e); + } + } finally { + lock.unlock(); + } + } + }, this.executor); + } + + /** + * Creates the subscription of the current connection again, inside the existing session, together with the + * monitored items it had. Does nothing if the subscription no longer belongs to the current connection or has been + * created again already. A node that cannot be monitored again is reported to its consumer, as when the consumer + * starts; if the subscription itself cannot be created again, the connection is rebuilt. */ - private void recreateSubscription(final OpcUaSubscription subscription, final Optional id) { + private void recreateSubscription(final OpcUaSubscription subscription) { + final List pendingValues = new ArrayList<>(); lock.lock(); try { - if (this.disposed || this.connected == null || this.connected.manager != subscription - || !subscription.getSubscriptionId().equals(id)) { - LOG.debug("Subscription {} is stale or already recreated", id.orElse(null)); + if (!isCurrent(subscription) || subscription.getSubscriptionId().isPresent()) { + LOG.debug("Subscription is stale or already recreated"); return; } - try { - subscription.delete(); - } catch (final UaException e) { - LOG.debug("Failed to delete subscription {}", id.orElse(null), e); - subscription.reset(); - } - subscription.create(); - if (!subscription.createMonitoredItems().stream().allMatch(MonitoredItemServiceOperationResult::isGood)) { - throw new UaException(StatusCodes.Bad_UnexpectedError, "Failed to create the monitored items again"); - } + pendingValues.addAll(this.connected.recreate(this.subscriptions)); LOG.info("Subscription recreated as {}", subscription.getSubscriptionId().orElse(null)); } catch (final Exception e) { handleConnectionFailure(e); } finally { lock.unlock(); } + + // deliver outside of the lock, the routes may use the connection + pendingValues.forEach(PendingValue::deliver); + } + + private boolean isCurrent(final OpcUaSubscription subscription) { + return !this.disposed && this.connected != null && this.connected.manager == subscription; } private void handleConnectionFailure(final Throwable e) { 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 f84f8fa9579d4..29189560ec2ba 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 @@ -36,6 +36,7 @@ import org.eclipse.milo.opcua.sdk.server.Session; 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.builtin.unsigned.UInteger; import org.junit.jupiter.api.Test; @@ -51,7 +52,7 @@ /** * Testing the monitored items of consumers sharing a connection, when a consumer is restarted, its node does not exist, - * the server rejects the CreateMonitoredItems service call or the subscription stops publishing + * the server rejects the CreateMonitoredItems service call or the subscription stops publishing or times out */ public class MonitorItemLifecycleTest extends AbstractMiloServerTest { @@ -322,16 +323,7 @@ void subscriptionThatTimesOutOnTheServerIsRecreatedInTheSameSession() throws Exc final OpcUaClient client = connection.getOpcUaClient(); final UInteger subscriptionId = client.getSubscriptions().get(0).getSubscriptionId().orElseThrow(); - // the server reports the lifetime of the subscription as expired and removes it - final OpcUaServer server = context.getComponent("milo-server", MiloServerComponent.class).getServer(); - for (final Session session : server.getSessionManager().getAllSessions()) { - final var subscriptions = session.getSubscriptionManager(); - final var subscription = subscriptions.getSubscription(subscriptionId); - if (subscription != null) { - subscriptions.sendStatusChangeNotification(subscription, new StatusCode(StatusCodes.Bad_Timeout)); - subscriptions.removeSubscription(subscriptionId); - } - } + timeOutSubscription(subscriptionId); producer1.sendBody("Bar"); await().atMost(30, TimeUnit.SECONDS) @@ -345,6 +337,67 @@ void subscriptionThatTimesOutOnTheServerIsRecreatedInTheSameSession() throws Exc } } + @Test + @Timeout(value = 60, unit = TimeUnit.SECONDS) + void nodeThatCannotBeMonitoredAgainIsReportedWithoutRebuildingTheConnection() throws Exception { + producer1.sendBody("Foo"); + producer2.sendBody("Bar"); + context.getRouteController().startRoute("item1"); + + items12Endpoint.expectedMessageCount(2); + context.getRouteController().startRoute("items12"); + items12Endpoint.assertIsSatisfied(); + + final MiloClientEndpoint endpoint = context.getEndpoint(resolve(MILO_CLIENT_ITEM_1), MiloClientEndpoint.class); + final MiloClientConnection connection = endpoint.createConnection(); + try { + final OpcUaClient client = connection.getOpcUaClient(); + final OpcUaSubscription subscription = client.getSubscriptions().get(0); + 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(); + timeOutSubscription(subscriptionId); + + // the subscription is recreated in the same session, with the monitored items of the first node only + await().atMost(30, TimeUnit.SECONDS) + .until(() -> subscription.getSubscriptionId().filter(id -> !id.equals(subscriptionId)).isPresent()); + assertSame(client, connection.getOpcUaClient()); + assertEquals(2, subscription.getMonitoredItems().size()); + + // the consumer of the second node is told that it is gone, the consumers of the first node keep receiving + await().atMost(10, TimeUnit.SECONDS).until(() -> items12Endpoint.getReceivedExchanges().stream() + .anyMatch(exchange -> NODE_ITEM_2.equals(exchange.getMessage().getHeader(MiloConstants.HEADER_NODE_ID)) + && new StatusCode(StatusCodes.Bad_NodeIdUnknown) + .equals(exchange.getMessage().getBody(DataValue.class).getStatusCode()))); + producer1.sendBody("Baz"); + await().atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> assertGoodValue("Baz").accept(lastExchange(item1Endpoint).getMessage() + .getBody(DataValue.class))); + assertSame(client, connection.getOpcUaClient()); + } finally { + endpoint.releaseConnection(connection); + } + } + + private OpcUaServer getServer() { + return context.getComponent("milo-server", MiloServerComponent.class).getServer(); + } + + /** + * The server reports the lifetime of the subscription as expired and removes it + */ + private void timeOutSubscription(final UInteger subscriptionId) { + for (final Session session : getServer().getSessionManager().getAllSessions()) { + final var subscriptions = session.getSubscriptionManager(); + final var subscription = subscriptions.getSubscription(subscriptionId); + if (subscription != null) { + subscriptions.sendStatusChangeNotification(subscription, new StatusCode(StatusCodes.Bad_Timeout)); + subscriptions.removeSubscription(subscriptionId); + } + } + } + private static Exchange lastExchange(final MockEndpoint endpoint) { final List exchanges = endpoint.getReceivedExchanges(); return exchanges.get(exchanges.size() - 1); From a00e7f67e73aa24cc67312cc5c7f34b19b43753a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Serdar=20G=C3=B6kay?= Date: Fri, 16 Oct 2026 19:15:59 +0300 Subject: [PATCH 3/5] CAMEL-25691: camel-milo - keep a subscription that the server still has after its watchdog elapses The watchdog also elapses while a route is slow to process a value: the client replaces each outstanding publish request only after its notifications have been processed, so the server has none left to answer. The subscription was then recreated although the server still had it, and every monitored item delivered its current value again. After the watchdog, a SetPublishingMode request that enables publishing now checks the subscription first; it is always created with publishing enabled, so the request does not change its settings. The outcome: - confirmed by the server: the subscription is kept; - already reset by the client: nothing, since every reset is followed by its own recovery; - failed because the connection or the session was lost: left to the session recovery of the client; - anything else: delete and recreate, as before. A Bad_Timeout status change from the server still recreates the subscription directly, without the check. A request that is sent after the client noticed a lost connection, for example the delete of a watchdog callback that waited behind a slow route, waits for the next reconnect attempt and, if that attempt fails, fails with its error, which was not taken for a lost session: the delete then either rebuilt the connection or removed the subscription that the reconnect had resumed and created it again. The listener of the subscription now follows the session state through a SessionActivityListener, registered before the client connects, and a check or a delete that fails while the session is not active is left to the session recovery. The status codes of a lost session also include the secure channel faults on which the client leaves the session. The outcome of the check and of the delete is now handled on the executor of the client instead of the shared scheduled executor. The recreate and the rebuild block until the server answers or a reconnect attempt fails, and the client schedules those attempts on that shared executor; with one processor it has a single thread, and the rebuild after a delete that failed while the server was unreachable blocked it for good. The executor of the client is Milo's shared cached pool, so a blocked thread does not keep a response from completing. New test: a route that holds one value longer than the watchdog keeps the subscription and receives the value once. The test of a node that cannot be monitored again now waits until the remaining monitored items are created on the server. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01YTrG4zh2t5bq7jniA5Nwvg --- .../catalog/docs/milo-client-component.adoc | 25 ++-- .../src/main/docs/milo-client-component.adoc | 25 ++-- .../client/internal/SubscriptionManager.java | 125 +++++++++++++++--- .../milo/MonitorItemLifecycleTest.java | 66 ++++++++- .../pages/camel-4x-upgrade-guide-4_24.adoc | 7 +- 5 files changed, 208 insertions(+), 40 deletions(-) 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 167ae0ad43c70..d7bfda50f74f1 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 @@ -168,16 +168,21 @@ does not learn that the node recovered. Set `omitNullValues=false` for a node wh === Subscriptions that stop publishing The consumers of a connection share one subscription on the server. When the server stops publishing for it (no -notification or keep-alive within about one and a half keep-alive intervals) or reports that it timed out -(`Bad_Timeout`), Camel deletes the subscription and, once the server has confirmed that it is gone, creates it again -in the existing session. A node that can no longer be monitored is reported as described above and does not affect -the other nodes. - -A network that drops silently can stop the publishing before the client notices that the connection is lost. The delete -then fails with the connection, and the client restores the session once the server can be reached again: the -subscription resumes publishing, or it is created again when the server reports that it timed out in the meantime. If -the network is back before the client has noticed the loss, the delete reaches the server and the subscription is -created again. +notification or keep-alive within about one and a half keep-alive intervals), Camel checks whether the server still +has the subscription. If the server does not confirm it, or when the server reports that the subscription timed out +(`Bad_Timeout`), Camel deletes it and, once the server has confirmed that it is gone, creates it again in the existing +session. A node that can no longer be monitored is reported as described above and does not affect the other nodes. + +A route that is slow to process a value stops the publishing as well, because the client asks the server for more +notifications only as the routes finish processing the ones it received. The subscription is then kept, unless the +route takes longer than the lifetime of the subscription, which lets it time out on the server. + +A check or a delete that fails because the connection was lost is left to the session recovery of the client, which +restores the session once the server can be reached again. The subscription then resumes publishing, or it is created +again as above when the server no longer has it. If the server no longer has the session either, for example after a +restart, the subscription cannot be transferred to a new session and the connection is rebuilt. A network that drops +silently can stop the publishing before the client notices that the connection is lost; if the network is back before +that, the check reaches the server and is handled as above. Any other failure of the delete, or a subscription that cannot be created again, rebuilds the connection to the server. After the subscription is recreated, every monitored item delivers its current value again, and value changes in 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 167ae0ad43c70..d7bfda50f74f1 100644 --- a/components/camel-milo/src/main/docs/milo-client-component.adoc +++ b/components/camel-milo/src/main/docs/milo-client-component.adoc @@ -168,16 +168,21 @@ does not learn that the node recovered. Set `omitNullValues=false` for a node wh === Subscriptions that stop publishing The consumers of a connection share one subscription on the server. When the server stops publishing for it (no -notification or keep-alive within about one and a half keep-alive intervals) or reports that it timed out -(`Bad_Timeout`), Camel deletes the subscription and, once the server has confirmed that it is gone, creates it again -in the existing session. A node that can no longer be monitored is reported as described above and does not affect -the other nodes. - -A network that drops silently can stop the publishing before the client notices that the connection is lost. The delete -then fails with the connection, and the client restores the session once the server can be reached again: the -subscription resumes publishing, or it is created again when the server reports that it timed out in the meantime. If -the network is back before the client has noticed the loss, the delete reaches the server and the subscription is -created again. +notification or keep-alive within about one and a half keep-alive intervals), Camel checks whether the server still +has the subscription. If the server does not confirm it, or when the server reports that the subscription timed out +(`Bad_Timeout`), Camel deletes it and, once the server has confirmed that it is gone, creates it again in the existing +session. A node that can no longer be monitored is reported as described above and does not affect the other nodes. + +A route that is slow to process a value stops the publishing as well, because the client asks the server for more +notifications only as the routes finish processing the ones it received. The subscription is then kept, unless the +route takes longer than the lifetime of the subscription, which lets it time out on the server. + +A check or a delete that fails because the connection was lost is left to the session recovery of the client, which +restores the session once the server can be reached again. The subscription then resumes publishing, or it is created +again as above when the server no longer has it. If the server no longer has the session either, for example after a +restart, the subscription cannot be transferred to a new session and the connection is rebuilt. A network that drops +silently can stop the publishing before the client notices that the connection is lost; if the network is back before +that, the check reaches the server and is handled as above. Any other failure of the delete, or a subscription that cannot be created again, rebuilds the connection to the server. After the subscription is recreated, every monitored item delivers its current value again, and value changes in 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 39b3c0f4bc83c..e8dab0b933369 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 @@ -31,6 +31,7 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executor; import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -49,6 +50,8 @@ import org.eclipse.milo.opcua.sdk.client.DiscoveryClient; import org.eclipse.milo.opcua.sdk.client.OpcUaClient; import org.eclipse.milo.opcua.sdk.client.OpcUaClientConfigBuilder; +import org.eclipse.milo.opcua.sdk.client.SessionActivityListener; +import org.eclipse.milo.opcua.sdk.client.UaSession; import org.eclipse.milo.opcua.sdk.client.identity.AnonymousProvider; import org.eclipse.milo.opcua.sdk.client.identity.CompositeProvider; import org.eclipse.milo.opcua.sdk.client.identity.IdentityProvider; @@ -99,21 +102,52 @@ public class SubscriptionManager { /** * The status codes of a request that failed because the connection or the session was lost, which the session - * recovery of the client deals with + * recovery of the client deals with: a closed connection, and the service faults on which the client leaves the + * active session because its session or its secure channel is no longer valid. The faults are those of + * {@code SESSION_ERROR} and {@code SECURE_CHANNEL_ERROR} in the session state machine of Milo 1.1 + * ({@code SessionFsmFactory}). */ private static final Set SESSION_LOST = Set.of( StatusCodes.Bad_ConnectionClosed, StatusCodes.Bad_SessionClosed, StatusCodes.Bad_SessionIdInvalid, - StatusCodes.Bad_SessionNotActivated); + StatusCodes.Bad_SessionNotActivated, StatusCodes.Bad_SecureChannelIdInvalid, + StatusCodes.Bad_SecurityChecksFailed, StatusCodes.Bad_TcpSecureChannelUnknown, + StatusCodes.Bad_RequestTypeInvalid); private final AtomicLong clientHandleCounter = new AtomicLong(); - private final class SubscriptionListenerImpl implements OpcUaSubscription.SubscriptionListener { + /** + * Listens to the subscription of a connection and to the state of its session. It is registered on the client + * before the client connects, so that it sees every change of the session state. + */ + private final class SubscriptionListenerImpl + implements OpcUaSubscription.SubscriptionListener, SessionActivityListener { + + private volatile boolean sessionActive; + + @Override + public void onSessionActive(final UaSession session) { + this.sessionActive = true; + } + + @Override + public void onSessionInactive(final UaSession session) { + this.sessionActive = false; + } + + /** + * Whether a request failed because the connection or the session was lost, which the session recovery of the + * client deals with. A request that fails while the session is not active counts too, whatever its status: one + * that was sent after the loss waits for that recovery and fails with the error of a failed reconnect attempt. + */ + boolean isSessionLost(final StatusCode status) { + return !this.sessionActive || SESSION_LOST.contains(status.getValue()); + } @Override public void onStatusChanged(final OpcUaSubscription subscription, final StatusCode status) { if (status.getValue() == StatusCodes.Bad_Timeout) { LOG.warn("Subscription timed out on the server, recreating it"); - recoverSubscription(subscription); + recoverSubscription(subscription, this); } else { LOG.info("Subscription status changed {} : {}", subscription.getSubscriptionId(), status); } @@ -121,8 +155,9 @@ public void onStatusChanged(final OpcUaSubscription subscription, final StatusCo @Override public void onWatchdogTimerElapsed(final OpcUaSubscription subscription) { - LOG.warn("Subscription {} stopped publishing, recreating it", subscription.getSubscriptionId().orElse(null)); - recoverSubscription(subscription); + LOG.debug("Subscription {} stopped publishing, checking it on the server", + subscription.getSubscriptionId().orElse(null)); + checkSubscription(subscription, this); } @Override @@ -719,26 +754,67 @@ public OpcUaClient getOpcUaClient() { } /** - * Recovers the subscription of a connection after it stopped publishing or timed out on the server. + * Checks the subscription of a connection on the server after it stopped publishing, and recovers it unless the + * server confirms that it still has it. + *

+ * A subscription that the server still has stops publishing too while a route is slow: the client replaces each of + * its outstanding publish requests only once the notifications of that request have been processed, so the server + * has none left to answer. The check is a SetPublishingMode request that enables publishing; the subscription is + * always created with publishing enabled, so the request does not change its settings. + *

+ * If the client has reset the subscription in the meantime, it is already being recovered: the client resets it + * when the server reports that it timed out, when it was deleted and when it could not be transferred, and each of + * these recovers it. A check that fails because the connection or the session was lost is left to the session + * recovery of the client, as a failed delete in + * {@link #recoverSubscription(OpcUaSubscription, SubscriptionListenerImpl)} is. Any other answer goes on to + * {@link #recoverSubscription(OpcUaSubscription, SubscriptionListenerImpl)}. + */ + private void checkSubscription(final OpcUaSubscription subscription, final SubscriptionListenerImpl listener) { + final UInteger subscriptionId = subscription.getSubscriptionId().orElse(null); + subscription.setPublishingModeAsync(true).whenCompleteAsync((unit, e) -> { + final StatusCode status = e == null + ? StatusCode.GOOD + : UaException.extractStatusCode(e).orElse(new StatusCode(StatusCodes.Bad_UnexpectedError)); + if (status.isGood()) { + LOG.info("Subscription {} stopped publishing but still exists on the server, keeping it", + subscriptionId); + } else if (status.getValue() == StatusCodes.Bad_InvalidState) { + LOG.debug("Subscription {} was reset while it was checked, it is already being recovered", + subscriptionId); + } else if (listener.isSessionLost(status)) { + LOG.info("Session lost while checking subscription {}, leaving it to the session recovery: {}", + subscriptionId, status); + } else { + LOG.warn("Subscription {} stopped publishing and was not confirmed by the server, recreating it: {}", + subscriptionId, status); + recoverSubscription(subscription, listener); + } + }, clientExecutor(subscription)); + } + + /** + * Recovers the subscription of a connection after the server did not confirm it or reported that it timed out. *

* The subscription is deleted first, and it is only created again, inside the existing session, once the server has * confirmed that it no longer exists; one that the client has already reset because it timed out is deleted without - * a service call. If the delete fails because the connection or the session was lost, the session recovery of the - * client takes over: it resumes the subscription, which arms its watchdog again, or reports that the subscription - * could not be transferred. Any other failure rebuilds the connection, which closes the session together with its + * a service call. A delete that fails because the connection or the session was lost, or that fails while the + * session is not active, is left to the session recovery of the client: it resumes the subscription, whose watchdog + * is armed again once publishing resumes, or reports that the subscription could not be transferred, which rebuilds + * the connection. Any other failure rebuilds the connection, which closes the session together with its * subscriptions. *

* The delete neither blocks nor holds {@code lock}: a request sent before the connection was lost only fails once - * the client notices the loss, which its keep-alive does on the threads of {@link #executor}. + * the client notices the loss, and holding {@code lock} until then would hold up the producers and + * {@link #getOpcUaClient()} as well. */ - private void recoverSubscription(final OpcUaSubscription subscription) { + private void recoverSubscription(final OpcUaSubscription subscription, final SubscriptionListenerImpl listener) { subscription.deleteAsync().whenCompleteAsync((unit, e) -> { final StatusCode status = e == null ? StatusCode.GOOD : UaException.extractStatusCode(e).orElse(new StatusCode(StatusCodes.Bad_UnexpectedError)); if (status.isGood() || status.getValue() == StatusCodes.Bad_SubscriptionIdInvalid) { recreateSubscription(subscription); - } else if (SESSION_LOST.contains(status.getValue())) { + } else if (listener.isSessionLost(status)) { LOG.info("Session lost while deleting subscription {}, leaving it to the session recovery: {}", subscription.getSubscriptionId().orElse(null), status); } else { @@ -751,7 +827,23 @@ private void recoverSubscription(final OpcUaSubscription subscription) { lock.unlock(); } } - }, this.executor); + }, clientExecutor(subscription)); + } + + /** + * The executor that the client of a subscription delivers its values on, which handles the outcome of a check or a + * delete. Creating the subscription again and rebuilding the connection block until the server answers, and a + * request sent while the connection is lost waits for the channel until a reconnect attempt restores it or fails. + * The client schedules those attempts on the shared scheduled executor of Milo, which {@link #executor} is as well + * and which has a single thread on a machine with one processor, so the wait must not hold one of its threads. + *

+ * Milo asks that blocking calls are not made on the threads of this executor either, because they complete the + * responses. The executor of the transport is not set here, so this is the shared executor of Milo, a cached pool + * that starts another thread for a response while one of its threads waits, as it does while a route processes a + * value. + */ + private static Executor clientExecutor(final OpcUaSubscription subscription) { + return subscription.getClient().getTransport().getConfig().getExecutor(); } /** @@ -958,6 +1050,9 @@ private Connected performConnect() throws Exception { } }); + final SubscriptionListenerImpl listener = new SubscriptionListenerImpl(); + client.addSessionActivityListener(listener); + try { // connect is called inside this try block to ensure that disconnect is called on exceptions @@ -966,7 +1061,7 @@ private Connected performConnect() throws Exception { // Create subscription synchronously - the create() method blocks until complete OpcUaSubscription subscription = new OpcUaSubscription(client); subscription.setPublishingInterval(this.configuration.getRequestedPublishingInterval()); - subscription.setSubscriptionListener(new SubscriptionListenerImpl()); + subscription.setSubscriptionListener(listener); subscription.create(); return new Connected(client, subscription); 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 29189560ec2ba..c2d3ad4b0ddbb 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 @@ -18,6 +18,7 @@ import java.util.List; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; import org.apache.camel.EndpointInject; @@ -46,6 +47,7 @@ import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNotSame; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -103,6 +105,11 @@ public class MonitorItemLifecycleTest extends AbstractMiloServerTest { private static final String MOCK_ITEMS_1_2 = "mock:items12"; + // the route of the first item takes longer than the watchdog of a shortened subscription for this value, once + private static final String SLOW_VALUE = "Slow"; + + private final AtomicBoolean slowValueProcessed = new AtomicBoolean(); + @EndpointInject(MOCK_ITEM_1) protected MockEndpoint item1Endpoint; @@ -135,7 +142,14 @@ public void configure() { from(resolve(MILO_CLIENT_UNKNOWN_ITEM)).routeId("unknownItem").to(MOCK_UNKNOWN_ITEM); // 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_ITEM_1); + from(resolve(MILO_CLIENT_ITEM_1)).routeId("item1").autoStartup(false) + .process(exchange -> { + final Object value = exchange.getMessage().getBody(DataValue.class).getValue().getValue(); + if (SLOW_VALUE.equals(value) && slowValueProcessed.compareAndSet(false, true)) { + Thread.sleep(3000); + } + }) + .to(MOCK_ITEM_1); from(resolve(MILO_CLIENT_ITEM_2)).routeId("item2").autoStartup(false).to(MOCK_ITEM_2); from(resolve(MILO_CLIENT_ITEMS_1_2)).routeId("items12").autoStartup(false).to(MOCK_ITEMS_1_2); @@ -337,6 +351,46 @@ void subscriptionThatTimesOutOnTheServerIsRecreatedInTheSameSession() throws Exc } } + @Test + @Timeout(value = 60, unit = TimeUnit.SECONDS) + void subscriptionThatASlowRouteHeldUpIsKept() throws Exception { + producer1.sendBody("Foo"); + + item1Endpoint.expectedMinimumMessageCount(1); + testBody(item1Endpoint.message(0), assertGoodValue("Foo")); + context.getRouteController().startRoute("item1"); + item1Endpoint.assertIsSatisfied(); + + final MiloClientEndpoint endpoint = context.getEndpoint(resolve(MILO_CLIENT_ITEM_1), MiloClientEndpoint.class); + final MiloClientConnection connection = endpoint.createConnection(); + try { + final OpcUaClient client = connection.getOpcUaClient(); + final OpcUaSubscription subscription = client.getSubscriptions().get(0); + // shortens the watchdog of the subscription to a few publishing intervals + subscription.setMaxKeepAliveCount(uint(1)); + subscription.modify(); + final UInteger subscriptionId = subscription.getSubscriptionId().orElseThrow(); + // replaced when the server confirms the subscription + final OpcUaSubscription.ServerState stateBefore = subscription.getServerState().orElseThrow(); + + // the client sends no publish request while the route processes the value, so the watchdog elapses + producer1.sendBody(SLOW_VALUE); + await().atMost(20, TimeUnit.SECONDS).until(() -> countValue(item1Endpoint, SLOW_VALUE) >= 1); + + // a recreated subscription would deliver the current value again + await().during(3, TimeUnit.SECONDS).atMost(5, TimeUnit.SECONDS) + .until(() -> countValue(item1Endpoint, SLOW_VALUE) == 1); + assertEquals(subscriptionId, subscription.getSubscriptionId().orElse(null)); + assertNotSame(stateBefore, subscription.getServerState().orElseThrow(), "the subscription was not checked"); + producer1.sendBody("Bar"); + await().atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> assertGoodValue("Bar").accept(lastExchange(item1Endpoint).getMessage() + .getBody(DataValue.class))); + } finally { + endpoint.releaseConnection(connection); + } + } + @Test @Timeout(value = 60, unit = TimeUnit.SECONDS) void nodeThatCannotBeMonitoredAgainIsReportedWithoutRebuildingTheConnection() throws Exception { @@ -363,7 +417,9 @@ void nodeThatCannotBeMonitoredAgainIsReportedWithoutRebuildingTheConnection() th await().atMost(30, TimeUnit.SECONDS) .until(() -> subscription.getSubscriptionId().filter(id -> !id.equals(subscriptionId)).isPresent()); assertSame(client, connection.getOpcUaClient()); - assertEquals(2, subscription.getMonitoredItems().size()); + // the items are added one by one before they are created, so the count alone passes 2 on the way to 3 + await().atMost(10, TimeUnit.SECONDS).until(() -> subscription.getMonitoredItems().size() == 2 + && subscription.isMonitoredItemsSynchronized()); // the consumer of the second node is told that it is gone, the consumers of the first node keep receiving await().atMost(10, TimeUnit.SECONDS).until(() -> items12Endpoint.getReceivedExchanges().stream() @@ -398,6 +454,12 @@ private void timeOutSubscription(final UInteger subscriptionId) { } } + private static long countValue(final MockEndpoint endpoint, final Object value) { + return endpoint.getReceivedExchanges().stream() + .filter(exchange -> value.equals(exchange.getMessage().getBody(DataValue.class).getValue().getValue())) + .count(); + } + private static Exchange lastExchange(final MockEndpoint endpoint) { final List exchanges = endpoint.getReceivedExchanges(); return exchanges.get(exchanges.size() - 1); 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 e552385b1f23b..204aba3951f85 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 @@ -1544,9 +1544,10 @@ Stopping a consumer now deletes its monitored items. Previously they stayed on t the shared connection open, so a restarted consumer received each value twice. While the server is unreachable, the stop waits until the client's next attempt to reconnect has failed. -A consumer whose subscription stops publishing, or times out on the server, now recreates the subscription in the -existing session; previously it stopped receiving values until the connection was rebuilt for another reason. The -monitored items deliver their current values again after the subscription is recreated. +When the server removes the subscription that the consumers of a connection share, or the subscription times out on the +server, it is now created again in the existing session; previously the consumers stopped receiving values until the +connection was rebuilt for another reason. The monitored items deliver their current values again after the +subscription is recreated. === camel-protobuf - a bytes field in a Map From 202f1e2b45e71592addd56daf1e8ce2aa2699348 Mon Sep 17 00:00:00 2001 From: Claus Ibsen Date: Sat, 17 Oct 2026 16:14:49 +0200 Subject: [PATCH 4/5] CAMEL-25691: camel-milo - the slow route of the test uses a synchronous delay instead of Thread.sleep Review follow-up: the route of the first item holds the delivery thread for the slow value with the delay EIP (syncDelayed) inside a filter, instead of calling Thread.sleep in a processor, which new test code should not add. Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Claus Ibsen --- .../component/milo/MonitorItemLifecycleTest.java | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) 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 c2d3ad4b0ddbb..f0a5513bceb7a 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 @@ -143,12 +143,10 @@ public void configure() { // 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) - .process(exchange -> { - final Object value = exchange.getMessage().getBody(DataValue.class).getValue().getValue(); - if (SLOW_VALUE.equals(value) && slowValueProcessed.compareAndSet(false, true)) { - Thread.sleep(3000); - } - }) + // the slow route under test: a synchronous delay holds the delivery thread, once + .filter(exchange -> isSlowValueOnce(exchange)) + .delay(3000).syncDelayed() + .end() .to(MOCK_ITEM_1); from(resolve(MILO_CLIENT_ITEM_2)).routeId("item2").autoStartup(false).to(MOCK_ITEM_2); from(resolve(MILO_CLIENT_ITEMS_1_2)).routeId("items12").autoStartup(false).to(MOCK_ITEMS_1_2); @@ -163,6 +161,11 @@ public void configure() { }; } + private boolean isSlowValueOnce(Exchange exchange) { + final Object value = exchange.getMessage().getBody(DataValue.class).getValue().getValue(); + return SLOW_VALUE.equals(value) && slowValueProcessed.compareAndSet(false, true); + } + @Test void restartedConsumerReceivesEachValueOnce() throws Exception { producer1.sendBody("Foo"); From 6f24e0573df59b21a4c3aec242d83357d337021b Mon Sep 17 00:00:00 2001 From: Claus Ibsen Date: Sat, 17 Oct 2026 16:50:20 +0200 Subject: [PATCH 5/5] CAMEL-25691: camel-milo - end the delay block before the filter in the test route The delay EIP is a block of its own, so the single end() closed the delay and left the mock endpoint inside the filter: only the slow value reached it. Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Claus Ibsen --- .../apache/camel/component/milo/MonitorItemLifecycleTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 f0a5513bceb7a..f6797ff65d196 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 @@ -145,7 +145,7 @@ public void configure() { from(resolve(MILO_CLIENT_ITEM_1)).routeId("item1").autoStartup(false) // the slow route under test: a synchronous delay holds the delivery thread, once .filter(exchange -> isSlowValueOnce(exchange)) - .delay(3000).syncDelayed() + .delay(3000).syncDelayed().end() // the delay is a block of its own .end() .to(MOCK_ITEM_1); from(resolve(MILO_CLIENT_ITEM_2)).routeId("item2").autoStartup(false).to(MOCK_ITEM_2);