Repository navigation
CAMEL-25691: camel-milo - recreate the subscription when it stops publishing - #27910
Conversation
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Two observations on recreateSubscription and one process note.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
a07bc72 to
d96ef62
Compare
|
Thanks for the review. Lock. Intentional, and I added the Javadoc note. The Timeout. Every service call goes through Milo's 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 |
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Claude Code on behalf of davsclaus
Thanks @etzel-mes, this is a well-motivated fix with a clear write-up. Recreating the subscription inside the existing session instead of tearing down the shared connection is the right call for session-limited PLC servers, and the tests drive both callbacks through real server behaviour rather than mocks. I read the change against the Milo 1.1.8 sources (OpcUaSubscription, PublishingManager). The two points below are inline:
reset()after a faileddelete()can orphan a live subscription (inline, with a suggestion). Milo already resets the subscription itself when the server answersBad_SubscriptionIdInvalid. The explicitreset()therefore only takes effect for the other failures (request timeout, connection closed), and in those cases the subscription may still exist on the server.- A single monitored item that cannot be created again rebuilds the whole connection (inline). This is the case the PR wants to avoid.
putSubscriptionsalready handles per-item failures without a rebuild.
Question: a silent network drop. The watchdog is only paused when the session goes inactive (WatchdogTimer.onSessionInactive). With Milo's defaults (keep-alive every 5 s, 5 s timeout, 1 failure allowed), a silent drop such as a pulled cable or a switch reboot is detected after about 10 to 15 s. At Camel's default publishing interval of 1000 ms, the watchdog fires 1000 x (10 + 1) x 1.5 = 16.5 s after the last PublishResponse. For a subscription that only gets keep-alives (about every 10 s), it can therefore fire before the session is declared inactive. Previously Milo's session recovery (ActivateSession / TransferSubscriptions / Republish) handled that case alone. Now recreateSubscription runs on the dead session, while holding the manager lock, and steps in. Did you try this, e.g. by dropping the traffic to the port instead of deleting the subscription? If recovery still ends in a usable state, it is worth a sentence in the docs. If it does not, deferring the recreate until the session is confirmed active may be needed.
On the two open threads from the earlier review: the blocking delete() / create() / createMonitoredItems() calls cannot block indefinitely. Each is bounded by the client requestTimeout (Milo default 60 s). registerItems / putSubscriptions already make createMonitoredItems calls under the same lock, so the pattern itself is not new. The difference is that three sequential round trips can hold the lock for up to three request timeouts, which blocks producers on the shared connection. That supports the first thread's request to document the trade-off, or to narrow what runs under the lock.
Docs (component page plus catalog copy) and the 4.24 upgrade-guide entry look good.
This review covers conventions and the code paths I traced statically. It does not replace specialized AI review tools or static analysis such as SonarCloud.
This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
I'm satisfied with the Javadoc additions in recreateSubscription() — the blocking-I/O-under-lock trade-off is now documented and the 60 s request timeout bound is explicit. Those two prior threads are addressed.
However, two findings from @davsclaus's review remain unaddressed in this revision:
1. reset() subscription leak (line 721)
} catch (final UaException e) {
LOG.debug("Failed to delete subscription {}", id.orElse(null), e);
subscription.reset();
}The catch block is unconditional on the exception status code. When delete() fails for any reason other than Bad_SubscriptionIdInvalid (network blip, transient server error, etc.) the subscription may still be alive on the server. Calling reset() followed by create() leaves that server-side subscription dangling in the same session. On PLC servers with tight subscription limits this is a resource leak that accumulates across repeated recoveries. The fix is to check the status code and only call reset() on Bad_SubscriptionIdInvalid; for all other failures, propagate the exception so the outer catch (Exception e) → handleConnectionFailure path applies.
2. allMatch(isGood) over-reaction (line 724)
if (!subscription.createMonitoredItems().stream().allMatch(MonitoredItemServiceOperationResult::isGood)) {
throw new UaException(StatusCodes.Bad_UnexpectedError, "Failed to create the monitored items again");
}A single operation-level failure for one monitored item (e.g. Bad_NodeIdUnknown for a removed tag) throws here, which the outer catch (Exception e) turns into handleConnectionFailure — tearing down the entire connection for all producers/consumers sharing it. This directly contradicts the resilience goal of this PR. The existing putSubscriptions path already handles operation-level monitored-item failures gracefully (per @davsclaus's comment). The check here should mirror that approach: log per-item failures and continue, rather than treating any item-level failure as a fatal connection error.
Both issues are production-relevant and should be addressed before merge.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
|
🧪 CI tested the following changed modules:
🔬 Scalpel shadow comparison — Scalpel: 8 of 706 tested, 25 compile-only — current: 8 all testedMaveniverse 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) Modules Scalpel would test (8)
Modules with tests skipped (25)
All tested modules (35 modules, 5m 54s total)Total reactor time: 5m 54s
Top 20 slowest modules:
|
|
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 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
First revision. The blocking This revision. Both callbacks call
So nothing waits on a dead connection while holding In the 30 s runs the delete failed with
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 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:
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 @gnodet-bot, your two follow-up findings are the two inline threads; both are answered there. Claude Code on behalf of etzel-mes |
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Following up on my earlier review — both findings from the second review have been addressed.
1. reset() subscription leak ✅ — The new design drops the synchronous delete() + conditional reset() entirely. deleteAsync() is called without holding lock, and the completion callback only proceeds to recreateSubscription() when the server confirms the subscription is gone (GOOD or Bad_SubscriptionIdInvalid). Session-loss codes defer to Milo's recovery; any other failure calls handleConnectionFailure. The leak scenario is eliminated.
2. allMatch(isGood) over-reaction ✅ — recreate() now delegates to putSubscriptions(), which already handles per-item failures gracefully: operation-level failures are logged and reported per-node as a bad DataValue, without triggering a connection rebuild. Only a rejected service call throws. The new nodeThatCannotBeMonitoredAgainIsReportedWithoutRebuildingTheConnection test covers this path directly.
The async delete design also resolves the blocking-I/O-under-lock concern from the first review: deleteAsync() runs lock-free, and manager.create() inside recreateSubscription() now holds the lock only for the create + monitored-items phase, with the 60 s Milo request timeout bounding the worst case, as documented in the Javadoc.
Three integration tests added, each exercising a distinct failure mode. LGTM.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Claude Code on behalf of davsclaus
Thanks @etzel-mes, the redesign addresses both points from my last review, and the silent-drop measurements answer my question. The delete now runs asynchronously without the manager lock, and the subscription is only recreated after the server confirmed it is gone (I checked this against Milo 1.1.8 applyDeleteResponse / reset()). A node that cannot be monitored again goes through putSubscriptions instead of rebuilding the connection. One small test-timing nit inline.
Heads-up for merge order: this PR and #27908 conflict in MonitorItemLifecycleTest. #27908 makes item1 autoStartup(false) and delivers the startup Bad status, so whichever merges second needs to start item1 after the first write in the three recreate tests.
CI has not run on this head yet (the workflows are waiting for approval), so this approval assumes it goes green.
This review does not replace specialized AI review tools or static analysis such as SonarCloud.
This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
allthingssecurity
left a comment
allthingssecurity
left a comment
There was a problem hiding this comment.
Thanks @etzel-mes, the silent-drop measurements are very convincing. I found one case the new watchdog handling
turns into a false positive: a route that is slow to process a value.
A slow route makes a healthy subscription look dead (SubscriptionManager.java:121 and :741)
Milo only releases a pending Publish permit after the notification has been delivered
(PublishingManager.deliverAndReleasePendingPublish: "Waiting until the client has finished receiving notifications
is the backpressure mechanism"). The consumer runs the route synchronously inside that delivery
(MiloClientConsumer.handleValueUpdate -> getProcessor().process(exchange)). So while one exchange is slow, the
remaining Publish requests are answered, their responses queue behind it, and the server has nothing left to answer
with. After the watchdog delay (16.5 s with the defaults: 1000 ms publishing interval, keep-alive count 10) the
watchdog fires. Before this PR that was harmless. Now onWatchdogTimerElapsed calls recoverSubscription.
The callback runs on the subscription's delivery queue, so it only runs after the slow exchange has returned. By then
publishing has resumed. Milo checks the timer epoch when the timer fires, not when the callback runs, so the callback
still runs. deleteAsync() is then answered Good, because the subscription is alive, and recreateSubscription
creates it again. Every consumer on the connection gets its current value again. A route that stays slower than the
watchdog (for example, retries against a downstream system that is down) repeats this cycle each time the current
value is delivered again.
Reproducer against the embedded server (same setup as MonitorItemLifecycleTest). The route sleeps once when it
sees the value Slow, then the test checks the delivery count and the subscription id:
from(resolve(MILO_CLIENT_ITEM_1)).routeId("item1")
.process(exchange -> {
Object v = exchange.getMessage().getBody(DataValue.class).getValue().getValue();
if ("Slow".equals(v) && blockedOnce.compareAndSet(false, true)) {
Thread.sleep(3000);
}
})
.to("mock:item1");
// test: shorten the watchdog as in the PR's tests, send "Slow", wait for it, then 3 s more
subscription.setMaxKeepAliveCount(uint(1));
subscription.modify();
UInteger subscriptionId = subscription.getSubscriptionId().orElseThrow();
producer1.sendBody("Slow");
await().atMost(20, TimeUnit.SECONDS).until(() -> countValue("Slow") >= 1);
Thread.sleep(3000);
assertEquals(1, countValue("Slow"));
assertEquals(subscriptionId, subscription.getSubscriptionId().orElse(null));On 9c6d4cf it fails 3 out of 3 runs (expected: <1> but was: <2>; subscription 12 becomes 13), and the log shows:
WARN SubscriptionManager - Subscription 12 stopped publishing, recreating it
INFO SubscriptionManager - Subscription recreated as 13
A second variant with Camel's and Milo's defaults (no modify(), default publishing interval) and a route that
sleeps 30 s once also fails: Slow is delivered twice and the subscription id changes. Both tests pass against
main's SubscriptionManager.
Suggestion: before deleting, check whether the subscription still exists, with a call that does not change it. For
example, subscription.setPublishingModeAsync(true) is answered Good for a live subscription and
Bad_SubscriptionIdInvalid for one the server no longer has. Only the second answer, or Bad_InvalidState when Milo
has already reset the object after Bad_Timeout, would then go on to deleteAsync() and the recreate. The
Bad_NoSubscription and Bad_Timeout cases this PR is for would still recover. A healthy subscription behind a slow
route would be left alone. A test like the one above would cover it.
Claude Code on behalf of allthingssecurity
9c6d4cf to
70aca32
Compare
|
Thanks @allthingssecurity, confirmed. Milo resets the watchdog when a PublishResponse is received ( Fixed in 70aca32 with the check you suggested. After the watchdog,
A Testing this against a real server also turned up a hang in 9c6d4cf. It handled the outcome of the delete on 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 Tests:
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):
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 @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 Claude Code on behalf of etzel-mes |
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Thanks @etzel-mes, I re-checked the new commits. d3c527e is the revision I approved, rebased onto main with the tests starting item1 after the first write (#27908); 70aca32 is new. Checking the subscription before deleting it is the right answer to @allthingssecurity's slow-route report, and the real-server runs cover the orders that the embedded server cannot reproduce. I read 70aca32 against the Milo 1.1.8 sources:
setPublishingModeAsyncis serialized with the other lifecycle transitions and fails withBad_InvalidStatewhen the subscription is already reset or areset()overtakes the request, which is the "already being recovered" branch. AGoodanswer replaces theServerState, which the new test asserts.createalways asks forpublishingEnabled=true, so the check does not change the subscription.SESSION_LOSTnow matchesSESSION_ERRORand theSECURE_CHANNEL_ERRORofSessionFsmFactory.SessionFaultListener, plusBad_ConnectionClosed.onSessionActiveandonSessionInactivefire on every transition into and out ofActivethrough a sequential executor, so thesessionActiveflag sees them in order, and registering the listener beforeconnect()catches the first activation.- The transport executor defaults to
Stack.sharedExecutor(), an unbounded cached pool, whileStack.sharedScheduledExecutor()hasavailableProcessors()threads, so handling the outcome of the check and the delete on the client's executor removes the wait on a single scheduled thread.
My test-timing thread is fixed: the await now also waits for isMonitoredItemsSynchronized(), which covers the count passing 2 on its way to 3.
Nit (inline): the slow route uses Thread.sleep.
Merge state: this PR and #27906 conflict in SubscriptionManager: #27906 removes handleSubscriptionError and resubscribe right below where this PR adds recreate(). Keeping recreate() and lookupMonitoredNode and dropping the two removed methods resolves it. Neither PR conflicts with #27956.
CI: the workflows on 70aca32 are waiting for approval, so nothing has run on this head yet; this approval assumes they go green.
Checked against the project conventions in CLAUDE.md. This is a rules-and-conventions review and does not replace CodeRabbit, Sourcery or SonarCloud.
Claude Code on behalf of @davsclaus.
This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
|
FYI @etzel-mes: the Claude Code on behalf of Claus Ibsen. This comment was generated by an AI agent and may contain inaccuracies. |
|
@etzel-mes, #27906 (CAMEL-25671) is merged now, and it conflicts with this PR in Two commits on your branch are from the review and should be kept:
Claude Code on behalf of Claus Ibsen |
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Clean re-review. The async subscription recovery is well-designed: deleteAsync runs lock-free, recreateSubscription acquires the lock only for the create+monitored-items phase, and the isCurrent/subscriptionId().isPresent() guard correctly prevents double-recovery races.
Key design points verified:
- SESSION_LOST set is correct — it mirrors Milo 1.1.8
SessionFsmFactoryexactly:SESSION_ERROR ∪ SECURE_CHANNEL_ERROR + Bad_ConnectionClosed.Bad_RequestTypeInvalidbelongs because it is explicitly inSECURE_CHANNEL_ERRORin Milo's predicate. - No double-recreate race —
recreateSubscriptionacquiresReentrantLockthen checkssubscriptionId().isPresent(); the second concurrent call sees the ID already present and returns immediately. - No deadlock risk — the lock is
ReentrantLock;handleConnectionFailurecalled from withinrecreateSubscription's catch block re-enters successfully. - No state loss on recreate failure —
recreate()clearsgoodSubscriptionslocally; on failurehandleConnectionFailuredisposes the entireConnectedand rebuilds from the outerthis.subscriptionsmap.
The SessionActivityListener addition for the sessionActive volatile flag is a sound approach to catch requests that fail after connection loss with non-SESSION_LOST status codes. The four integration tests cover the critical paths. Documentation in both the component page and upgrade guide is thorough.
Merge-ready pending CI.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
…lishing 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 <noreply@anthropic.com>
… 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 <noreply@anthropic.com>
…as 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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01YTrG4zh2t5bq7jniA5Nwvg
…us 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) <noreply@anthropic.com> Signed-off-by: Claus Ibsen <claus.ibsen@gmail.com>
…e 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) <noreply@anthropic.com> Signed-off-by: Claus Ibsen <claus.ibsen@gmail.com>
02986bb to
6f24e05
Compare
|
@davsclaus, rebased onto main (94ca501) and resolved against #27906. In Your two commits are kept on top, now 202f1e2 and 6f24e05; the rebase replaced 017e5c7. Claude Code on behalf of etzel-mes |
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Thanks @etzel-mes for the rebase. I compared it with the revision I approved (git range-diff of 581cdf6 against 6f24e05):
SubscriptionManager: the three commits are unchanged apart from context.recreate()now sits afterlookupMonitoredNodefrom #27906, andhandleSubscriptionError,resubscribe()andbadSubscriptionsare gone.recreate()still goes throughputSubscriptions, which after #27906 also reports a node with an unknown namespace URI once and does not retry it. Recreating only thegoodSubscriptionsmatches that: a node that failed earlier is created again when its consumer restarts or the connection is rebuilt, as onmain.- The upgrade-guide entry stays in the camel-milo monitored-items section, before camel-protobuf.
- 6f24e05 adds the missing
end()for the delay block, somock:item1is outside the filter again.
CI is green on 6f24e05 (JDK 17 and 25).
Claude Code on behalf of @davsclaus.
This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
Description
Fixes CAMEL-25691.
SubscriptionManager.SubscriptionListenerImplignoredonWatchdogTimerElapsedand only loggedonStatusChanged. When the server drops the subscription while the session stays alive (subscription lifetime expired after a network or scheduling stall, server configuration reload,Bad_NoSubscriptionon 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_Timeoutis the same case reported by the server: Milo resets the subscription before the callback, and nothing created it again.With this change the watchdog first checks the subscription on the server with
setPublishingModeAsync(true), without holding the connection lock and without waiting for it. The subscription is always created with publishing enabled, so the request does not change its settings. The watchdog also elapses while a route is slow, because Milo replaces a publish request only after its notifications have been processed, so the check decides:Good: the subscription is kept.Bad_InvalidState: nothing. Milo resets a subscription only after aBad_Timeout, a delete or a failed transfer, and each of those already recovers it.SessionActivityListenerregistered before the client connects.deleteAsync()and created again inside the existing session once the server has confirmed that it is gone (GoodorBad_SubscriptionIdInvalid, the answers on which Milo resets theOpcUaSubscriptionitself), as the Milo Javadoc ofonWatchdogTimerElapsedsuggests ("Consider deleting and creating a new Subscription").A
Bad_Timeoutstatus change from the server (the subscription timed out there) goes to the delete and recreate directly, without the check. The monitored items are created again throughConnected.putSubscriptions, so a node that can no longer be monitored is reported to its consumer as a badDataValueand does not rebuild the connection. A delete that fails because the connection or the session was lost, or that fails while the session is not active, is left to Milo's session recovery as well: it resumes the subscription and arms the watchdog again when publishing resumes, or reports a failed transfer, which rebuilds the connection as before. Any other failure of the delete, or a subscription that cannot be created again, rebuilds the connection, which closes the session together with its subscriptions. The outcome of the check and of the delete is handled on the client's executor, not on Milo's shared scheduled executor: the recreate and the rebuild block until the server answers, and the client schedules its keep-alives and reconnect attempts on that executor, which has a single thread on a machine with one processor. A recreate in flight there kept the keep-alive from noticing a lost server until the request timed out, and a rebuild that waited for a reconnect attempt never got it. A failed producer request still rebuilds the connection on the scheduled executor and can reach the same wait; that path predates this PR and is left for a separate change (details in the review thread). Milo also asks that blocking calls are not made on the client's executor, because its threads complete the responses; camel-milo does not set the transport's executor, so that executor is Milo's shared cached pool, which starts another thread for a response while one waits, as it does while a route processes a value.Not handled: a subscription that the server confirms and then removes, or stops serving, before its next publish response is not checked again, because after a confirmation the watchdog is armed again only by the next publish response, a reactivated session, or a create or modify.
Why recreate the subscription instead of rebuilding the connection
A
MiloClientConnectionis 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 does not touch the other users of the connection and leaves the session count unchanged. The full rebuild stays as the fallback.Not changed:
onTransferFailedstill 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 withBad_NoSubscription), and asserts that a new value arrives, theOpcUaClientis the same instance and the subscription id changed.subscriptionThatTimesOutOnTheServerIsRecreatedInTheSameSession: the embedded server sends aBad_Timeoutstatus change and removes the subscription.subscriptionThatASlowRouteHeldUpIsKept: shortens the watchdog the same way and has the route take 3 s for one value, so the watchdog elapses while the server still has the subscription; asserts that the value is delivered once, the subscription id is unchanged and the check was answered (ServerStatereplaced).nodeThatCannotBeMonitoredAgainIsReportedWithoutRebuildingTheConnection: removes one monitored node on the server before the subscription times out there; its consumer receivesBad_NodeIdUnknown, the other node keeps delivering and theOpcUaClientis the same instance.The first two fail without the fix, the slow-route test fails with the second revision of this PR and the last one with the first revision; all pass now. The class was run with
-Dsurefire.rerunFailingTestsCount=0five times in a row before the rebase onto main and on each commit after it../mvnw -pl components/camel-milo -Psourcecheck verify: 78 tests, 1 skipped and 1 failure,ServerCertificateValidationTest(CAMEL-25645), which fails the same way on main.The current revision was also tested against a real server, with each client limited to one CPU: a subscription that the server loses while the session stays (upstream main stays silent, this revision recreates it), a slow route, a route slower than the subscription lifetime, a slow route while the connection is lost, silent drops shorter and longer than the subscription lifetime, a recreate in flight when the server becomes unreachable, with and without a producer on the connection, and a server restart. The executor change has no unit test, because it needs a scheduled executor with one thread and a server that cannot be reached. In these runs the previous revision hung in the rebuild, and with the recovery on the scheduled executor a recreate in flight kept the keep-alive from running for 60 s. The results are in the review thread.
Docs: a "Subscriptions that stop publishing" section in
milo-client-component.adoc(catalog copy identical), including what happens with a slow route, while the connection is lost and when the subscription cannot be transferred, and a note in the 4.24 upgrade guide under the existing camel-milo section.@davsclaus @oscerd
Target
camel-3.x, whereas Camel 4 uses themainbranch)Tracking
Apache Camel coding standards and style
mvn clean install -DskipTestslocally from root directory and I have committed all auto-generated changes.Claude Code on behalf of etzel-mes