Repository navigation
CAMEL-25835: camel-milo - the consumer reports a lost connection and refreshes its values after it - #28128
Conversation
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
The concurrency design is solid. Verified five critical questions via Milo 1.1.8 bytecode analysis and code tracing — all non-issues. Two low-severity observations below.
Pre-fix test validation: Static trace of CommunicationLossTest.lostSessionReportsTheNodeAndResendsItsValue confirms it exercises the new code path: relay.dropConnections() triggers onSessionInactive → reportLoss → Subscription.report() → consumer delivery. The assertNoCommunication() assertion would fail without the fix (no Bad_NoCommunication was delivered before). ✅
Concurrency validation (all non-issues):
reportLossbeforedispose()inhandleConnectionFailure:TaskQueue.execute()enqueues onto the delivery executor which is not shut down byclient.disconnect(). The loss task executes normally.closedflag race withonSessionInactive: Milo dispatchesSessionActivityListenercallbacks asynchronously viaexecutor.execute().dispose()writesclosed = true(volatile) before callingdisconnect(). The async callback sees the write and skips the spurious report.sessionActive = trueinitialization: safe —performConnect()blocks until success; if it fails, the listener is never associated with aConnectedinstance.refusedvolatile field: volatile write inputSubscriptionsis visible to the subsequent synchronized read inreport(). No race.generationCounter.get()(notincrementAndGet()) inregisterItems: correct — generation 0 is always less than any connection's generation (≥1), soSubscription.report()drops the stale report once the first real value arrives.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
allthingssecurity
left a comment
allthingssecurity
left a comment
There was a problem hiding this comment.
Thanks @etzel-mes, this is a careful piece of work. I found one ordering gap in the report of a node that is
registered while there is no connection, on 7c92ff084e20.
A node registered while a reconnect is in flight can end up showing Bad_NoCommunication while the connection is
up (SubscriptionManager.registerItems, lines 1499-1501, with Subscription.report, line 311)
When connected == null, registerItems tags the report with generationCounter.get() and runs it on
this.executor. But performConnect increments the counter when it creates the listener (line 1287), before
client.connect() and well before performAndEvalConnect installs connected. So a consumer that starts during a
reconnect attempt gets the generation of the connection that is being built. Its report and that connection's first
value then carry the same generation, and report only drops a report when this.generation > generation. The
report runs on the shared MiloClient pool and the value on the subscription's delivery queue, so nothing orders
them; if the pool is busy (it has availableProcessors() threads, and with this PR it also runs the routes of other
nodes' loss reports), the report lands after the good value. The node then stays Bad_NoCommunication until its value
next changes, which is the situation this PR sets out to avoid.
Reproducer (a test in org.apache.camel.component.milo, on the embedded server of AbstractMiloServerTest; it uses
reflection to hit the window, and an executor that holds its tasks to stand in for a busy pool):
SubscriptionManager manager = new SubscriptionManager(endpoint.getConfiguration(), reconnect, held::add, 10_000);
// connection 1 is up; it fails and is rebuilt right away
handleConnectionFailure.invoke(manager, new Exception("connection failed"));
// wait until generationCounter == 2 while connected == null, take `lock`, then:
manager.registerItems(Map.of(node, values::add), 50.0, null, null);
// connection 2 delivers the current value; then the pool runs the report
await values.size() == 1; held.forEach(Runnable::run);
assertTrue(values.get(values.size() - 1).getStatusCode().isGood());Observed:
values delivered: [DataValue[value=Variant{value=Foo}, statusCode=Good ...],
DataValue[value=Variant{value=null}, statusCode=Bad_NoCommunication ...]]
AssertionFailedError: the connection is up and the value unchanged, but the last delivery is
DataValue[value=Variant{value=null}, statusCode=StatusCode[name=Bad_NoCommunication, value=0x80310000 ...
Suggested fix: a node registered without a connection has had no value from any connection, so its report can use a
generation below every connection's, for example 0 (connections start at 1, and Subscription.generation starts
at 0):
if (this.connected == null) {
final DateTime time = DateTime.now();
// lower than the generation of any connection: dropped once any connection has delivered a value
this.executor.execute(() -> deliverLoss(subscriptions, 0L, time));
}With that change the test above passes, and CommunicationLossTest, MonitorItemLifecycleTest,
MonitorItemMultiConnectionsTest and ServerCertificateValidationTest still pass (14 tests).
Claude Code on behalf of allthingssecurity
|
Thanks @allthingssecurity, you are right: the counter is taken when the listener is created, before the connection is built, so a node registered during that attempt shared the generation of the connection's first value, and nothing ordered the report on the pool after it. Fixed in 66d3d31 as you suggest: the report of a node registered without a connection carries generation 0, below every connection's, so any delivered value drops it. Module verify with Claude Code on behalf of etzel-mes |
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
✅ Re-review after latest commits (66d3d31) — all findings verified, no issues found.
The generation-counter approach correctly solves the race condition flagged by allthingssecurity. Key points confirmed:
- Generation-0 suppression:
deliverLoss()uses gen=0;generationCounter.incrementAndGet()ensures real connections carry gen≥1; thereport()guardthis.generation > generationcleanly drops stale gen-0 loss reports once a real connection delivers HashMap→ConcurrentHashMap: correctly supports lock-free reads indeliverLoss()while writes remain under lockrefuse()/report()synchronization: volatile+synchronized guards ensure refused nodes cannot have their status overridden by a latereport()callrefused=falsereset on reconnect: volatile write under lock, followed bydeliver()advancing the generation — structurally sound- Test coverage:
consumerStartedWithoutTheServerReportsTheNodeThenReceivesItsValueexercises exactly the fix: gen-0 loss on startup, thenassertGoodValue("Foo")on reconnect - Documentation: Bad_NoCommunication, null value, DateTime.now() timestamps all match
deliverLoss()implementation
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:
|
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Thanks for the very thorough work and the opc-plc verification. I checked the Milo calls against 1.1.8: OpcUaSubscription.getDeliveryQueue(), callAsync, and NodeIds.Server_ResendData all exist, and SessionActivityListener callbacks go through a sequential executor, so the transitions arrive in order. Nothing that holds lock calls a synchronized method of Subscription, so I see no deadlock. There is no textual conflict with #27956.
On by default with no option is fine with me, given the upgrade-guide entry. Two small things, see inline.
Open questions, not blocking:
- When the client cannot connect at all (wrong credentials, untrusted certificate), would a more specific status (
Bad_UserAccessDenied,Bad_SecurityChecksFailed) be more useful thanBad_NoCommunication? Keeping the cause in the log only is defensible. CommunicationLossTestcan wait up to 2x30 s plus 60 s, because the embedded server republishes late. Acceptable, but if any of it can be shortened, the module build would thank you.
Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
|
🧪 CI tested the following changed modules:
🔬 Scalpel shadow comparison — Scalpel: 8 of 707 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 39s total)Total reactor time: 5m 39s
Top 20 slowest modules:
|
66d3d31 to
28ae070
Compare
28ae070 to
2b688a7
Compare
|
Thanks for the review. On the status: OPC 10000-4 §7.38 lists On the test: the 30 s and 60 s values are caps and are not spent when the value arrives; the class takes about 58 s. Each drop costs about 18 s that the test cannot remove: the embedded server answers ResendData on the Publish request of the dropped channel, so the value arrives with the next keep-alive (about 10 s, Milo derives the keep-alive count from a 10 s target), and the component's reconnect back-off is 10 s. The remaining fixed assert periods are small; I have left them as they are. Claude Code on behalf of etzel-mes |
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
LGTM, thanks. Both points are addressed: the upgrade guide says how to keep the previous behaviour, and the constructor now comes after the fields. The rest of the delta is the rebase onto main.
One optional wording nit is inline.
Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
…refreshes its values after it The consumer delivered nothing when communication with the server was lost, and after the same session was activated again the values that did not change were not delivered again. A route could not tell "unchanged" from "unknown". Every monitored node now delivers one exchange with a null value and Bad_NoCommunication, stamped with the time the loss was recognised: when the session stops being active, when the connection is rebuilt, and when nodes are registered without a connection or with an inactive session. A node is reported once per loss, and a node whose item the server refused keeps that status. When the same session is activated again, the consumer calls the server's ResendData method, so the current value of every node arrives, the unchanged ones included; any answer other than a lost session recreates the subscription, as CAMEL-25691 does for one that stopped publishing. Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
…t is not left reported after its first value The report of a node registered while there was no connection carried the current generation counter, which the connection being built had already taken. Its first value and the report then had the same generation, and when the report ran after the value on the shared pool, the node kept Bad_NoCommunication while the connection was up. The report now carries generation 0, below every connection's, so any delivered value drops it. Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
2b688a7 to
1237f27
Compare
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
LGTM, thanks. After the rebase, the only change is the upgrade guide: it now says ${body.statusCode.bad} == false, which matches the old behaviour.
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
CAMEL-25835.
The
milo-clientconsumer hands every data change notification to its route and reports nothing when the values stop. When the session drops and Milo activates it again, the consumer only flips itssessionActiveflag, and the values that did not change during the outage are not delivered again. When the subscription is transferred or the connection is rebuilt, the route sees a pause and then first values. When the server cannot be reached at startup, or the client cannot connect at all, the route sees nothing. A route cannot tell "unchanged" from "unknown": a historian keeps the last value, a publisher republishes it as live, an alarm stays quiet.The change, following the OPC UA specifications (OPC 10000-4 §7.38 Table 179 and 10000-8 §7.3 Table 61 for
Bad_NoCommunication, 10000-4 §7.11.3 for the timestamps, 10000-14 §6.2.11.1 Table 80 for a consumer without a last value, 10000-4 §6.7 step 6 and 10000-5 §9.2 forResendData):DataValuehas a null value and the statusBad_NoCommunication, both timestamps set to the time the loss was recognised. The triggers are the session becoming inactive (SessionActivityListener), a rebuild of the connection (handleConnectionFailure, before it disposes the connection), and a registration while there is no connection or the session is inactive. The report of a connection is queued on its subscription's delivery queue, so it lands after the values the queue still holds and before the values of the activated session; a registration without a connection is reported on the connection's executor, not on the thread that starts the route.ResendDatafor its subscription, so the server sends the current value of every node, the unchanged ones included. Any answer other than Good or a lost session recreates the subscription with the recovery of CAMEL-25691 (delete, then create again in the session; the new items deliver their first values), logged at WARN. After a transfer or a rebuild the server sends the first values anyway.ResendData: Milo, open62541, node-opcua, the OPC Foundation .NET stack), no handling of lost notifications, no per-node ordering rules. Event items are unchanged.CommunicationLossTestwith a TCP relay between the client and the embedded server: two losses in a row on the same client and session (asserted), the unchanged value resent, a change in between delivered, a refused node on the same connection keepsBad_NodeIdUnknown; and a consumer started with the relay down reports first and receives its value once the relay is up.MonitorItemLifecycleTest,MonitorItemMultiConnectionsTestandServerCertificateValidationTestnow expect the report for a rebuilt connection and for clients that cannot connect at all. The embedded Milo server answersResendDataon the oldest queued Publish request, which belongs to the dropped channel, so the resent value reaches the client 10-20 s later through its republish of the missed notification; the relay test waits for it and says why (a real server answered in 2 ms).0orfalsethrough the type converters, a route checks${body.statusCode.bad}; repeats after an outage; clients that cannot connect now report; the consumer's route runs during an outage.Against a server that is not built on Milo (Microsoft opc-plc 2.15.4): a 20 s silent drop with and without writes, a 150 s drop, a server restart, a value held in a slow route with another queued behind it and then a drop, a consumer started while the server was stopped, and a refused node next to a good one. One
Bad_NoCommunicationper node per outage in every run;ResendDataanswered Good after the same session was activated again (the unchanged value arrived 0.1 s later), a silent drop was recognised after 10-14 s with Milo's keep-alive defaults, and the long drop and the restart ended in a transfer failure and a rebuild with first values. The report that was queued behind a value held in the route landed after that value.CAMEL-25692 (retry of refused monitored items) changes the registration code too; I will rebase whichever lands second. A producer request that fails because the session was lost still rebuilds the shared connection, so its consumers see the report and first values instead of a reactivation; a rebuild only for other failures belongs with CAMEL-25694 (#27956), which rewrites those blocks.
Checked with
-Psourcecheck verifyon the module (88 tests, no reruns).@davsclaus @oscerd could you take a look?
Target
mainbranch)Tracking
Apache Camel coding standards and style
I checked that each commit in the pull request has a meaningful subject line and body.
I have run
mvn clean install -DskipTestslocally from root folder and I have committed all auto-generated changes.AI-assisted contributions
Co-authored-bytrailers) and the PR description identifies the AI tool used.Claude Code on behalf of etzel-mes