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 48757552dee8 CAMEL-26462: camel-nats - a consumer that stops before
its task subscribes closes the dispatcher of that run (#29042)
48757552dee8 is described below
commit 48757552dee805b3c9bdb962db90ea7f61e14d62
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Nov 3 13:25:57 2026 +0530
CAMEL-26462: camel-nats - a consumer that stops before its task subscribes
closes the dispatcher of that run (#29042)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/component/nats/NatsConsumer.java | 73 +++++++--
.../nats/NatsConsumerJetStreamPushStopTest.java | 171 ++++++++++++++++++++-
2 files changed, 224 insertions(+), 20 deletions(-)
diff --git
a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java
b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java
index 7a31579e1d61..e5e0d7343996 100644
---
a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java
+++
b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java
@@ -64,7 +64,8 @@ public class NatsConsumer extends DefaultConsumer {
private final Processor processor;
private ExecutorService executor;
private Connection connection;
- private Dispatcher dispatcher;
+ // set by the consuming task and read when the consumer stops
+ private volatile Dispatcher dispatcher;
// the message handler of the dispatcher, which keeps the messages in
flight of that dispatcher (one per start)
private volatile NatsConsumingTask.CamelNatsMessageHandler
dispatcherHandler;
private boolean active;
@@ -123,11 +124,12 @@ public class NatsConsumer extends DefaultConsumer {
this.connection.flush(Duration.ofMillis(configuration.getFlushTimeout()));
}
- if (ObjectHelper.isNotEmpty(this.dispatcher)) {
+ final Dispatcher stoppedDispatcher = this.dispatcher;
+ if (stoppedDispatcher != null) {
try {
- this.dispatcher.unsubscribe(configuration.getTopic());
+ stoppedDispatcher.unsubscribe(configuration.getTopic());
} catch (final Exception e) {
- this.getExceptionHandler().handleException("Error during
unsubscribing", e);
+ unsubscribeFailed(stoppedDispatcher, e);
}
// a JetStream push subscription belongs to the dispatcher (it
cannot unsubscribe directly) and is on its
// deliver subject, not on the topic: end it through the
dispatcher, which matters when the connection is
@@ -135,15 +137,14 @@ public class NatsConsumer extends DefaultConsumer {
JetStreamSubscription pushSubscription =
this.jetStreamSubscription;
if (pushSubscription != null && pushSubscription.isActive()) {
try {
- this.dispatcher.unsubscribe(pushSubscription);
+ stoppedDispatcher.unsubscribe(pushSubscription);
} catch (final Exception e) {
- this.getExceptionHandler().handleException("Error during
unsubscribing", e);
+ unsubscribeFailed(stoppedDispatcher, e);
}
}
// the dispatcher was created by this consumer: close it (which
ends its thread) once no message is in
// flight, as closing it interrupts its thread, and Camel stops a
consumer before it waits for the exchanges
// in flight
- final Dispatcher stoppedDispatcher = this.dispatcher;
final Connection stoppedConnection = this.connection;
Runnable close = () -> closeDispatcher(stoppedConnection,
stoppedDispatcher);
NatsConsumingTask.CamelNatsMessageHandler handler =
this.dispatcherHandler;
@@ -185,6 +186,19 @@ public class NatsConsumer extends DefaultConsumer {
super.doStop();
}
+ /**
+ * Reports a failed unsubscribe of the stopping consumer, unless its
dispatcher is closed already: by the consuming
+ * task of this run, which saw the stop before it subscribed, or by the
close of the connection. jnats does not open
+ * a closed dispatcher again, so the check after the failure is not racy.
+ */
+ private void unsubscribeFailed(Dispatcher stoppedDispatcher, Exception e) {
+ if (stoppedDispatcher.isActive()) {
+ this.getExceptionHandler().handleException("Error during
unsubscribing", e);
+ } else {
+ LOG.debug("Not unsubscribing the dispatcher of the stopped
consumer, as it is closed already", e);
+ }
+ }
+
/**
* Closes a dispatcher of this consumer, unless its connection is closed
(which closed it).
*/
@@ -324,8 +338,12 @@ public class NatsConsumer extends DefaultConsumer {
NatsConsumer.this.setActive(true);
runPullFetchLoop(NatsConsumer.this.jetStreamSubscription,
messageHandler, cc);
} else {
- NatsConsumer.this.dispatcher =
this.connection.createDispatcher(messageHandler);
+ Dispatcher pushDispatcher =
this.connection.createDispatcher(messageHandler);
+ NatsConsumer.this.dispatcher = pushDispatcher;
NatsConsumer.this.dispatcherHandler = messageHandler;
+ if (stoppedBeforeSubscribe(pushDispatcher)) {
+ return;
+ }
PushSubscribeOptions pushOptions =
PushSubscribeOptions.builder()
.configuration(cc)
@@ -342,12 +360,12 @@ public class NatsConsumer extends DefaultConsumer {
// more than its inactive threshold, server restart); a
durable consumer is not created again
NatsJetStreamPushConsumerKeeper keeper = new
NatsJetStreamPushConsumerKeeper(
NatsConsumer.this, this.connection, streamName,
configuration.getTopic(), queueName,
- dispatcher, messageHandler, cc,
NatsConsumer.this.executor,
+ pushDispatcher, messageHandler, cc,
NatsConsumer.this.executor,
sub -> NatsConsumer.this.jetStreamSubscription = sub);
NatsConsumer.this.jetStreamSubscription =
this.connection.jetStream().subscribe(
NatsConsumer.this.getEndpoint().getConfiguration().getTopic(),
queueName,
- dispatcher,
+ pushDispatcher,
keeper,
autoAck,
pushOptions);
@@ -357,8 +375,10 @@ public class NatsConsumer extends DefaultConsumer {
streamInfo != null && streamInfo.getStreamState() !=
null
?
streamInfo.getStreamState().getLastSequence() : -1);
if (!NatsConsumer.this.isRunAllowed()) {
- // stopped while subscribing: doStop did not see the keeper
+ // stopped while subscribing: doStop did not see the
keeper (it saw the dispatcher, and ends the
+ // subscription through it)
keeper.stop();
+ return;
}
NatsConsumer.this.setActive(true);
@@ -587,19 +607,40 @@ public class NatsConsumer extends DefaultConsumer {
}
}
+ /**
+ * Whether the consumer stopped before this run subscribed with the
dispatcher it just set; if so, closes that
+ * dispatcher, which has no subscription yet. The dispatcher field is
volatile and written before the run state
+ * is read here, and a stop sets the state before doStop reads the
field: so either this run sees the stop, or
+ * doStop sees the dispatcher and closes it (on the stopping thread,
and a subscribe on the closed dispatcher
+ * fails). Closing it here, before any subscribe, matters: doStop
interrupts this thread, and jnats drops the
+ * UNSUB of a close on an interrupted thread, so the server would keep
a subscription that a later close ended.
+ */
+ private boolean stoppedBeforeSubscribe(Dispatcher runDispatcher) {
+ if (NatsConsumer.this.isRunAllowed()) {
+ return false;
+ }
+ LOG.debug("Not subscribing to topic {} as the consumer is
stopping", configuration.getTopic());
+ closeDispatcher(this.connection, runDispatcher);
+ return true;
+ }
+
private void setupStandardNatsConsumer(String topic, String queueName,
Integer maxMessages) {
CamelNatsMessageHandler messageHandler = new
CamelNatsMessageHandler();
- NatsConsumer.this.dispatcher =
connection.createDispatcher(messageHandler);
+ Dispatcher standardDispatcher =
connection.createDispatcher(messageHandler);
+ NatsConsumer.this.dispatcher = standardDispatcher;
NatsConsumer.this.dispatcherHandler = messageHandler;
+ if (stoppedBeforeSubscribe(standardDispatcher)) {
+ return;
+ }
if (ObjectHelper.isNotEmpty(queueName)) {
- NatsConsumer.this.dispatcher =
NatsConsumer.this.dispatcher.subscribe(topic, queueName);
+ standardDispatcher.subscribe(topic, queueName);
} else {
- NatsConsumer.this.dispatcher =
NatsConsumer.this.dispatcher.subscribe(topic);
+ standardDispatcher.subscribe(topic);
}
if (maxMessages != null) {
- NatsConsumer.this.dispatcher.unsubscribe(topic, maxMessages);
+ standardDispatcher.unsubscribe(topic, maxMessages);
}
- if (NatsConsumer.this.dispatcher.isActive()) {
+ if (standardDispatcher.isActive()) {
NatsConsumer.this.setActive(true);
}
}
diff --git
a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerJetStreamPushStopTest.java
b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerJetStreamPushStopTest.java
index 10f7f123e8ff..e36ee1b8e1ca 100644
---
a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerJetStreamPushStopTest.java
+++
b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerJetStreamPushStopTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.nats;
+import java.io.IOException;
import java.lang.reflect.Proxy;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
@@ -30,6 +31,7 @@ import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import io.nats.client.Connection;
import io.nats.client.Dispatcher;
@@ -219,6 +221,116 @@ class NatsConsumerJetStreamPushStopTest extends
CamelTestSupport {
assertEquals(0, server.activeSubscriptions(), "no subscription left
once the consumer stopped");
}
+ @Test
+ void
stopBeforeTheRunSetsItsDispatcherClosesThatDispatcherBeforeItSubscribes()
throws Exception {
+ DefaultConsumer consumer = (DefaultConsumer)
context.getRoute("orders").getConsumer();
+ ServiceHelper.stopService(consumer);
+ assertEquals(1, server.closedDispatchers(), "the stop closes the
dispatcher of the first run");
+ List<Throwable> reported = reportedBy(consumer);
+
+ // the consumer stops right after its task created the dispatcher,
before the task set it: the stop sees no
+ // dispatcher to close, and it interrupts the task. The task must
close its dispatcher before it subscribes: a
+ // subscribe on the interrupted thread fails, and a close there would
leave a subscription on the server
+ server.afterCreateDispatcher = () -> {
+ server.afterCreateDispatcher = null;
+ ServiceHelper.stopService(consumer);
+ };
+ server.lastTask = null;
+ ServiceHelper.startService(consumer);
+ awaitTaskOfTheLastRunDone();
+
+ assertEquals(1, server.subscribes(), "the stopped run did not
subscribe");
+ assertEquals(0, server.interruptedSubscribes(), "the stopped run did
not subscribe");
+ assertEquals(2, server.closedDispatchers(), "the dispatcher of each
run is closed");
+ assertEquals(0, server.activeSubscriptions(), "no subscription left
once the consumer stopped");
+ // a stop is not an error
+ assertTrue(reported.isEmpty(), () -> "reported: " + reported);
+ }
+
+ @Test
+ void stopThatSeesTheDispatcherClosedByTheRunIsNotReportedAsAnError()
throws Exception {
+ DefaultConsumer consumer = (DefaultConsumer)
context.getRoute("orders").getConsumer();
+ ServiceHelper.stopService(consumer);
+ List<Throwable> reported = reportedBy(consumer);
+ int closedBefore = server.closedDispatchers();
+
+ // the stop begins (on another thread) before the task sets its
dispatcher, and reads the dispatcher only after
+ // its flush: the task sees the stop and closes its dispatcher, and
then the stop sees that closed dispatcher
+ AtomicReference<Thread> stopper = new AtomicReference<>();
+ server.afterCreateDispatcher = () -> {
+ server.afterCreateDispatcher = null;
+ server.onFlush = () -> {
+ server.onFlush = null;
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
server.closedDispatchers() == closedBefore + 1);
+ };
+ Thread thread = new Thread(() ->
ServiceHelper.stopService(consumer), "stopper");
+ stopper.set(thread);
+ thread.start();
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
!consumer.isRunAllowed());
+ };
+ server.lastTask = null;
+ ServiceHelper.startService(consumer);
+ awaitTaskOfTheLastRunDone();
+ stopper.get().join(TimeUnit.SECONDS.toMillis(10));
+ assertFalse(stopper.get().isAlive(), "the stop ended");
+
+ assertTrue(ServiceHelper.isStopped(consumer), "the consumer stopped");
+ assertEquals(closedBefore + 1, server.closedDispatchers(), "the
dispatcher of the stopped run is closed");
+ assertEquals(1, server.subscribes(), "the stopped run did not
subscribe");
+ // the closed dispatcher is not an error of the stop
+ assertTrue(reported.isEmpty(), () -> "reported: " + reported);
+ }
+
+ @Test
+ void stopBeforeAStandardConsumerSetsItsDispatcherClosesThatDispatcher()
throws Exception {
+ // the same window for a consumer without JetStream on the shared
connection: its dispatcher would get the
+ // messages of its subject after the stop
+ DefaultConsumer standard = (DefaultConsumer)
context.getEndpoint("nats:plain?connection=#conn")
+ .createConsumer(exchange -> {
+ });
+ List<Throwable> reported = reportedBy(standard);
+ int closedBefore = server.closedDispatchers();
+ server.afterCreateDispatcher = () -> {
+ server.afterCreateDispatcher = null;
+ ServiceHelper.stopService(standard);
+ };
+ server.lastTask = null;
+ ServiceHelper.startService(standard);
+ awaitTaskOfTheLastRunDone();
+
+ assertEquals(closedBefore + 1, server.closedDispatchers(), "the
dispatcher of the stopped run is closed");
+ assertTrue(reported.isEmpty(), () -> "reported: " + reported);
+ }
+
+ private static List<Throwable> reportedBy(DefaultConsumer consumer) {
+ List<Throwable> reported = new CopyOnWriteArrayList<>();
+ consumer.setExceptionHandler(new ExceptionHandler() {
+ @Override
+ public void handleException(Throwable exception) {
+ reported.add(exception);
+ }
+
+ @Override
+ public void handleException(String message, Throwable exception) {
+ reported.add(exception);
+ }
+
+ @Override
+ public void handleException(String message, Exchange exchange,
Throwable exception) {
+ reported.add(exception);
+ }
+ });
+ return reported;
+ }
+
+ /**
+ * Waits until the task of the last run of the consumer returned: the stop
shut its executor down, so its thread
+ * ends once the task is done.
+ */
+ private void awaitTaskOfTheLastRunDone() {
+ await().atMost(10, TimeUnit.SECONDS).until(() -> server.lastTask !=
null && !server.lastTask.isAlive());
+ }
+
/**
* A JetStream server with one stream, faked at the level of the jnats
interfaces the consumer calls. Each push
* subscription has its own consumer (DeliverPolicy.New starts after the
last published message) which delivers on
@@ -226,7 +338,10 @@ class NatsConsumerJetStreamPushStopTest extends
CamelTestSupport {
* unsubscribe directly, the dispatcher unsubscribes it by subscription
(not by the topic: a push subscription is on
* its deliver subject), and closing the dispatcher unsubscribes all its
subscriptions and interrupts the thread
* that delivers its messages (jnats NatsDispatcher.stop cancels its
thread with Future.cancel(true)). A closed
- * dispatcher refuses to subscribe or unsubscribe ("Dispatcher is closed").
+ * dispatcher refuses to subscribe or unsubscribe ("Dispatcher is
closed"), and closing it again is refused
+ * ("Dispatcher is already closed."). A subscribe on an interrupted thread
fails with an IOException, as the request
+ * that jnats makes to the server first does, and a close on an
interrupted thread leaves its subscriptions on the
+ * server, as jnats drops the UNSUB that it queues on an interrupted
thread.
*/
static final class FakeServer {
final List<byte[]> stream = new ArrayList<>();
@@ -235,7 +350,15 @@ class NatsConsumerJetStreamPushStopTest extends
CamelTestSupport {
final Set<Dispatcher> closed = Collections.newSetFromMap(new
IdentityHashMap<>());
// runs on the thread of the consumer, before a push subscribe reaches
the server
volatile Runnable beforeSubscribe;
+ // runs on the thread of the consumer, after it created a dispatcher
and before the consumer gets it
+ volatile Runnable afterCreateDispatcher;
+ // the thread of the consumer that created the last dispatcher
+ volatile Thread lastTask;
+ // runs when the consumer flushes the connection, which doStop does
before it reads the dispatcher
+ volatile Runnable onFlush;
int refusedSubscribes;
+ int interruptedSubscribes;
+ int subscribes;
final class FakeSubscription {
final Dispatcher dispatcher;
@@ -259,6 +382,14 @@ class NatsConsumerJetStreamPushStopTest extends
CamelTestSupport {
return refusedSubscribes;
}
+ synchronized int interruptedSubscribes() {
+ return interruptedSubscribes;
+ }
+
+ synchronized int subscribes() {
+ return subscribes;
+ }
+
synchronized int closedDispatchers() {
return closed.size();
}
@@ -284,11 +415,26 @@ class NatsConsumerJetStreamPushStopTest extends
CamelTestSupport {
case "getServerInfo" -> info;
case "jetStreamManagement" -> jsm;
case "jetStream" -> js;
- case "createDispatcher" -> createDispatcher();
+ case "createDispatcher" -> {
+ lastTask = Thread.currentThread();
+ Dispatcher d = createDispatcher();
+ Runnable hook = afterCreateDispatcher;
+ if (hook != null) {
+ hook.run();
+ }
+ yield d;
+ }
case "closeDispatcher" -> {
closeDispatcher((Dispatcher) args[0]);
yield null;
}
+ case "flush" -> {
+ Runnable hook = onFlush;
+ if (hook != null) {
+ hook.run();
+ }
+ yield null;
+ }
case "getStatus" -> Connection.Status.CONNECTED;
default -> null;
});
@@ -298,6 +444,10 @@ class NatsConsumerJetStreamPushStopTest extends
CamelTestSupport {
Dispatcher[] self = new Dispatcher[1];
self[0] = proxy(Dispatcher.class, (name, args) -> {
synchronized (this) {
+ if ("isActive".equals(name)) {
+ // as jnats: false once the dispatcher is closed
+ return !closed.contains(self[0]);
+ }
if (closed.contains(self[0])) {
throw new IllegalStateException("Dispatcher is
closed");
}
@@ -323,9 +473,14 @@ class NatsConsumerJetStreamPushStopTest extends
CamelTestSupport {
if (!closed.add(d)) {
throw new IllegalArgumentException("Dispatcher is already
closed.");
}
+ // jnats queues the UNSUB of each subscription of the dispatcher,
and its queue drops a message queued on an
+ // interrupted thread (WriterMessageQueue.push with tryLock): the
server then keeps the subscription
+ boolean unsubscribeSent = !Thread.currentThread().isInterrupted();
for (FakeSubscription f : subscriptions) {
if (f.dispatcher == d) {
- f.active = false;
+ if (unsubscribeSent) {
+ f.active = false;
+ }
// the thread of the dispatcher is cancelled with an
interrupt
f.thread.interrupt();
}
@@ -334,11 +489,19 @@ class NatsConsumerJetStreamPushStopTest extends
CamelTestSupport {
}
private synchronized JetStreamSubscription subscribe(
- ConsumerConfiguration cc, Dispatcher dispatcher,
MessageHandler handler) {
+ ConsumerConfiguration cc, Dispatcher dispatcher,
MessageHandler handler)
+ throws IOException {
if (closed.contains(dispatcher)) {
refusedSubscribes++;
throw new IllegalStateException("Dispatcher is closed");
}
+ if (Thread.currentThread().isInterrupted()) {
+ // the subscribe asks the server first (jnats
NatsJetStreamImpl.makeRequestResponseRequired), and an
+ // interrupted request fails with an IOException (the thread
stays interrupted)
+ interruptedSubscribes++;
+ throw new IOException(new InterruptedException());
+ }
+ subscribes++;
FakeSubscription f = new FakeSubscription(dispatcher);
f.next = cc != null && cc.getDeliverPolicy() == DeliverPolicy.New
? stream.size() + 1 : 1;
subscriptions.add(f);
