Skip to content

CAMEL-25912: camel-milo - client producers no longer wait for the manager lock or rebuild the connection #28371

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
davsclaus merged 3 commits into from
Oct 24, 2026
Merged

CAMEL-25912: camel-milo - client producers no longer wait for the manager lock or rebuild the connection #28371

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
fbd934e
CAMEL-25912: camel-milo - client producers no longer wait for the man…
etzel-mes Oct 23, 2026
1b5493a
CAMEL-25912: camel-milo - move the javadoc of assertStatus to the method
etzel-mes Oct 23, 2026
9871e1b
CAMEL-25912: camel-milo - the upgrade guide names the endpoint and ce…
etzel-mes Oct 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion ...rc/generated/resources/org/apache/camel/catalog/docs/milo-client-component.adoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,7 +239,11 @@ or detects changes must tolerate this. A value that did not change during the lo

Routes run while communication is lost. An OPC UA request that a route makes while the client reconnects, for example
a write, waits for the next reconnect attempt and is sent if that attempt succeeds, so it can take effect after
communication is restored. A request that fails rebuilds the connection.
communication is restored. A request that fails completes with its failure; the connection is kept, and a lost
session is recovered by the client as described above. The client recovers it with the endpoint and the server
certificate it found when it connected, so a server that changes either while it cannot be reached is reached again
only when Camel rebuilds the connection, for example when the Camel context restarts. A request does not wait while a
consumer of the same connection creates or deletes its monitored items.

Camel keeps no last value. A route that needs one keeps the last good value of each node itself. To recognise the
status, check the status code of the `DataValue` body, for example with `${body.statusCode.bad}`.
Expand Down
6 changes: 5 additions & 1 deletion components/camel-milo/src/main/docs/milo-client-component.adoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,7 +239,11 @@ or detects changes must tolerate this. A value that did not change during the lo

Routes run while communication is lost. An OPC UA request that a route makes while the client reconnects, for example
a write, waits for the next reconnect attempt and is sent if that attempt succeeds, so it can take effect after
communication is restored. A request that fails rebuilds the connection.
communication is restored. A request that fails completes with its failure; the connection is kept, and a lost
session is recovered by the client as described above. The client recovers it with the endpoint and the server
certificate it found when it connected, so a server that changes either while it cannot be reached is reached again
only when Camel rebuilds the connection, for example when the Camel context restarts. A request does not wait while a
consumer of the same connection creates or deletes its monitored items.

Camel keeps no last value. A route that needs one keeps the last good value of each node itself. To recognise the
status, check the status code of the `DataValue` body, for example with `${body.statusCode.bad}`.
Expand Down
15 changes: 7 additions & 8 deletions ...camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConnection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,15 +138,14 @@ MonitoringState getMonitoringState() {
protected void init() {
ObjectHelper.notNull(this.camelContext, "camelContext");

// A failed request disconnects the client, which can wait for a reconnect attempt that Milo schedules on its shared
// scheduled executor, so this connection handles its requests and reconnects on executors of its own.
// The outcome of a request continues its route, and a reconnect waits for the server; neither may hold a thread of
// Milo, whose shared scheduled executor runs the reconnect attempts of the clients. So this connection handles its
// requests and reconnects on executors of its own.
final ExecutorServiceManager executorServiceManager = this.camelContext.getExecutorServiceManager();
final ThreadPoolProfile profile = new ThreadPoolProfile("MiloClient");
// as many threads as that executor, but they end when they are idle; with virtual threads and more than one
// processor, a virtual thread for each task. Not more platform threads, because when the connection is lost,
// every request in flight fails and waits for the disconnect of the first failure. With one processor this is
// one thread, so a route that continues on it after an awaited request must not wait for another request of
// this connection.
// processor, a virtual thread for each task. With one processor this is one thread, so a route that continues on
// it after an awaited request must not wait for another request of this connection.
final int threads = Runtime.getRuntime().availableProcessors();
profile.setPoolSize(threads);
profile.setMaxPoolSize(threads);
Expand All @@ -156,8 +155,8 @@ protected void init() {
// a request that completes after the connection is closed fails instead of being dropped
profile.setRejectedPolicy(ThreadPoolRejectedPolicy.Abort);
this.executor = executorServiceManager.newThreadPool(this, "MiloClient", profile);
// the reconnects of the connection run one after the other, on a thread that starts with the first one and ends
// when the connection is closed
// the reconnects of the connection and its attempts to create the refused monitored items again run one after
// the other, on a thread that starts with the first one and ends when the connection is closed
this.reconnectExecutor = executorServiceManager.newSingleThreadScheduledExecutor(this, "MiloClientReconnect");

this.manager = new SubscriptionManager(
Expand Down
173 changes: 79 additions & 94 deletions ...lo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -343,7 +343,12 @@ void deliver() {
}

private class Connected {
private OpcUaClient client;
/**
* The client, which is kept after the connection is disposed: a producer reads the connection without
* {@code lock}, so it can still send a request on a connection that is disposed, which the client then fails
* with the status of its closed session
*/
private final OpcUaClient client;
private final OpcUaSubscription manager;
private final SubscriptionListenerImpl listener;
private final long generation;
Expand Down Expand Up @@ -517,16 +522,18 @@ private void refuse(

/**
* Schedules the next attempt to create the refused nodes again, unless there is none or one is scheduled. The
* timer runs on the reconnect thread, the attempt, which blocks until the server answers, on the executor of
* the connection, so that a slow attempt does not hold up a reconnect.
* attempt blocks until the server answers, so it runs on the reconnect thread, not on the executor of the
* connection, whose threads complete the requests of the producers: with one processor that executor has a
* single thread, and the outcome of every request would wait for the attempt. A rebuild of the connection does
* not wait for the reconnect thread longer than it waits for the attempt anyway: it needs {@code lock}, which
* the attempt holds until the server answers.
*/
private void scheduleRetry() {
if (this.refused.isEmpty() || this.retryJob != null || SubscriptionManager.this.disposed) {
return;
}
this.retryJob = SubscriptionManager.this.reconnectExecutor.schedule(
() -> SubscriptionManager.this.executor.execute(() -> retryRefused(this)),
SubscriptionManager.this.reconnectTimeout, TimeUnit.MILLISECONDS);
() -> retryRefused(this), SubscriptionManager.this.reconnectTimeout, TimeUnit.MILLISECONDS);
}

/**
Expand Down Expand Up @@ -623,16 +630,15 @@ private CompletableFuture<UShort> lookupNamespaceIndex(final String namespaceUri
}

public void dispose() {
if (this.listener.closed) {
return;
}
// the session closed by the disconnect is no loss of communication
this.listener.closed = true;
if (this.client != null) {
try {
this.client.disconnect();
this.client = null;
} catch (final UaException e) {
LOG.warn("Failed to disconnect client", e);
}

try {
this.client.disconnect();
} catch (final UaException e) {
LOG.warn("Failed to disconnect client", e);
}
}

Expand Down Expand Up @@ -972,7 +978,8 @@ nodeId, direction, NodeIds.References, includeSubTypes, uint(nodeClasses),
private final Lock lock = new ReentrantLock();

/**
* The current connection, written under {@code lock} and read by the monitoring state without it
* The current connection, written under {@code lock} and read by the producers, {@link #getOpcUaClient()} and the
* monitoring state without it
*/
private volatile Connected connected;
private volatile Throwable lastConnectFailure;
Expand Down Expand Up @@ -1003,12 +1010,9 @@ public SubscriptionManager(final MiloClientConfiguration configuration, final Sc
* @return the active client, or {@code null} if there is no current connection
*/
public OpcUaClient getOpcUaClient() {
lock.lock();
try {
return this.connected != null ? this.connected.client : null;
} finally {
lock.unlock();
}
// read without lock, which a service call holds until the server answers
final Connected connected = this.connected;
return connected != null ? connected.client : null;
}

/**
Expand Down Expand Up @@ -1079,8 +1083,8 @@ private void checkSubscription(final OpcUaSubscription subscription, final Subsc
* subscriptions.
* <p>
* The delete neither blocks nor holds {@code lock}: a request sent before the connection was lost only fails once
* the client notices the loss, and holding {@code lock} until then would hold up the producers and
* {@link #getOpcUaClient()} as well.
* the client notices the loss, and holding {@code lock} until then would hold up the consumers that start or stop
* as well.
*/
private void recoverSubscription(final OpcUaSubscription subscription, final SubscriptionListenerImpl listener) {
subscription.deleteAsync().whenCompleteAsync((unit, e) -> {
Expand Down Expand Up @@ -1156,10 +1160,11 @@ private void recreateSubscription(final OpcUaSubscription subscription) {
* failure is the status of the nodes until the next attempt.
* <p>
* Like the registration of a consumer, the attempt holds {@code lock} until the server answers, so that a consumer
* that stops or a subscription that is recreated in the meantime cannot interleave with it, and the producers and
* {@link #getOpcUaClient()} wait for it. The wait is bounded: the attempt is only sent while the session is active,
* a call in flight when the connection drops silently fails once the keep-alive of the client declares the session
* lost, and a call that the server does not answer fails after the request timeout.
* that stops or a subscription that is recreated in the meantime cannot interleave with it, and a consumer that
* starts or stops waits for it; the producers and {@link #getOpcUaClient()} do not. The wait is bounded: the
* attempt is only sent while the session is active, a call in flight when the connection drops silently fails once
* the keep-alive of the client declares the session lost, and a call that the server does not answer fails after
* the request timeout.
*/
private void retryRefused(final Connected connected) {
lock.lock();
Expand Down Expand Up @@ -1270,14 +1275,17 @@ private void deliverLoss(final Map<UInteger, Subscription> nodes, final long gen
private void handleConnectionFailure(final Throwable e) {
lock.lock();
try {
if (this.connected != null) {
// the nodes lose their communication with the connection, after the values it still delivers
reportLoss(this.connected.manager, this.connected.generation, this.subscriptions);
this.connected.cancelRetry();
this.connected.dispose();
// set first: a producer that finds no connection while it is torn down fails with this cause
this.lastConnectFailure = e;
final Connected failed = this.connected;
if (failed != null) {
// the producers and getOpcUaClient() no longer reach the connection that is torn down
this.connected = null;
// the nodes lose their communication with the connection, after the values it still delivers
reportLoss(failed.manager, failed.generation, this.subscriptions);
failed.cancelRetry();
failed.dispose();
}
this.lastConnectFailure = e;

// log

Expand Down Expand Up @@ -1498,6 +1506,8 @@ public void dispose() {
}
this.disposed = true;
connected = this.connected;
// the producers and getOpcUaClient() no longer reach the connection that is disposed
this.connected = null;
if (connected != null) {
// the next attempt is cancelled under the lock that guards it, the connection is disposed outside of it
connected.cancelRetry();
Expand Down Expand Up @@ -1702,82 +1712,57 @@ public void unregisterItems(final List<UInteger> clientHandles) {

public CompletableFuture<CallMethodResult> call(
final ExpandedNodeId nodeId, final ExpandedNodeId methodId, final Variant[] inputArguments) {
lock.lock();
try {
if (this.connected == null) {
return newNotConnectedResult();
}

return this.connected.call(nodeId, methodId, inputArguments).whenCompleteAsync((status, e) -> {
// handle outside the lock, running using
// whenCompleteAsync (preserves the original exceptional completion)
if (e != null) {
handleConnectionFailure(e);
}
}, this.executor);
} finally {
lock.unlock();
final Connected connected = this.connected;
if (connected == null) {
return newNotConnectedResult();
}

return outcomeOf(connected.call(nodeId, methodId, inputArguments));
}

public CompletableFuture<?> write(final ExpandedNodeId nodeId, final DataValue value) {
lock.lock();
try {
if (this.connected == null) {
return newNotConnectedResult();
}

return this.connected.write(nodeId, value).whenCompleteAsync((status, e) -> {
// handle outside the lock, running using
// whenCompleteAsync (preserves the original exceptional completion)
if (e != null) {
handleConnectionFailure(e);
}
}, this.executor);
} finally {
lock.unlock();
final Connected connected = this.connected;
if (connected == null) {
return newNotConnectedResult();
}

return outcomeOf(connected.write(nodeId, value));
}

public CompletableFuture<?> readValues(final List<ExpandedNodeId> nodeIds) {
lock.lock();
try {
if (this.connected == null) {
return newNotConnectedResult();
}

return this.connected.readValues(nodeIds).whenCompleteAsync((nodes, e) -> {
// handle outside the lock, running using
// whenCompleteAsync (preserves the original exceptional completion)
if (e != null) {
handleConnectionFailure(e);
}
}, this.executor);
} finally {
lock.unlock();
final Connected connected = this.connected;
if (connected == null) {
return newNotConnectedResult();
}

return outcomeOf(connected.readValues(nodeIds));
}

public CompletableFuture<Map<ExpandedNodeId, BrowseResult>> browse(
List<ExpandedNodeId> expandedNodeIds, BrowseDirection direction, int nodeClasses, int maxDepth, String filter,
boolean includeSubTypes, int maxNodesPerRequest) {
lock.lock();
try {
if (this.connected == null) {
return newNotConnectedResult();
}

return this.connected.browse(expandedNodeIds, direction, nodeClasses, 1, maxDepth,
null != filter ? Pattern.compile(filter) : null, includeSubTypes, maxNodesPerRequest)
.whenCompleteAsync((browseResult, e) -> {
// handle outside the lock, running using
// whenCompleteAsync (preserves the original exceptional completion)
if (e != null) {
handleConnectionFailure(e);
}
}, this.executor);
} finally {
lock.unlock();
final Connected connected = this.connected;
if (connected == null) {
return newNotConnectedResult();
}

return outcomeOf(connected.browse(expandedNodeIds, direction, nodeClasses, 1, maxDepth,
null != filter ? Pattern.compile(filter) : null, includeSubTypes, maxNodesPerRequest));
}

/**
* The outcome of a producer request, completed on the executor of the connection, so that the route continues there
* and not on a thread of Milo, which completes the responses; unlike thenApplyAsync, whenCompleteAsync hands a
* failure to the executor too. A request that fails completes with its failure and does not rebuild the connection:
* the client recovers a lost session itself, and a session that the server no longer has fails the transfer of the
* subscription, which rebuilds the connection.
* <p>
* The producers read the current connection without {@code lock}, which a service call of the consumers holds until
* the server answers, so they do not wait for it. A request on a connection that is disposed meanwhile fails with
* the status of its closed session.
*/
private <T> CompletableFuture<T> outcomeOf(final CompletableFuture<T> request) {
return request.whenCompleteAsync((result, e) -> {
}, this.executor);
}
}
7 changes: 3 additions & 4 deletions ...ents/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientExecutorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,10 +69,9 @@ public void configure() {
}

/**
* A failed request disconnects the client, and the disconnect can wait for a reconnect attempt that the client
* schedules on the shared scheduled executor of Milo, which has a single thread on a machine with one processor.
* The outcome of a request is therefore not handled on that executor: an awaited write completes while all of its
* threads are busy.
* An awaited write completes while every thread of the shared scheduled executor of Milo is busy: neither the
* request nor the handling of its outcome needs a thread of that executor, which has a single thread on a machine
* with one processor.
*/
@Test
void awaitedWriteCompletesWhileTheSharedScheduledExecutorOfMiloIsBusy() throws Exception {
Expand Down
Loading
Loading