etzel-mes commented on code in PR #28212:
URL: https://github.com/apache/camel/pull/28212#discussion_r4334245382
##########
components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java:
##########
@@ -1040,6 +1103,43 @@ private void recreateSubscription(final
OpcUaSubscription subscription) {
pendingValues.forEach(PendingValue::deliver);
}
+ /**
+ * Creates the nodes that the server refused to monitor on a connection
again, in one CreateMonitoredItems call,
+ * while the connection is the current one. Nothing is delivered to the
consumers: a node created now delivers its
+ * first value from the server. While the session is not active, the
attempt is skipped. A call that fails is made
+ * again with the next attempt and never rebuilds the connection, which
would interrupt the nodes that are
+ * monitored: a lost session or subscription is recovered without it.
+ */
+ private void retryRefused(final Connected connected) {
+ lock.lock();
+ try {
+ connected.retryJob = null;
+ if (this.disposed || this.connected != connected ||
connected.refused.isEmpty()) {
+ return;
+ }
+ if (!connected.listener.sessionActive) {
+ LOG.debug("Session not active, the refused monitored items are
created again later");
+ connected.scheduleRetry();
+ return;
+ }
+ final Map<UInteger, Subscription> refused = new LinkedHashMap<>();
+ for (final UInteger clientHandle : connected.refused) {
+ final Subscription subscription =
this.subscriptions.get(clientHandle);
+ if (subscription != null) {
+ refused.put(clientHandle, subscription);
+ }
+ }
+ try {
+ connected.putSubscriptions(refused, true);
Review Comment:
Thanks, I looked at this again with the numbers. You are right that it is
not rare. An attempt runs every 10 s, and a silent drop is only noticed after
10 to 15 s (Milo's keep-alive defaults: a read every 5 s, 5 s timeout, one
failure allowed). So with a refused node, almost every silent drop catches an
attempt in flight. Until the drop is noticed, that attempt holds `lock`: a
producer blocks its thread on it instead of waiting asynchronously, and a route
stop, `getOpcUaClient()` and `dispose()` wait as well.
It does not last for the request timeout, though. When the keep-alive gives
up, Milo closes the channel and the pending request fails with
`Bad_ConnectionClosed`. Against a real server (node-opcua in Docker, iptables
DROP), an attempt sent 2 s into the drop failed this way about 10 s later, and
a call to `getOpcUaClient()` made meanwhile waited 9.8 s for `lock`. The next
attempt was scheduled from there without a rebuild. A 45 s drop gave the same
single wait, 9.9 s, because the following attempts were skipped while the
session was inactive. A server that answers the keep-alives but not the
CreateMonitoredItems is bounded by `requestTimeout`, as the registration,
recreate and route-stop calls under `lock` are.
I would still keep it under `lock`, and the javadoc of `retryRefused` now
says this (c92824359d3):
- Outside the lock: `OpcUaSubscription` creates monitored items only with a
blocking call, and `putSubscriptions` waits for the namespace read before it.
Making them outside `lock` means splitting `putSubscriptions` into a prepare, a
call and an apply step for the retry. It also means reconciling the results
with what changed during the call, for example a consumer stopped meanwhile,
whose item may now exist on the server and has to be deleted again. That is
more code and new interleavings, and the wait it removes is bounded as above.
- A shorter timeout: `OpcUaSubscription` sends the request with the request
timeout of the client. A raw service call with its own timeout would not stop
the server either, since the `timeoutHint` of the request header is only a hint
to the server (OPC UA Part 4, 7.32). The server may still create the items
after the client gave up, and the next attempt would create them a second time.
That can happen at the 60 s timeout too, but a short timeout makes it likely,
and it would repeat every 10 s.
The wait comes from the producers taking `lock` at all. Under it they only
read the current connection and start an asynchronous request, so they wait
behind whatever service call holds it: the registration and the route stop as
much as this retry. Reading the connection without `lock` would remove that
wait for all of them, as long as a request sent on a connection that is
replaced meanwhile fails without taking down the new one. That is a change to
the producers rather than to the retry, so I would do it in a separate JIRA,
and can open one if you agree.
_Claude Code on behalf of etzel-mes_
##########
components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java:
##########
@@ -538,6 +597,10 @@ private CompletableFuture<UShort>
lookupNamespaceIndex(final String namespaceUri
public void dispose() {
// the session closed by the disconnect is no loss of communication
this.listener.closed = true;
+ if (this.retryJob != null) {
+ this.retryJob.cancel(false);
+ this.retryJob = null;
+ }
Review Comment:
Done in c92824359d3: `Connected.cancelRetry()` is called under `lock`
wherever a connection is replaced or disposed, including the locked section of
`SubscriptionManager.dispose()`, and `Connected.dispose()` no longer touches
`retryJob`, so the field is only used under `lock`, as documented.
_Claude Code on behalf of etzel-mes_
##########
components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java:
##########
@@ -195,6 +198,62 @@ void lostSessionReportsTheNodeAndResendsItsValue() throws
Exception {
}
}
+ @Test
+ void refusedNodeIsCreatedOnceTheSessionIsActiveAgain() throws Exception {
+ producer1.sendBody("Foo");
+
+ item1Endpoint.expectedMessageCount(1);
+ testBody(item1Endpoint.message(0), assertGoodValue("Foo"));
+ context.getRouteController().startRoute("item1");
+ item1Endpoint.assertIsSatisfied();
+
+ unknownEndpoint.expectedMessageCount(1);
+ testBody(unknownEndpoint.message(0),
+ value -> assertEquals(StatusCodes.Bad_NodeIdUnknown,
value.getStatusCode().getValue()));
+ context.getRouteController().startRoute("unknown");
+ unknownEndpoint.assertIsSatisfied();
+
+ final MiloClientEndpoint endpoint = context.getEndpoint(
+ resolve(MILO_CLIENT_ITEM_1).replace("@@relay@@",
Integer.toString(relayPort.getPort())),
+ MiloClientEndpoint.class);
+ final MiloClientConnection connection = endpoint.createConnection();
+ try {
+ final OpcUaClient client = connection.getOpcUaClient();
+ final NodeId sessionId = client.getSession().getSessionId();
+
+ // the refused node is not reported, and delivers its value once
it is created after the session is back
+ unknownEndpoint.reset();
+ unknownEndpoint.expectedMessageCount(1);
+ unknownEndpoint.setResultWaitTime(60_000);
+ unknownEndpoint.setAssertPeriod(2_000);
+ testBody(unknownEndpoint.message(0), assertGoodValue("Baz"));
+ item1Endpoint.reset();
+ item1Endpoint.expectedMessageCount(2);
+ item1Endpoint.setResultWaitTime(60_000);
+ item1Endpoint.setAssertPeriod(2_000);
+ testBody(item1Endpoint.message(0), assertNoCommunication());
+ testBody(item1Endpoint.message(1), assertGoodValue("Foo"));
+
+ // the server stays unreachable for longer than the retry interval
of the refused node
+ relay.stop();
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
item1Endpoint.getReceivedCounter() >= 1);
+ await().pollDelay(12, TimeUnit.SECONDS).atMost(13,
TimeUnit.SECONDS).until(() -> true);
Review Comment:
Done in c92824359d3 with your `during` wait, with `atMost` 15 s so that a
slow CI machine has some slack. The 12 s now also check that the refused node
is not reported while the server is unreachable.
_Claude Code on behalf of etzel-mes_
##########
components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java:
##########
@@ -451,9 +476,42 @@ public List<PendingValue> putSubscriptions(final
Map<UInteger, Subscription> sub
}
}
+ scheduleRetry();
return pendingValues;
}
+ /**
+ * Records that the server refused to monitor a node. The first
refusal is logged at WARN and delivered to the
+ * consumer, a refusal on a later attempt only logged at DEBUG.
+ */
+ private void refuse(
+ final UInteger clientHandle, final Subscription s, final
StatusCode status, final String reason,
+ final boolean retry, final List<PendingValue> pendingValues) {
+ this.refused.add(clientHandle);
+ final String node = s.getNodeId().toParseableString();
+ final String because = reason == null ? "" : ", " + reason;
+ if (retry) {
+ LOG.debug("Monitored item for node {} could not be created on
retry{}: {}", node, because, status);
+ } else {
+ LOG.warn("Monitored item for node {} could not be created{}:
{}", node, because, status);
Review Comment:
Done in c92824359d3: the warning now reads `Monitored item for node {} could
not be created{}, trying again every {} ms: {}`, for example `... could not be
created, trying again every 10000 ms: StatusCode[name=Bad_NodeIdUnknown, ...]`.
Later attempts stay at DEBUG, and a node created on retry is logged at INFO.
The interval is the fixed reconnect interval of 10 s; CAMEL-25873 is about
making the `reconnectTimeout` option set 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
