gnodet-bot commented on code in PR #28212:
URL: https://github.com/apache/camel/pull/28212#discussion_r4333677563
##########
components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java:
##########
@@ -344,6 +346,17 @@ private class Connected {
private final Map<String, UShort> namespaceCache = new
ConcurrentHashMap<>();
+ /**
+ * The client handles of the nodes that the server refused to monitor
on this connection, which are created
+ * again every {@code reconnectTimeout} milliseconds, guarded by
{@code lock}
+ */
+ private final Set<UInteger> refused = new LinkedHashSet<>();
+
+ /**
+ * The next attempt to create the refused nodes again, if one is
scheduled, guarded by {@code lock}
+ */
+ private Future<?> retryJob;
Review Comment:
💡 `retryJob` is written under `lock` (in `scheduleRetry` and `retryRefused`)
but read and written in `Connected.dispose()` without the lock when called from
`SubscriptionManager.dispose()`. The JMM allows a stale `null` to be read,
skipping `Future.cancel()`. The scheduled task still fires, acquires the lock,
sees `disposed=true`, and exits immediately — so there is no correctness bug —
but the cancel is not guaranteed to fire. Making the field `volatile` closes
the gap and matches `sessionActive` and `closed` in `SubscriptionListenerImpl`.
```suggestion
private volatile Future<?> retryJob;
```
##########
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);
Review Comment:
When a consumer is unregistered _between_ two retry cycles (after
`retryRefused` has already snapshotted the refused set but before it runs
again), its `clientHandle` remains in `connected.refused`. The null check at
line 1125 correctly skips it this cycle, but it stays in `refused` until a
subsequent `deactivate()` removes it — causing one extra no-op pass per retry
cycle for each orphaned handle. Removing null-subscription handles from
`connected.refused` in this loop would keep the set tidy.
```suggestion
final Iterator<UInteger> it = connected.refused.iterator();
while (it.hasNext()) {
final UInteger clientHandle = it.next();
final Subscription subscription =
this.subscriptions.get(clientHandle);
if (subscription != null) {
refused.put(clientHandle, subscription);
} else {
it.remove();
}
}
```
--
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]
