davsclaus commented on code in PR #28212:
URL: https://github.com/apache/camel/pull/28212#discussion_r4333716365
##########
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:
**A.** The retry runs `putSubscriptions` while holding `lock`, so the
namespace read and the CreateMonitoredItems request block every user of the
lock: producers, `getOpcUaClient()`, `unregisterItems` (route stop) and
`dispose()`. That happens every 10 seconds for as long as a node is refused.
The register and recreate paths do the same, but only once; here it recurs for
the life of the connection, which is what the note on `recoverSubscription`
tries to avoid.
With a silent network drop, or a server that answers keep-alives but is slow
on CreateMonitoredItems, an attempt that starts before `sessionActive` turns
false can hold the lock for up to the request timeout (60 s by default). The
`CommunicationLossTest` case closes the sockets, so it only covers the skipped
attempt.
Could you either keep the blocking part out of the lock (collect the
subscriptions under the lock, make the calls, then re-check `this.connected ==
connected` before applying the results), or give the attempt a shorter timeout?
If you think it is acceptable as is, a sentence in the javadoc of
`retryRefused` saying why it is bounded in practice would help the next reader.
##########
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:
**C.** This is effectively a 12 s sleep, which the test guidelines ask us to
avoid. Waiting with `during` would also check what the comments promise, that
the refused node is not reported while the server is unreachable (please adjust
the endpoint name to the one of the refused node):
```java
await().during(12, TimeUnit.SECONDS).atMost(13, TimeUnit.SECONDS).until(()
-> unknownEndpoint.getReceivedCounter() == 0);
```
##########
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:
**B.** `retryJob` is documented as guarded by `lock`, but
`SubscriptionManager.dispose()` calls `connected.dispose()` after releasing the
lock, so this cancel races with `retryRefused` and `scheduleRetry`. It is
harmless today, because `disposed` is set under the lock first and
`retryRefused` checks it. Could you cancel the job inside the locked section of
`SubscriptionManager.dispose()`, or adjust the field comment?
##########
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:
**D.** Nit: after this WARN the later attempts log only at DEBUG, so an
operator sees one warning and then nothing. Could the message say the item is
retried, for example `Monitored item for node {} could not be created{}, it is
created again every {} ms: {}`? Then the WARN alone explains what Camel is
doing.
--
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]
