You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by "etzel-mes (via GitHub)" <gi...@apache.org> on 2026/10/15 17:00:31 UTC
[PR] CAMEL-25691: camel-milo - recreate the subscription when it stops publishing [camel]
etzel-mes opened a new pull request, #27910:
URL: https://github.com/apache/camel/pull/27910
# Description
Fixes [CAMEL-25691](https://issues.apache.org/jira/browse/CAMEL-25691).
`SubscriptionManager.SubscriptionListenerImpl` ignored `onWatchdogTimerElapsed` and only logged `onStatusChanged`. When the server drops the subscription while the session stays alive (subscription lifetime expired after a network or scheduling stall, server configuration reload, `Bad_NoSubscription` on publish), Milo stops sending publish requests, fires the watchdog and leaves the reaction to the listener. The consumer then kept its last value forever: no exchange, no reconnect, one INFO line at most. `Bad_Timeout` is the same case reported by the server: Milo resets the subscription before the callback, and nothing created it again.
With this change the consumer recreates the subscription inside the existing session on both callbacks: `delete()` best effort (`reset()` if the delete fails), `create()`, `createMonitoredItems()` on the same `OpcUaSubscription` and the same monitored items, which Milo keeps across `reset()` together with their client handles and listeners. The work is handed to the manager's executor and runs under the connection lock, never on Milo's delivery thread. A guard skips callbacks that belong to a disposed connection or to a subscription that was already recreated. If the recreation fails, the existing `handleConnectionFailure` rebuilds the connection as before.
## Why recreate the subscription instead of rebuilding the connection
A `MiloClientConnection` is shared by every consumer and producer with the same connection parameters. Rebuilding it for one dead subscription closes the session, leaves writes, method calls and reads with "No connected" until the reconnect finishes, re-subscribes every consumer and opens a new session on the server. PLC-integrated OPC UA servers (Siemens S7-1500/S7-1200 and similar controllers) allow only a small number of sessions, and a session whose CloseSession did not reach the server counts against that limit until its timeout expires. Recreating the subscription in place is the pattern Milo documents for `reset()`, does not touch the other users of the connection and leaves the session count unchanged. The full rebuild stays as the fallback.
Not changed: `onTransferFailed` still rebuilds the connection; value changes during the gap are not republished (the monitored items deliver their current value again after recreation, as after a reconnect). Both are documented.
## Tests
`MonitorItemLifecycleTest`:
* `subscriptionThatStopsPublishingIsRecreatedInTheSameSession`: shortens the watchdog through Milo's public API (`setMaxKeepAliveCount(1)`, `modify()`), deletes the subscription through the connection's own session (the server then answers the pending publish requests with `Bad_NoSubscription`), and asserts that a new value arrives, the `OpcUaClient` is the same instance and the subscription id changed. The watchdog fires after about 0.4 s.
* `subscriptionThatTimesOutOnTheServerIsRecreatedInTheSameSession`: the embedded server sends a `Bad_Timeout` status change and removes the subscription.
Both fail without the fix (timeout after 30 s, last value stays the old one) and pass with it; the class was run repeatedly with `-Dsurefire.rerunFailingTestsCount=0`. `./mvnw -pl components/camel-milo -Psourcecheck verify`: 57 tests, 0 failures.
Docs: a "Subscriptions that stop publishing" section in `milo-client-component.adoc` (catalog copy identical), and a note in the 4.24 upgrade guide under the existing camel-milo section.
@davsclaus @oscerd
# Target
- [x] I checked that the commit is targeting the correct branch (note that Camel 3 uses `camel-3.x`, whereas Camel 4 uses the `main` branch)
# Tracking
- [x] If this is a large change, bug fix, or code improvement, I checked there's a JIRA issue (CAMEL-25691)
# Apache Camel coding standards and style
- [x] I checked that each commit in the pull request has a meaningful subject line and body.
- [x] I have run `mvn clean install -DskipTests` locally from root directory and I have committed all auto-generated changes.
- [x] I used AI to generate this pull request. The hand-written code and tests were reviewed and verified by me.
_Claude Code on behalf of etzel-mes_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25691: camel-milo - recreate the subscription when it stops publishing [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27910:
URL: https://github.com/apache/camel/pull/27910#issuecomment-6198172786
:star2: Thank you for your contribution to the Apache Camel project! :star2:
:robot: CI automation will test this PR automatically.
:camel: Apache Camel Committers, please review the following items:
* First-time contributors **require MANUAL approval** for the GitHub Actions to run
* You can use the command `/component-test (camel-)component-name1 (camel-)component-name2..` to request a test from the test bot although they are normally detected and executed by CI.
* You can label PRs using `skip-tests` and `test-dependents` to fine-tune the checks executed by this PR.
* Build and test logs are available in the summary page. **Only** [Apache Camel committers](https://camel.apache.org/community/team/#committers) have access to the summary.
:warning: Be careful when sharing logs. Review their contents before sharing them publicly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25691: camel-milo - recreate the subscription when it stops publishing [camel]
Posted by "gnodet-bot (via GitHub)" <gi...@apache.org>.
gnodet-bot commented on code in PR #27910:
URL: https://github.com/apache/camel/pull/27910#discussion_r4286084953
##########
components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java:
##########
@@ -683,6 +697,37 @@ 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.
+ */
+ private void recreateSubscription(final OpcUaSubscription subscription, final Optional<UInteger> 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();
Review Comment:
⚠️ **No timeout on `subscription.create()`** — If the OPC UA server becomes unresponsive after `delete()` has already been called, `create()` will block indefinitely while holding `lock`. This would stall all routing threads until the underlying TCP stack eventually times out (often minutes).
Consider applying the same timeout strategy used elsewhere in Milo client interactions, or documenting that the caller relies on TCP-level timeouts.
##########
components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java:
##########
@@ -683,6 +697,37 @@ 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.
+ */
+ private void recreateSubscription(final OpcUaSubscription subscription, final Optional<UInteger> 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)) {
Review Comment:
⚠️ **Blocking network I/O under lock** — The method holds `lock` across `subscription.delete()` → `subscription.create()` → `subscription.createMonitoredItems()`, all of which are synchronous network roundtrips to the OPC UA server.
This differs from the existing `performAndEvalConnect()` pattern, where `performConnect()` (including `subscription.create()`) runs **outside** the lock, and `lock.lock()` is only acquired afterwards to update `this.connected`. Any Camel routing thread calling `write()`, `read()`, `activate()`, `deactivate()`, or `dispose()` will block on `lock.lock()` for the full duration of these network roundtrips.
If this is intentional (e.g. to prevent concurrent recreation attempts), please add a brief Javadoc note explaining the trade-off.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25691: camel-milo - recreate the subscription when it stops publishing [camel]
Posted by "etzel-mes (via GitHub)" <gi...@apache.org>.
etzel-mes commented on PR #27910:
URL: https://github.com/apache/camel/pull/27910#issuecomment-6220104224
Thanks @allthingssecurity, confirmed. Milo resets the watchdog when a PublishResponse is received (`PublishingManager`: "reset when the response is received rather than when it is eventually processed") but replaces a publish request only after its notifications have been delivered, and the callback waits on the same delivery queue. A route slower than the watchdog therefore makes a live subscription look dead. Your reproducer failed on 9c6d4cf for me as well (`expected: <1> but was: <2>`, the subscription was recreated).
Fixed in 70aca32 with the check you suggested. After the watchdog, `setPublishingModeAsync(true)` checks the subscription first; Milo creates it with publishing enabled, so the request does not change its settings.
- `Good`: the subscription is kept.
- `Bad_InvalidState`: nothing. Milo resets the subscription only after a `Bad_Timeout`, after a delete or after a failed transfer, and each of those already recovers it. A second recovery could delete the subscription that the first one has just created. This is where I departed from your suggestion.
- Connection or session lost: left to Milo's session recovery. `SESSION_LOST` now also has the secure channel faults on which Milo leaves the session (`SessionFsmFactory.SessionFaultListener`). The listener also follows the session state with a `SessionActivityListener`, registered before the client connects: a request sent after the client noticed the loss, for example the check 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, such as `Bad_Timeout`. A check or a delete that fails while the session is not active is therefore left to Milo as well. The delete of the previous revision had this gap.
- Anything else, including `Bad_SubscriptionIdInvalid`: delete and recreate, as before. A server that rejected SetPublishingMode would still get the recreate, not a rebuild.
A `Bad_Timeout` status change from the server still recreates the subscription directly, without the check, since the server has already said that the subscription is gone. A route that takes longer than the lifetime of the subscription lets it time out and gets it recreated; the docs say so now.
Testing this against a real server also turned up a hang in 9c6d4cf. It handled the outcome of the delete on `executor`, which is `Stack.sharedScheduledExecutor()`, and the recreate (`create()`) and the rebuild (`disconnect()`) block there while holding the lock. With one processor that executor has a single thread, and a request sent while the server cannot be reached waits until a reconnect attempt restores the channel or fails; Milo schedules those attempts on that same thread. In a container limited to one CPU, 9c6d4cf sent the delete of a watchdog that had waited behind a slow route while the session was being reactivated. The delete failed with the `Bad_Timeout` of the failed reactivation, and the rebuild blocked that thread in `OpcUaClient.disconnect()` for good: the consumer stayed silent after the network came back, and `getOpcUaClient()` waited for the lock. The outcome of the check and of the delete is now handled on the client's executor (`getTransport().getConfig().ge
tExecutor()`). Milo asks that blocking calls are not made there either, because its threads complete the responses. camel-milo does not set the transport's executor, so this is Milo's shared cached pool, which starts another thread for a response while one waits, as it already does while a route processes a value.
The producers can reach the same wait, and this PR does not change that path: the failure of a write, a method call, a read or a browse rebuilds the connection on `executor` as well. With a producer writing every 2 s on the connection, one CPU and a 30 s silent outage, upstream main (a16cccbde62) hung in 3 of 11 runs and this revision in 4 of 8, each time in `SubscriptionManager.write` → `handleConnectionFailure` → `OpcUaClient.disconnect()` on the scheduled thread. From the logs it is a race: after the failed keep-alive Milo closes the channel, and a CloseSession that reaches the transport before the channel state machine has handled the close fails at once, while a later one waits for the reconnect. Handling those failures on the client's executor recovered in all 9 runs of the same test, but it also moves the successful completions to that executor and needs a guard so that a stale failure does not start a second reconnect, so I will take it up separately; it touches the sa
me handlers as CAMEL-25690. The reconnect job (`connect()`) and `resubscribe` also still run on `executor`, where the same wait is possible in a narrower window; CAMEL-25692 already proposes an executor owned by the component.
Tests:
- New test `MonitorItemLifecycleTest#subscriptionThatASlowRouteHeldUpIsKept`, based on your reproducer: red on 9c6d4cf, green now. Besides the delivery count and the subscription id, it asserts that the subscription's `ServerState` was replaced, which, with the id unchanged, only a `Good` answer to the check does here, so the test fails if the check never ran.
- Your second variant (Camel's and Milo's defaults, a route that holds one value for 30 s) I ran against a real server with this round's code, the first item in the list below: the watchdog fires after 16.5 s, the check is answered `Good`, the value is delivered once and the subscription id is unchanged. It is not in the suite because of its length.
- The recovery still works: in `subscriptionThatStopsPublishingIsRecreatedInTheSameSession` the check is answered `Bad_SubscriptionIdInvalid` and the subscription is recreated. In the chaos run from my previous comment, the server's log shows that it had deleted the subscription 14.7 to 16.5 s before each of the 26 watchdog firings, so none of them would have been answered `Good`.
- The executor change has no unit test: it needs a scheduled executor with one thread and a server that cannot be reached. The real-server runs below show what it changes.
Against a real server (opc-plc 2.15.4, Milo defaults: watchdog 16.5 s, lifetime 50 s; each client in a container limited to one CPU; traffic blocked with iptables in the client's network namespace):
- A route that holds one value for 30 s: the check is answered `Good`, the subscription is kept and the value is delivered once.
- The server loses the subscription while the session stays (the client deletes it on the server behind the back of its `OpcUaSubscription`): upstream main delivered nothing for 90 s. With this round the check is answered `Bad_SubscriptionIdInvalid`, the subscription is recreated in the same session and a value written after the loss reached the route within 13 s.
- A route that holds one value for 90 s, longer than the lifetime: the server reports `Bad_Timeout` and the subscription is recreated once; the check that had waited behind the route is answered `Bad_InvalidState` and does nothing (4 of 4 runs). With a 60 s hold the server still had the subscription (3 of 3): it answers the last outstanding publish request with a keep-alive first, and the lifetime counts from there.
- A slow route, with the network blocked after the watchdog has elapsed behind the route and before the route finishes, so that the queued check is sent while the session is not active: the check fails with `Bad_Timeout` from a reconnect attempt and is left to Milo. The session is reactivated with the same subscription, and a value written after the network was back reached the route within 2.2 to 2.3 s (three runs). 9c6d4cf went on to the delete and the rebuild and hung as described above. This round's check without the session state, with its outcome still on the scheduled executor, hung the same way in three of three runs; with the session state it recovered.
- A 30 s silent drop starting 8 s after a keep-alive: the watchdog elapses first, the check is in flight when the client notices the loss and fails with `Bad_ConnectionClosed`. Same session and subscription; a value written after the drop reached the route within 2.1 s.
- A 10 s drop: the check reaches the server once the drop ends and is answered `Good`, so the subscription is kept; the previous revision recreated it.
- A 75 s silent drop, longer than the lifetime: the client notices the loss before the watchdog elapses, so Milo pauses the watchdog and no check is sent. After the session is reactivated the server reports `Bad_Timeout`, and the subscription is recreated in the same session; a value written after the drop reached the route within 9.9 s.
- A server restart: the transfer fails with `Bad_SubscriptionIdInvalid` and the connection is rebuilt, as before.
- The executor of the recovery, with a recreate in flight when the server becomes unreachable (the subscription deleted on the server, 3 s of added latency so that CreateMonitoredItems is still on its way when the traffic is dropped for 90 s): with the outcome on the scheduled executor, the recreate held its only thread until the request timed out after 60 s. No keep-alive ran in that time, so the loss went unnoticed and the lock was held for 68 s (2 of 2 runs). On the client's executor the keep-alive noticed the loss after 8 s and the connection was rebuilt. With a producer writing on the same connection, the same scenario with a 30 s drop recovered in 7 of 7 runs.
One case is left out on purpose: a subscription that the server confirms and then removes, or stops serving, before its next publish response. After a `Good` answer Milo arms the watchdog again only on the next PublishResponse, a reactivated session, or a create or modify; neither the SetPublishingMode answer nor a `Bad_NoSubscription` publish fault does, so such a subscription is not checked again. For a server that keeps the keep-alive rule of the subscription model the window ends with its next keep-alive. A ModifySubscription after each confirmation would close it, at the cost of one more request per check; I left it out, but can add it if you prefer. The previous revision only caught these cases because it recreated every subscription that stopped publishing.
Before the rebase below, the class passed 5 runs in a row with reruns disabled; after it, the class passes on each of the three commits. Module `-Psourcecheck verify`: 78 tests, 1 skipped and 1 failure, `ServerCertificateValidationTest` (CAMEL-25645, merged today), which fails the same way on main (16b6e9dfbe8): `mock:trustAll` receives the Bad status of the server item that was not written yet before `Foo`, and the consumer delivers that status since #27908. The docs section (now also covering a slow route and a failed transfer) and the 4.24 note are updated, and so is the PR description.
@davsclaus, this round changes more than the revision you approved: the check, the session state and the executor of the recovery. Your inline suggestion is applied in the same commit, with one addition (details in the thread). #27908 has been merged in the meantime, so the branch is rebased onto main; 9c6d4cf, the revision you approved, is now d3c527e. Since #27908 the tests start `item1` after the first write, and the recreate tests and the new one do that now as well. `nodeThatCannotBeMonitoredAgainIsReportedWithoutRebuildingTheConnection` starts `item1` too, so the subscription has the same three monitored items as before. #27906 changes the same part of `SubscriptionManager` (it removes `resubscribe` next to `recreate`); whichever of the two is merged second, I will rebase it.
_Claude Code on behalf of etzel-mes_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25691: camel-milo - recreate the subscription when it stops publishing [camel]
Posted by "etzel-mes (via GitHub)" <gi...@apache.org>.
etzel-mes commented on code in PR #27910:
URL: https://github.com/apache/camel/pull/27910#discussion_r4287647723
##########
components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java:
##########
@@ -683,6 +697,37 @@ 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.
+ */
+ private void recreateSubscription(final OpcUaSubscription subscription, final Optional<UInteger> 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();
+ }
Review Comment:
Agreed. With a closed connection it was worse than an orphan: the `reset()` and `create()` that followed waited for the session while holding `lock`, and with a producer on the connection that never ended (details in the main comment). The new version has no `reset()`. It creates the subscription again only when `deleteAsync()` returns `Good` or `Bad_SubscriptionIdInvalid`, that is, after Milo has reset the object itself in `applyDeleteResponse`. A lost connection or session is left to Milo's session recovery, and any other failure, such as a request timeout, rebuilds the connection through `handleConnectionFailure`, which closes the session together with the subscription. The `Bad_Timeout` path goes through the same code; there the delete completes without a service call, as you say.
_Claude Code on behalf of etzel-mes_
##########
components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java:
##########
@@ -683,6 +697,37 @@ 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.
+ */
+ private void recreateSubscription(final OpcUaSubscription subscription, final Optional<UInteger> 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");
+ }
Review Comment:
Done. The recreate clears `goodSubscriptions`, drops the monitored items that were reset with the subscription, and passes the subscriptions it held to `putSubscriptions`, so `goodSubscriptions` is rebuilt from the results. A node that cannot be monitored again is removed from the subscription and from `goodSubscriptions`, and is reported once to its consumer as a bad `DataValue` after `lock` has been released. Only a failed `create()` or a rejected service call rebuilds the connection. `nodeThatCannotBeMonitoredAgainIsReportedWithoutRebuildingTheConnection` deletes one of two monitored nodes on the embedded server before the subscription times out there, and checks that its consumer receives `Bad_NodeIdUnknown`, that the other node keeps delivering and that the `OpcUaClient` is the same instance. It fails with the first revision, which rebuilt the connection.
_Claude Code on behalf of etzel-mes_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25691: camel-milo - recreate the subscription when it stops publishing [camel]
Posted by "etzel-mes (via GitHub)" <gi...@apache.org>.
etzel-mes commented on PR #27910:
URL: https://github.com/apache/camel/pull/27910#issuecomment-6198200622
Thanks for the review.
**Lock.** Intentional, and I added the Javadoc note. The `OpcUaSubscription` is shared with `registerItems` and `unregisterItems`, which already run `createMonitoredItems` / `deleteMonitoredItems` under `lock` (through `whenConnected`), and `handleConnectionFailure` disposes the client under it. Milo's `OpcUaSubscription` has no synchronisation of its own, so `lock` is what serialises access to it. `performConnect` can run outside the lock because it works on a new client and a new subscription that nobody else sees yet; the recreation changes the subscription the consumers are using. Building a replacement outside the lock would still need the lock for `createMonitoredItems` and the swap, and a consumer started in between would register its items on the dead subscription. In the case this change is for, a server that dropped the subscription and answers immediately, the three calls take a few milliseconds in the tests.
**Timeout.** Every service call goes through Milo's `UascClientMessageHandler`, which fails a request with `Bad_Timeout` after the request timeout hint (`OpcUaClientConfig.requestTimeout`, 60 s by default; the endpoint's `requestTimeout` option sets it). So the worst case against an unresponsive server is one request timeout per call, after which the catch block falls back to `handleConnectionFailure`, the path a failed reconnect takes today. The Javadoc now says so.
The review mentions a process note that I cannot see on the PR; happy to address it if you post it.
_Claude Code on behalf of etzel-mes_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25691: camel-milo - recreate the subscription when it stops publishing [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on code in PR #27910:
URL: https://github.com/apache/camel/pull/27910#discussion_r4303037104
##########
components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorItemLifecycleTest.java:
##########
@@ -128,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);
Review Comment:
Done in 581cdf637118: the route of the first item now holds the delivery thread for the slow value with `.filter(...).delay(3000).syncDelayed().end()`, the check moved into `isSlowValueOnce(Exchange)`, so the test no longer calls `Thread.sleep()`. The branch also has main merged in (017e5c7097de) to pick up the `ServerCertificateValidationTest` fix of CAMEL-25645, with the 4.24 upgrade guide conflict resolved by keeping both sides.
_Claude Code on behalf of Claus Ibsen_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25691: camel-milo - recreate the subscription when it stops publishing [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on PR #27910:
URL: https://github.com/apache/camel/pull/27910#issuecomment-6236711658
@etzel-mes, #27906 (CAMEL-25671) is merged now, and it conflicts with this PR in `SubscriptionManager.java`: it removed `badSubscriptions` and the resubscribe that served the old `Bad_InvalidArgument` branch, while `handleSubscriptionError` and `resubscribe()` here still use them. Could you rebase onto main and resolve that against #27906? GitHub does not run the PR checks while the PR conflicts.
Two commits on your branch are from the review and should be kept:
- 581cdf637118 and 02986bba36e8: the slow route of `MonitorItemLifecycleTest` holds the delivery thread with `.filter(...).delay(3000).syncDelayed().end().end()` instead of `Thread.sleep(3000)` (the delay is a block of its own, hence the two `end()`).
- 017e5c7097de is a merge of main for the `ServerCertificateValidationTest` fix (CAMEL-25645); a rebase replaces it.
_Claude Code on behalf of Claus Ibsen_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25691: camel-milo - recreate the subscription when it stops publishing [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on code in PR #27910:
URL: https://github.com/apache/camel/pull/27910#discussion_r4298280367
##########
components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorItemLifecycleTest.java:
##########
@@ -128,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);
Review Comment:
Nit: CLAUDE.md asks new test code not to add `Thread.sleep()`. Here the sleep is the slow route under test rather than a wait for something, so keeping it is fine with a short comment that says so. If avoiding the call is preferred, the `delay` EIP with `syncDelayed()` inside a `filter` on the slow value holds the delivery thread the same way.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25691: camel-milo - recreate the subscription when it stops publishing [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on PR #27910:
URL: https://github.com/apache/camel/pull/27910#issuecomment-6231593157
FYI @etzel-mes: the `Build and test` failure here (`ServerCertificateValidationTest.onlyClientsThatCanValidateTheServerCertificateConnect`: `mock://trustAll Received message count. Expected: <1> but was: <2>`) does not come from this PR. It fails on `main` too (reproduced locally at 55ed20f14e90). CAMEL-25626 (#27908) and CAMEL-25645 (#27824) were merged 26 seconds apart on 2026-10-16, and neither PR's CI ran with the other change. With CAMEL-25626 the `trustAll` client now also receives the initial `DataValue[value=null, statusCode=Bad]` of `milo-server:myitem1` before `Foo`, which the new test does not expect. The test needs to accept that first Bad value, for example by checking only the Good values, as the tests updated in CAMEL-25626 do. Once that is fixed on `main`, a rebase should make this build green.
_Claude Code on behalf of Claus Ibsen. This comment was generated by an AI agent and may contain inaccuracies._
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25691: camel-milo - recreate the subscription when it stops publishing [camel]
Posted by "etzel-mes (via GitHub)" <gi...@apache.org>.
etzel-mes commented on PR #27910:
URL: https://github.com/apache/camel/pull/27910#issuecomment-6201032249
Thanks @davsclaus, the silent drop was the right question. I tried it, and the first revision did not always end in a usable state, so the recovery works differently now.
**Setup.** A real server instead of the embedded one: [opc-plc](https://github.com/Azure-Samples/iot-edge-opc-plc) 2.15.4, an OPC UA PLC simulator on the OPC Foundation .NET stack, in Docker, with each Camel client in its own container on Milo's defaults (keep-alive every 5 s, 5 s timeout, 1 failure allowed) and the default publishing interval of 1000 ms. The drop is an iptables `DROP` of the traffic to and from the server port in the client's network namespace, so nothing is closed or reset and both sides keep retransmitting, as with a pulled cable. It starts 8 s after a keep-alive PublishResponse, so the watchdog fired 7 to 8 s into the drop, 2.4 to 7.5 s before the keep-alive declared the session lost: the order you describe. A separate client writes a new value every second, and the times below run from the end of the drop to the first new value the consumer delivers. Each run dropped two clients at once, main and one revision.
| Drop | main | first revision | this revision |
|---|---|---|---|
| 30 s | 1.1 to 2.2 s | 9.6 s, connection rebuilt | 2.2 s, same session |
| 30 s, 1 CPU | 1.1 to 2.1 s | 26.5 s, connection rebuilt | 1.1 s, same session |
| 30 s, with a producer writing every 2 s on the same connection | 11.7 to 11.8 s, rebuilt after the failed writes | no value within 240 s | 11.8 s, rebuilt after the failed writes |
| 75 s: the subscription times out on the server, the session does not | no value within 240 s | 10.0 s, connection rebuilt | 8.9 s, same session |
| 150 s: the session times out as well | 19.7 s, rebuilt after the failed transfer | not run | 19.7 s, rebuilt after the failed transfer |
**First revision.** The blocking `delete()` went into the dead connection and only failed when the keep-alive closed the channel. `reset()` and `create()` then waited for the session on a thread of `Stack.sharedScheduledExecutor()` while holding `lock`. With a producer, the failed writes run `handleConnectionFailure` on the same executor, and those tasks parked on `lock` until they held the remaining threads (the thread dump showed four in `handleConnectionFailure` from `write` and one in `OpcUaSubscription.create`). Milo schedules the session reconnect on that executor too, so the session never came back and `create()` never returned. Without a producer, `create()` failed while the network was still down, and the fallback rebuilt a connection that Milo would otherwise have resumed.
**This revision.** Both callbacks call `deleteAsync()`, without `lock` and without waiting for it, and act on its result:
- `Good` or `Bad_SubscriptionIdInvalid`, the answers on which Milo resets the object itself: the server has confirmed in this session that the subscription is gone. Only then is it created again under `lock`, with its monitored items going through `putSubscriptions`. For `Bad_Timeout`, Milo has already reset the subscription, so the delete completes without a service call.
- `Bad_ConnectionClosed`, `Bad_SessionClosed`, `Bad_SessionIdInvalid` or `Bad_SessionNotActivated`: nothing more. Milo's session recovery resumes the subscription and arms its watchdog again when publishing resumes (`PublishingManager.resumePublishing`), so a subscription that turns out to be gone is caught again on the live session. A failed transfer still rebuilds through `onTransferFailed`.
- Anything else, such as a request timeout: the connection is rebuilt, which closes the session and the subscription with it.
So nothing waits on a dead connection while holding `lock`: it is taken for `create()` and the monitored items only after the server has answered the delete. This follows the Milo Javadoc of `onWatchdogTimerElapsed` ("Consider deleting and creating a new Subscription") and leaves the connection-level recovery to Milo.
In the 30 s runs the delete failed with `Bad_ConnectionClosed`, and the server's diagnostics show the original subscription alive after the recovery, so a delete that failed with the connection was not applied later. Three more runs covered the other orders, each followed by a minute of values written every second, during which both clients kept delivering values with no gap longer than 2.1 s:
- A drop starting 2 s after a keep-alive: the session is declared lost before the watchdog fires, the watchdog is paused, and Milo recovers alone (1.1 s for both).
- A drop starting 5 s after a keep-alive: the watchdog fires first, the delete fails with the connection, and the subscription keeps its id (1.1 s for both).
- A 10 s drop that ends before the keep-alive gives up: the delete reaches the server once the network is back, and the subscription is created again in the same session. The first value arrived after 5.4 s, against 4.4 s for main, which resumed the old subscription.
The 75 s row is the case the PR is about, caused by the network: the session survives the drop and the subscription does not. main only logs the `Bad_Timeout` that the server reports after the reconnect, and stays silent; this revision creates the subscription again 26 ms later, in the same session.
**Without a network fault.** opc-plc has a chaos mode that, 56 times in 31 minutes here, deletes a subscription without telling the client, closes one session (keeping or deleting its subscriptions) or all sessions, closes all subscriptions or injects service faults. I ran one client per variant against it, with a third client writing a value every second:
- main was silent for 24 of the 30 minutes. Each deleted subscription left it without values until the chaos monkey happened to close its session, which failed the transfer and rebuilt the connection: five outages of 110 to 829 s, and it was silent again at the end.
- This revision was silent for 7.7 minutes, in 25 outages of about 17.6 s. Its subscription was created again in the same session 25 times, 24 times after the 16.5 s watchdog and once after a `Bad_Timeout` from the server; the longest outage, 34 s, was a recreated subscription that the next chaos round deleted right away. Two deletes took the other branches: one was answered with `Bad_SessionIdInvalid` because the server had closed the session, so Milo's recovery opened a new session and the failed transfer rebuilt the connection; the other was answered with an injected `Bad_SecureChannelClosed` fault and rebuilt the connection directly, which had a new subscription 46 ms later. Sessions the chaos monkey closed rebuilt it through `onTransferFailed`, as on main.
**Docs.** The section now covers the silent drop: the delete fails with the connection, the client restores the session once the server can be reached again, and the subscription resumes publishing, or is created again when the server reports that it timed out in the meantime. It also says that a network that is back before the client has noticed the loss has the subscription created again.
**A correction to my earlier reply.** Milo 1.1.8 does serialize the lifecycle transitions of `OpcUaSubscription` (`create`, `modify`, `delete`) itself; only the monitored item management is not serialized. That is why the delete can run without `lock` now, while `create()` and the monitored items still take it.
@gnodet-bot, your two follow-up findings are the two inline threads; both are answered there.
_Claude Code on behalf of etzel-mes_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25691: camel-milo - recreate the subscription when it stops publishing [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on code in PR #27910:
URL: https://github.com/apache/camel/pull/27910#discussion_r4291166014
##########
components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorItemLifecycleTest.java:
##########
@@ -241,6 +248,128 @@ void failedCreateMonitoredItemsCallRebuildsTheConnection() throws Exception {
.getBody(DataValue.class)));
}
+ @Test
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ void subscriptionThatStopsPublishingIsRecreatedInTheSameSession() throws Exception {
+ item1Endpoint.expectedMinimumMessageCount(1);
+ testBody(item1Endpoint.message(0), assertGoodValue("Foo"));
+ producer1.sendBody("Foo");
+ 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 {
+ item1Endpoint.expectedMinimumMessageCount(1);
+ testBody(item1Endpoint.message(0), assertGoodValue("Foo"));
+ producer1.sendBody("Foo");
+ 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();
+
+ timeOutSubscription(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);
+ }
+ }
+
+ @Test
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ void nodeThatCannotBeMonitoredAgainIsReportedWithoutRebuildingTheConnection() throws Exception {
+ producer1.sendBody("Foo");
+ producer2.sendBody("Bar");
+
+ 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());
Review Comment:
The await above returns as soon as `create()` installs the new SubscriptionId, which happens before `putSubscriptions` has added and created the monitored items. In that window `getMonitoredItems()` reports 0 (the items were removed in `recreate()`) or 3 (all added, the failed one not yet removed), so this assertion can fail intermittently. Waiting for the count avoids it:
```suggestion
await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(2, subscription.getMonitoredItems().size()));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25691: camel-milo - recreate the subscription when it stops publishing [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus merged PR #27910:
URL: https://github.com/apache/camel/pull/27910
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25691: camel-milo - recreate the subscription when it stops publishing [camel]
Posted by "etzel-mes (via GitHub)" <gi...@apache.org>.
etzel-mes commented on PR #27910:
URL: https://github.com/apache/camel/pull/27910#issuecomment-6237104696
@davsclaus, rebased onto main (94ca5018123) and resolved against #27906. In `SubscriptionManager`, `lookupMonitoredNode` from #27906 stays and `recreate(...)` follows it; `handleSubscriptionError`, `resubscribe()` and `badSubscriptions` are gone. `recreate(...)` only recreates the items in `goodSubscriptions`, so a node with an unknown namespace URI is still reported once and not retried. The 4.24 upgrade guide conflict is resolved as in your merge.
Your two commits are kept on top, now 202f1e2b45e and 6f24e0573df; the rebase replaced 017e5c7. `camel-milo` with `-Psourcecheck clean verify`: 79 tests, 0 failures, 1 skipped; `MonitorItemLifecycleTest` and `MonitorNodesTest` also passed three runs each with `-Dsurefire.rerunFailingTestsCount=0`.
_Claude Code on behalf of etzel-mes_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25691: camel-milo - recreate the subscription when it stops publishing [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on code in PR #27910:
URL: https://github.com/apache/camel/pull/27910#discussion_r4286228686
##########
components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java:
##########
@@ -683,6 +697,37 @@ 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.
+ */
+ private void recreateSubscription(final OpcUaSubscription subscription, final Optional<UInteger> 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();
+ }
Review Comment:
**`reset()` here can leave a live subscription running on the server.**
When the server no longer has the subscription, `delete()` gets `Bad_SubscriptionIdInvalid`. Milo then resets the object itself in `applyDeleteResponse` before it throws, so this `reset()` is a no-op in the case the PR targets. It only matters for other failures: a request timeout on a slow server, or a closed connection. In those cases the subscription may still exist on the server. After `reset()` + `create()`, the old one is abandoned in the same session, where it keeps its monitored items. It can stay alive as long as the session keeps sending Publish requests, and Milo discards its responses. PLC servers often have tight subscription limits, so a leak like this matters there.
Suggestion: only continue in place when the server confirmed the subscription is gone. Otherwise let the outer `catch` fall back to `handleConnectionFailure`, which closes the session and the subscriptions on it:
```suggestion
try {
subscription.delete();
} catch (final UaException e) {
if (e.getStatusCode().getValue() != StatusCodes.Bad_SubscriptionIdInvalid) {
// the subscription may still exist on the server, rebuilding the connection closes its session
throw e;
}
LOG.debug("Subscription {} no longer exists on the server", id.orElse(null));
}
```
(For the `Bad_Timeout` path, Milo has already reset the subscription, so `delete()` returns without calling the server.)
##########
components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java:
##########
@@ -683,6 +697,37 @@ 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.
+ */
+ private void recreateSubscription(final OpcUaSubscription subscription, final Optional<UInteger> 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");
+ }
Review Comment:
**One item that cannot be created again rebuilds the whole connection.**
`isGood()` needs both the service result and the operation result to be good, so one operation-level failure fails the whole check. For example, the PR description mentions a server configuration reload: a node removed by that reload comes back as `Bad_NodeIdUnknown`. The check then throws, and `handleConnectionFailure` closes the session for every consumer and producer on the connection. That is the cost the PR sets out to avoid. It also handles this case differently from `Connected.putSubscriptions`. There, operation-level failures are removed from the subscription, reported once to their consumer as a bad `DataValue` (outside the lock), and do not cause a rebuild. Only a bad *service* result rebuilds.
Could the recreate reuse that handling, so that only a rejected service call falls back to the rebuild? Then the failed item should also be removed from `goodSubscriptions`, so that a later `deactivate` does not try to delete it. A test that removes one monitored node on the server before the recreate would cover it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25691: camel-milo - recreate the subscription when it stops publishing [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27910:
URL: https://github.com/apache/camel/pull/27910#issuecomment-6199800703
<!-- ci-tested-modules -->
:test_tube: **CI tested the following changed modules:**
- `catalog/camel-catalog`
- `components/camel-milo`
- `docs`
---
<details><summary>:microscope: Scalpel shadow comparison — Scalpel: 8 of 705 tested, 25 compile-only — current: 8 all tested</summary>
[Maveniverse Scalpel](https://github.com/maveniverse/scalpel) detected **8 affected modules** (current approach: 8).
**Skip-tests mode would test 8 modules** (3 direct + 7 downstream), **skip tests for 25** (generated code, meta-modules)
<details><summary>Modules Scalpel would test (8)</summary>
- `camel-jbang-mcp` ← downstream of org.apache.camel:camel-catalog
- `camel-jbang-plugin-mcp` ← downstream of org.apache.camel:camel-jbang-core
- `camel-jbang-plugin-tui` ← downstream of org.apache.camel:camel-catalog
- `camel-jbang-plugin-validate` ← downstream of org.apache.camel:camel-yaml-dsl-validator
- `camel-launcher-container` ← downstream of org.apache.camel:camel-launcher
- `camel-milo` ← components/camel-milo/src/main/docs/milo-client-component.adoc, components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java, components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorItemLifecycleTest.java
- `camel-yaml-dsl-validator` ← downstream of org.apache.camel:camel-catalog
- `camel-yaml-dsl-validator-maven-plugin` ← downstream of org.apache.camel:camel-yaml-dsl-validator
</details>
<details><summary>Modules with tests skipped (25)</summary>
- `apache-camel`
- `camel-allcomponents`
- `camel-catalog-console`
- `camel-catalog-maven`
- `camel-catalog-suggest`
- `camel-componentdsl`
- `camel-endpointdsl`
- `camel-endpointdsl-support`
- `camel-itest`
- `camel-jbang-core`
- `camel-jbang-it`
- `camel-jbang-main`
- `camel-jbang-plugin-edit`
- `camel-jbang-plugin-generate`
- `camel-jbang-plugin-kubernetes`
- `camel-jbang-plugin-test`
- `camel-kamelet-main`
- `camel-launcher`
- `camel-report-maven-plugin`
- `camel-route-parser`
- `camel-yaml-dsl`
- `camel-yaml-dsl-deserializers`
- `camel-yaml-dsl-maven-plugin`
- `coverage`
- `dummy-component`
</details>
> :information_source: Shadow mode — Scalpel observes but does not affect test execution. [Learn more](https://github.com/maveniverse/scalpel)
</details>
<details><summary>All tested modules (35 modules, 5m 51s total)</summary>
**Total reactor time:** 5m 51s
| Module | Duration | Status |
| --- | --- | --- |
| Camel :: Milo | 54.8s | SUCCESS |
| Camel :: Launcher | 48.6s | SUCCESS |
| Camel :: JBang :: MCP | 38.9s | SUCCESS |
| Camel :: JBang :: Plugin :: Validate | 33.9s | SUCCESS |
| Camel :: Component DSL | 23.3s | SUCCESS |
| Camel :: Catalog :: Camel Catalog | 21.0s | SUCCESS |
| Camel :: YAML DSL :: Validator | 18.7s | SUCCESS |
| Camel :: YAML DSL | 17.7s | SUCCESS |
| Camel :: Docs | 14.4s | SUCCESS |
| Camel :: Kamelet Main | 14.0s | SUCCESS |
| Camel :: JBang :: Plugin :: Kubernetes | 11.6s | SUCCESS |
| Camel :: YAML DSL :: Deserializers | 7.4s | SUCCESS |
| Camel :: Catalog :: Camel Route Parser | 7.3s | SUCCESS |
| Camel :: JBang :: Plugin :: Testing | 6.6s | SUCCESS |
| Camel :: Catalog :: Camel Report Maven Plugin | 6.3s | SUCCESS |
| Camel :: All Components Sync point | 4.8s | SUCCESS |
| Camel :: YAML DSL :: Validator Maven Plugin | 3.5s | SUCCESS |
| Camel :: YAML DSL :: Maven Plugins | 2.7s | SUCCESS |
| Camel :: Catalog :: Maven | 2.6s | SUCCESS |
| Camel :: Catalog :: Suggest (deprecated) | 2.1s | SUCCESS |
| Camel :: JBang :: Plugin :: Edit | 1.6s | SUCCESS |
| Camel :: Assembly | 1.5s | SUCCESS |
| Camel :: Catalog :: Dummy Component | 1.1s | SUCCESS |
| Camel :: JBang :: Plugin :: Generate | 1.1s | SUCCESS |
| Camel :: Catalog :: Console | 1.0s | SUCCESS |
| Camel :: Coverage | 1.0s | SUCCESS |
| Camel :: Endpoint DSL :: Support | 1.0s | SUCCESS |
| Camel :: JBang :: Main | 0.9s | SUCCESS |
| Camel :: JBang :: Integration tests | 0.7s | SUCCESS |
| Camel :: Launcher :: Container | 0.7s | SUCCESS |
| Camel :: JBang :: Plugin :: MCP | 0.5s | SUCCESS |
| Camel :: Endpoint DSL | n/a | |
| Camel :: Integration Tests | n/a | |
| Camel :: JBang :: Core | n/a | |
| Camel :: JBang :: Plugin :: TUI | n/a | |
**Top 20 slowest modules:**
- `Camel :: Milo` (54.8s)
- `Camel :: Launcher` (48.6s)
- `Camel :: JBang :: MCP` (38.9s)
- `Camel :: JBang :: Plugin :: Validate` (33.9s)
- `Camel :: Component DSL` (23.3s)
- `Camel :: Catalog :: Camel Catalog` (21.0s)
- `Camel :: YAML DSL :: Validator` (18.7s)
- `Camel :: YAML DSL` (17.7s)
- `Camel :: Docs` (14.4s)
- `Camel :: Kamelet Main` (14.0s)
- `Camel :: JBang :: Plugin :: Kubernetes` (11.6s)
- `Camel :: YAML DSL :: Deserializers` (7.4s)
- `Camel :: Catalog :: Camel Route Parser` (7.3s)
- `Camel :: JBang :: Plugin :: Testing` (6.6s)
- `Camel :: Catalog :: Camel Report Maven Plugin` (6.3s)
- `Camel :: All Components Sync point` (4.8s)
- `Camel :: YAML DSL :: Validator Maven Plugin` (3.5s)
- `Camel :: YAML DSL :: Maven Plugins` (2.7s)
- `Camel :: Catalog :: Maven` (2.6s)
- `Camel :: Catalog :: Suggest (deprecated)` (2.1s)
</details>
---
:gear: [View full build and test results](https://github.com/apache/camel/actions/runs/38777725541)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25691: camel-milo - recreate the subscription when it stops publishing [camel]
Posted by "etzel-mes (via GitHub)" <gi...@apache.org>.
etzel-mes commented on code in PR #27910:
URL: https://github.com/apache/camel/pull/27910#discussion_r4296942625
##########
components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorItemLifecycleTest.java:
##########
@@ -241,6 +248,128 @@ void failedCreateMonitoredItemsCallRebuildsTheConnection() throws Exception {
.getBody(DataValue.class)));
}
+ @Test
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ void subscriptionThatStopsPublishingIsRecreatedInTheSameSession() throws Exception {
+ item1Endpoint.expectedMinimumMessageCount(1);
+ testBody(item1Endpoint.message(0), assertGoodValue("Foo"));
+ producer1.sendBody("Foo");
+ 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 {
+ item1Endpoint.expectedMinimumMessageCount(1);
+ testBody(item1Endpoint.message(0), assertGoodValue("Foo"));
+ producer1.sendBody("Foo");
+ 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();
+
+ timeOutSubscription(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);
+ }
+ }
+
+ @Test
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ void nodeThatCannotBeMonitoredAgainIsReportedWithoutRebuildingTheConnection() throws Exception {
+ producer1.sendBody("Foo");
+ producer2.sendBody("Bar");
+
+ 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());
Review Comment:
Applied in 70aca32 as the condition of the await, with one addition: it also waits for `isMonitoredItemsSynchronized()`. `putSubscriptions` adds the items one by one before it creates them, so the count alone passes 2 on its way to 3.
_Claude Code on behalf of etzel-mes_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org