This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 74c573d323c6 CAMEL-26570: camel-milo - deliver the loss report of a
consumer started without a connection off the request pool (#29197)
74c573d323c6 is described below
commit 74c573d323c6c42ce3511c3152d9600e00eb7366
Author: allthingssecurity <[email protected]>
AuthorDate: Thu Nov 5 12:34:19 2026 +0530
CAMEL-26570: camel-milo - deliver the loss report of a consumer started
without a connection off the request pool (#29197)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../milo/client/internal/SubscriptionManager.java | 6 ++-
.../component/milo/CommunicationLossTest.java | 62 ++++++++++++++++++++++
2 files changed, 66 insertions(+), 2 deletions(-)
diff --git
a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java
b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java
index f62ec04dee41..4c7d45fe8385 100644
---
a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java
+++
b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java
@@ -66,6 +66,7 @@ import
org.eclipse.milo.opcua.sdk.client.subscriptions.OpcUaMonitoredItem;
import org.eclipse.milo.opcua.sdk.client.subscriptions.OpcUaSubscription;
import org.eclipse.milo.opcua.stack.core.AttributeId;
import org.eclipse.milo.opcua.stack.core.NodeIds;
+import org.eclipse.milo.opcua.stack.core.Stack;
import org.eclipse.milo.opcua.stack.core.StatusCodes;
import org.eclipse.milo.opcua.stack.core.UaException;
import org.eclipse.milo.opcua.stack.core.security.CertificateValidator;
@@ -1710,9 +1711,10 @@ public class SubscriptionManager {
if (this.connected == null) {
// no communication with the server: the nodes are reported
off the thread that starts the consumer,
// with a generation below every connection's, so the report
is dropped once any connection has
- // delivered their values, also one that was being built while
they were registered
+ // delivered their values, also one that was being built while
they were registered. Not on the request
+ // pool: the route may wait for a request of this connection,
whose outcome needs a thread of that pool
final DateTime time = DateTime.now();
- this.executor.execute(() -> deliverLoss(subscriptions, 0L,
time));
+ Stack.sharedExecutor().execute(() ->
deliverLoss(subscriptions, 0L, time));
} else if (!this.connected.listener.sessionActive) {
// the session is lost; the nodes are reported before the
values of their items, on the same queue
reportLoss(this.connected.manager, this.connected.generation,
subscriptions);
diff --git
a/components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java
index 48d16171fff9..467f5543d90a 100644
---
a/components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java
+++
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java
@@ -26,6 +26,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.camel.EndpointInject;
@@ -45,11 +46,14 @@ import org.eclipse.milo.opcua.stack.core.StatusCodes;
import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
import org.eclipse.milo.opcua.stack.core.types.builtin.ExpandedNodeId;
import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
+import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
+import org.eclipse.milo.opcua.stack.core.types.builtin.Variant;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.extension.RegisterExtension;
import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -113,6 +117,9 @@ public class CommunicationLossTest extends
AbstractMiloServerTest {
@RegisterExtension
AvailablePortFinder.Port disposedRelayPort = AvailablePortFinder.find();
+ @RegisterExtension
+ AvailablePortFinder.Port reportRelayPort = AvailablePortFinder.find();
+
private final TcpRelay relay = new TcpRelay();
private final TcpRelay lateRelay = new TcpRelay();
@@ -121,6 +128,8 @@ public class CommunicationLossTest extends
AbstractMiloServerTest {
private final TcpRelay disposedRelay = new TcpRelay();
+ private final TcpRelay reportRelay = new TcpRelay();
+
@Override
protected RoutesBuilder createRouteBuilder() {
return new RouteBuilder() {
@@ -166,6 +175,7 @@ public class CommunicationLossTest extends
AbstractMiloServerTest {
lateRelay.stop();
reconnectTimeoutRelay.stop();
disposedRelay.stop();
+ reportRelay.stop();
}
@Test
@@ -391,6 +401,58 @@ public class CommunicationLossTest extends
AbstractMiloServerTest {
}
}
+ /**
+ * The report of a consumer that starts while the server cannot be reached
runs its route, which can wait for a
+ * request of the same connection, here a write once the connection is up,
as a redelivery does. The outcome of the
+ * request does not wait for the thread that delivers the report, also
when the pool that completes the requests has
+ * a single thread, as on a machine with one processor.
+ */
+ @Test
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ void consumerThatWaitsForAWriteOnItsLossReportCompletesWithOneThread()
throws Exception {
+ producer1.sendBody("Foo");
+
+ final MiloClientEndpoint endpoint = context.getEndpoint(
+ resolve(MILO_CLIENT_ITEM_1).replace("@@relay@@",
Integer.toString(reportRelayPort.getPort())),
+ MiloClientEndpoint.class);
+ final ScheduledExecutorService reconnectExecutor =
Executors.newSingleThreadScheduledExecutor();
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ // the relay is down: the manager cannot connect
+ final SubscriptionManager manager
+ = new SubscriptionManager(endpoint.getConfiguration(),
reconnectExecutor, executor, 200);
+ final CountDownLatch connected = new CountDownLatch(1);
+ final CompletableFuture<Object> write = new CompletableFuture<>();
+ try {
+ final ExpandedNodeId item =
ExpandedNodeId.parse(endpoint.getNode());
+ manager.registerItems(Map.of(item, value -> {
+ if (value.getStatusCode().getValue() !=
StatusCodes.Bad_NoCommunication || write.isDone()) {
+ return;
+ }
+ try {
+ if (!connected.await(30, TimeUnit.SECONDS)) {
+ throw new TimeoutException("The connection did not
come up");
+ }
+ write.complete(manager.write(item, new DataValue(new
Variant("Bar"))).get(10, TimeUnit.SECONDS));
+ } catch (final Exception e) {
+ write.completeExceptionally(e);
+ }
+ }), 50.0, null, new MonitorFilterConfiguration());
+
+ reportRelay.start(reportRelayPort.getPort(), getServerPort());
+ await().atMost(30, TimeUnit.SECONDS).until(() ->
manager.getOpcUaClient() != null);
+ connected.countDown();
+
+ final Object status = assertDoesNotThrow(() -> write.get(30,
TimeUnit.SECONDS),
+ "The write of the consumer did not complete: its outcome
waits for the thread that delivers the report");
+ assertEquals(StatusCode.GOOD, status);
+ } finally {
+ connected.countDown();
+ manager.dispose();
+ reconnectExecutor.shutdownNow();
+ executor.shutdownNow();
+ }
+ }
+
/**
* A connection that the manager completes after it has been disposed is
closed: the manager no longer holds it, so
* nothing else would close its session.
