From e59ecd134dd1b6da7985ce934ef848742262d06c Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Tue, 20 Oct 2026 08:46:54 +0530 Subject: [PATCH 1/4] CAMEL-25847: camel-nats - a stopped JetStream push consumer unsubscribes through its dispatcher A JetStream push subscription belongs to the dispatcher of the consumer. On stop the consumer unsubscribed the topic from the dispatcher (the push subscription is on its deliver subject, not on the topic) and then called unsubscribe() on the subscription, which jnats refuses for a subscription of a dispatcher ("cannot respond to unsubscribe directly"; only a WARN). With a shared connection (connection option), which stays open, the subscription stayed active: the stopped route kept getting messages, which failed and were nak'ed and delivered again at once, and after the route was started again every message was routed twice (the old and the new subscription). The dispatcher and its thread were also never closed. The consumer now unsubscribes the push subscription through its dispatcher when it stops. It also closes the dispatcher it created, which ends its thread, once no message is in flight on it: jnats closes a dispatcher by cancelling its thread with an interrupt, and that thread runs the route, which Camel stops before it waits for the exchanges in flight; when a message is in flight, the dispatcher is closed when that message is done. Co-Authored-By: Claude Opus 5.5 --- .../camel/component/nats/NatsConsumer.java | 68 ++++ .../NatsConsumerJetStreamPushStopTest.java | 363 ++++++++++++++++++ 2 files changed, 431 insertions(+) create mode 100644 components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerJetStreamPushStopTest.java 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 567fe9673e91f..25cef4dd0002b 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 @@ -65,6 +65,11 @@ public class NatsConsumer extends DefaultConsumer { private ExecutorService executor; private Connection connection; private Dispatcher dispatcher; + // the messages that the handler is processing, and the close of the dispatcher of a stopped consumer that waits + // until none is: closing a dispatcher interrupts its thread, which runs the route + private final Object inFlightLock = new Object(); + private int messagesInFlight; + private Runnable closeDispatcherWhenIdle; private boolean active; // set by the consuming task, and again by the push consumer keeper, and read when the consumer stops private volatile JetStreamSubscription jetStreamSubscription; @@ -127,6 +132,34 @@ protected void doStop() throws Exception { } catch (final Exception e) { this.getExceptionHandler().handleException("Error during unsubscribing", 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 + // shared and stays open. This does not interrupt the dispatcher thread, which may still run an exchange + JetStreamSubscription pushSubscription = this.jetStreamSubscription; + if (pushSubscription != null && pushSubscription.isActive()) { + try { + this.dispatcher.unsubscribe(pushSubscription); + } catch (final Exception e) { + this.getExceptionHandler().handleException("Error during unsubscribing", 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); + synchronized (this.inFlightLock) { + if (this.messagesInFlight > 0) { + // closed by the handler once the message is done + this.closeDispatcherWhenIdle = close; + close = null; + } + } + if (close != null) { + close.run(); + } + this.dispatcher = null; } if (ObjectHelper.isNotEmpty(this.jetStreamSubscription) && this.jetStreamSubscription.isActive()) { @@ -156,6 +189,19 @@ protected void doStop() throws Exception { super.doStop(); } + /** + * Closes a dispatcher of this consumer, unless its connection is closed (which closed it). + */ + private void closeDispatcher(Connection c, Dispatcher d) { + if (c != null && c.getStatus() != Connection.Status.CLOSED) { + try { + c.closeDispatcher(d); + } catch (final Exception e) { + LOG.debug("Error closing the dispatcher of the stopped consumer", e); + } + } + } + public boolean isActive() { return this.active; } @@ -559,6 +605,28 @@ class CamelNatsMessageHandler implements MessageHandler { @Override public void onMessage(Message msg) throws InterruptedException { + synchronized (NatsConsumer.this.inFlightLock) { + NatsConsumer.this.messagesInFlight++; + } + try { + processMessage(msg); + } finally { + Runnable close = null; + synchronized (NatsConsumer.this.inFlightLock) { + if (--NatsConsumer.this.messagesInFlight == 0) { + close = NatsConsumer.this.closeDispatcherWhenIdle; + NatsConsumer.this.closeDispatcherWhenIdle = null; + } + } + if (close != null) { + // the consumer stopped while this message was processed: close its dispatcher now (on this + // thread, which the close interrupts once the message is done) + close.run(); + } + } + } + + private void processMessage(Message msg) throws InterruptedException { LOG.debug("Received Message: {}", msg); final Exchange exchange = NatsConsumer.this.createExchange(false); 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 new file mode 100644 index 0000000000000..cd021910ed5d7 --- /dev/null +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerJetStreamPushStopTest.java @@ -0,0 +1,363 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.nats; + +import java.lang.reflect.Proxy; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.IdentityHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import io.nats.client.Connection; +import io.nats.client.Dispatcher; +import io.nats.client.JetStream; +import io.nats.client.JetStreamManagement; +import io.nats.client.JetStreamSubscription; +import io.nats.client.Message; +import io.nats.client.MessageHandler; +import io.nats.client.PushSubscribeOptions; +import io.nats.client.api.ConsumerConfiguration; +import io.nats.client.api.DeliverPolicy; +import io.nats.client.api.ServerInfo; +import io.nats.client.impl.NatsJetStreamMetaData; +import io.nats.client.impl.NatsMessage; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.support.service.ServiceHelper; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.junit.jupiter.api.Test; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * A JetStream push subscription belongs to the dispatcher of the consumer, and jnats refuses to unsubscribe it directly + * ("Subscriptions that belong to a dispatcher cannot respond to unsubscribe directly"). When the connection is shared + * (connection option) it stays open when the route stops, so the consumer must unsubscribe through its dispatcher; + * otherwise the stopped route keeps getting messages, and after a restart every message is routed twice. + */ +class NatsConsumerJetStreamPushStopTest extends CamelTestSupport { + + private final FakeServer server = new FakeServer(); + private final CountDownLatch slowEntered = new CountDownLatch(1); + private final CountDownLatch releaseSlow = new CountDownLatch(1); + private final AtomicBoolean slowInterrupted = new AtomicBoolean(); + private final AtomicBoolean slowDone = new AtomicBoolean(); + + @Override + protected RouteBuilder createRouteBuilder() { + context.getRegistry().bind("conn", server.connection()); + context.getRegistry().bind("cc", ConsumerConfiguration.builder().deliverPolicy(DeliverPolicy.New).build()); + return new RouteBuilder() { + @Override + public void configure() { + from("nats:orders?connection=#conn&jetstreamEnabled=true&jetstreamName=ORDERS&pullSubscription=false" + + "&consumerConfiguration=#cc") + .routeId("orders") + .convertBodyTo(String.class) + .process(e -> { + if ("slow".equals(e.getMessage().getBody())) { + // an exchange in flight while the route stops, such as a call that waits on I/O + slowEntered.countDown(); + try { + releaseSlow.await(20, TimeUnit.SECONDS); + } catch (InterruptedException ie) { + slowInterrupted.set(true); + Thread.currentThread().interrupt(); + throw ie; + } + slowDone.set(true); + } + }) + .to("mock:result"); + } + }; + } + + @Test + void stoppedRouteGetsNoMessagesAndRestartedRouteGetsEachMessageOnce() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedBodiesReceived("order-1"); + server.publish("order-1"); + mock.assertIsSatisfied(); + + context.getRouteController().stopRoute("orders"); + int afterStop = server.activeSubscriptions(); + + // the route is started again: one subscription, each message routed once + context.getRouteController().startRoute("orders"); + await().atMost(10, TimeUnit.SECONDS).until(() -> server.activeSubscriptions() > 0); + mock.reset(); + mock.expectedBodiesReceived("order-2"); + // and no second copy shortly after + mock.setAssertPeriod(500); + server.publish("order-2"); + mock.assertIsSatisfied(); + assertEquals(0, afterStop, "no subscription left once the route stopped"); + assertEquals(1, server.activeSubscriptions(), "one subscription once the route started again"); + assertEquals(1, server.closedDispatchers(), "the dispatcher of the first run is closed"); + } + + @Test + void stoppingTheRouteDoesNotInterruptTheExchangeInFlight() throws Exception { + server.publish("slow"); + assertTrue(slowEntered.await(10, TimeUnit.SECONDS), "the exchange is in flight"); + + // the shutdown strategy stops the consumer (it is not Suspendable) and then waits for the exchange in flight + Thread stopper = new Thread(() -> { + try { + context.getRouteController().stopRoute("orders"); + } catch (Exception e) { + throw new IllegalStateException(e); + } + }, "route-stopper"); + stopper.start(); + await().atMost(10, TimeUnit.SECONDS) + .until(() -> ServiceHelper.isStopped(context.getRoute("orders").getConsumer())); + assertEquals(0, server.activeSubscriptions(), "no subscription left once the consumer stopped"); + releaseSlow.countDown(); + stopper.join(TimeUnit.SECONDS.toMillis(20)); + + assertFalse(slowInterrupted.get(), "the step in flight must not be interrupted by the stop"); + assertTrue(slowDone.get(), "the step in flight completes"); + // the dispatcher (and its thread) is closed once the message is done + await().atMost(10, TimeUnit.SECONDS).until(() -> server.closedDispatchers() == 1); + } + + /** + * 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 + * its own thread while the subscription is active. Like jnats, a subscription that belongs to a dispatcher cannot + * 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"). + */ + static final class FakeServer { + final List stream = new ArrayList<>(); + final List subscriptions = new ArrayList<>(); + final Map bySubscription = new IdentityHashMap<>(); + final Set closed = Collections.newSetFromMap(new IdentityHashMap<>()); + + final class FakeSubscription { + final Dispatcher dispatcher; + Thread thread; + long next; + boolean active = true; + final List redeliver = new ArrayList<>(); + final Map naks = new HashMap<>(); + + FakeSubscription(Dispatcher dispatcher) { + this.dispatcher = dispatcher; + } + } + + synchronized void publish(String body) { + stream.add(body.getBytes(StandardCharsets.UTF_8)); + notifyAll(); + } + + synchronized int closedDispatchers() { + return closed.size(); + } + + synchronized int activeSubscriptions() { + return (int) subscriptions.stream().filter(s -> s.active).count(); + } + + Connection connection() { + JetStreamManagement jsm = proxy(JetStreamManagement.class, (name, args) -> null); + JetStream js = proxy(JetStream.class, (name, args) -> { + if ("subscribe".equals(name) && args.length == 6 && args[5] instanceof PushSubscribeOptions o) { + return subscribe(o.getConsumerConfiguration(), (Dispatcher) args[2], (MessageHandler) args[3]); + } + throw new UnsupportedOperationException(name); + }); + ServerInfo info = new ServerInfo("INFO {\"server_id\":\"fake\",\"jetstream\":true}"); + return proxy(Connection.class, (name, args) -> switch (name) { + case "getServerInfo" -> info; + case "jetStreamManagement" -> jsm; + case "jetStream" -> js; + case "createDispatcher" -> createDispatcher(); + case "closeDispatcher" -> { + closeDispatcher((Dispatcher) args[0]); + yield null; + } + case "getStatus" -> Connection.Status.CONNECTED; + default -> null; + }); + } + + private Dispatcher createDispatcher() { + Dispatcher[] self = new Dispatcher[1]; + self[0] = proxy(Dispatcher.class, (name, args) -> { + synchronized (this) { + if (closed.contains(self[0])) { + throw new IllegalStateException("Dispatcher is closed"); + } + } + if ("unsubscribe".equals(name) && args[0] instanceof JetStreamSubscription s) { + unsubscribe(s); + } + // unsubscribe by subject: a push subscription is on its deliver subject, not on the topic + return self[0]; + }); + return self[0]; + } + + private synchronized void unsubscribe(JetStreamSubscription s) { + FakeSubscription f = bySubscription.get(s); + if (f != null) { + f.active = false; + notifyAll(); + } + } + + private synchronized void closeDispatcher(Dispatcher d) { + if (!closed.add(d)) { + throw new IllegalArgumentException("Dispatcher is already closed."); + } + for (FakeSubscription f : subscriptions) { + if (f.dispatcher == d) { + f.active = false; + // the thread of the dispatcher is cancelled with an interrupt + f.thread.interrupt(); + } + } + notifyAll(); + } + + private synchronized JetStreamSubscription subscribe( + ConsumerConfiguration cc, Dispatcher dispatcher, MessageHandler handler) { + if (closed.contains(dispatcher)) { + throw new IllegalStateException("Dispatcher is closed"); + } + FakeSubscription f = new FakeSubscription(dispatcher); + f.next = cc != null && cc.getDeliverPolicy() == DeliverPolicy.New ? stream.size() + 1 : 1; + subscriptions.add(f); + JetStreamSubscription sub = proxy(JetStreamSubscription.class, (name, args) -> switch (name) { + case "unsubscribe" -> throw new IllegalStateException( + "Subscriptions that belong to a dispatcher cannot respond to unsubscribe directly."); + case "isActive" -> { + synchronized (this) { + yield f.active; + } + } + case "getDispatcher" -> f.dispatcher; + default -> null; + }); + bySubscription.put(sub, f); + Thread t = new Thread(() -> deliver(f, sub, handler), "fake-push-" + subscriptions.size()); + t.setDaemon(true); + f.thread = t; + t.start(); + return sub; + } + + private void deliver(FakeSubscription f, JetStreamSubscription sub, MessageHandler handler) { + while (true) { + long seq; + synchronized (this) { + while (f.active && f.redeliver.isEmpty() && f.next > stream.size()) { + try { + wait(); + } catch (InterruptedException e) { + return; + } + } + if (!f.active) { + return; + } + seq = f.redeliver.isEmpty() ? f.next++ : f.redeliver.remove(0); + } + try { + handler.onMessage(message(f, sub, seq)); + } catch (InterruptedException e) { + return; + } + } + } + + private Message message(FakeSubscription f, JetStreamSubscription sub, long seq) { + byte[] data; + synchronized (this) { + data = stream.get((int) seq - 1); + } + // a JetStream message carries its metadata in the reply subject of the acknowledgement + NatsJetStreamMetaData meta = new NatsJetStreamMetaData( + new NatsMessage("orders", "$JS.ACK.ORDERS.c.1." + seq + "." + seq + ".0.0", data) { + @Override + public boolean isJetStream() { + return true; + } + }); + return proxy(Message.class, (name, args) -> switch (name) { + case "getData" -> data; + case "getSubject" -> "orders"; + case "getSID" -> "1"; + case "getSubscription" -> sub; + case "isJetStream" -> true; + case "isStatusMessage" -> false; + case "metaData" -> meta; + case "nak", "nakWithDelay" -> { + synchronized (this) { + // like maxDeliver=3: a message nak'ed three times is not delivered again + if (f.naks.merge(seq, 1, Integer::sum) < 3 && f.active) { + f.redeliver.add(seq); + notifyAll(); + } + } + yield null; + } + default -> null; + }); + } + + @FunctionalInterface + interface Handler { + Object handle(String method, Object[] args) throws Exception; + } + + @SuppressWarnings("unchecked") + static T proxy(Class type, Handler handler) { + return (T) Proxy.newProxyInstance(FakeServer.class.getClassLoader(), new Class[] { type }, + (p, m, a) -> { + if (m.getDeclaringClass() == Object.class) { + return switch (m.getName()) { + case "hashCode" -> System.identityHashCode(p); + case "equals" -> p == a[0]; + default -> type.getSimpleName() + "@fake"; + }; + } + Object r = handler.handle(m.getName(), a == null ? new Object[0] : a); + if (r == null && m.getReturnType() == boolean.class) { + return false; + } + return r; + }); + } + } +} From 99f01a1f409c2ba2bf8c2ce4afcb7ee3cc9c1539 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Fri, 23 Oct 2026 15:07:57 +0530 Subject: [PATCH 2/4] CAMEL-25847: camel-nats - keep the messages in flight per dispatcher, not per consumer Review of #28320 (optional point 1): the in-flight count and the pending close of the dispatcher were fields of the consumer, while each start makes a new dispatcher and a new CamelNatsMessageHandler. A consumer that is stopped and started again in place (as a route policy such as a throttling policy does with a consumer that is not Suspendable) while an exchange of the previous run is still in flight closed the old dispatcher only once the new run was idle too, and a second stop before that overwrote the pending close, so the first dispatcher was never closed. The count and the pending close now live on the CamelNatsMessageHandler, one per dispatcher. The consumer keeps the handler next to its dispatcher, and doStop asks that handler to close the dispatcher now or once its message in flight is done. Test: eachRunOfTheConsumerClosesItsOwnDispatcher (stop, start and stop the same consumer while a message of the first run is in flight) fails without the change: the dispatcher of the second run is not closed when it stops. Co-Authored-By: Claude Opus 5.5 --- .../camel/component/nats/NatsConsumer.java | 58 ++++++++++++------- .../NatsConsumerJetStreamPushStopTest.java | 24 ++++++++ 2 files changed, 62 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 25cef4dd0002b..3546af5248643 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 @@ -65,11 +65,8 @@ public class NatsConsumer extends DefaultConsumer { private ExecutorService executor; private Connection connection; private Dispatcher dispatcher; - // the messages that the handler is processing, and the close of the dispatcher of a stopped consumer that waits - // until none is: closing a dispatcher interrupts its thread, which runs the route - private final Object inFlightLock = new Object(); - private int messagesInFlight; - private Runnable closeDispatcherWhenIdle; + // 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; // set by the consuming task, and again by the push consumer keeper, and read when the consumer stops private volatile JetStreamSubscription jetStreamSubscription; @@ -149,17 +146,16 @@ protected void doStop() throws Exception { final Dispatcher stoppedDispatcher = this.dispatcher; final Connection stoppedConnection = this.connection; Runnable close = () -> closeDispatcher(stoppedConnection, stoppedDispatcher); - synchronized (this.inFlightLock) { - if (this.messagesInFlight > 0) { - // closed by the handler once the message is done - this.closeDispatcherWhenIdle = close; - close = null; - } - } - if (close != null) { + NatsConsumingTask.CamelNatsMessageHandler handler = this.dispatcherHandler; + if (handler != null) { + // the handler of this dispatcher only: a message of an earlier run that is still in flight (after a + // forced stop) closes the dispatcher of its own run + handler.closeWhenIdle(close); + } else { close.run(); } this.dispatcher = null; + this.dispatcherHandler = null; } if (ObjectHelper.isNotEmpty(this.jetStreamSubscription) && this.jetStreamSubscription.isActive()) { @@ -322,6 +318,7 @@ private void setupJetStreamConsumer(String topic, String queueName) throws IOExc runPullFetchLoop(NatsConsumer.this.jetStreamSubscription, messageHandler, cc); } else { NatsConsumer.this.dispatcher = this.connection.createDispatcher(messageHandler); + NatsConsumer.this.dispatcherHandler = messageHandler; PushSubscribeOptions pushOptions = PushSubscribeOptions.builder() .configuration(cc) @@ -584,7 +581,9 @@ private void acknowledged(Message msg) { } private void setupStandardNatsConsumer(String topic, String queueName, Integer maxMessages) { - NatsConsumer.this.dispatcher = connection.createDispatcher(new CamelNatsMessageHandler()); + CamelNatsMessageHandler messageHandler = new CamelNatsMessageHandler(); + NatsConsumer.this.dispatcher = connection.createDispatcher(messageHandler); + NatsConsumer.this.dispatcherHandler = messageHandler; if (ObjectHelper.isNotEmpty(queueName)) { NatsConsumer.this.dispatcher = NatsConsumer.this.dispatcher.subscribe(topic, queueName); } else { @@ -602,20 +601,39 @@ class CamelNatsMessageHandler implements MessageHandler { final boolean ackPolicyNone = configuration.getAckPolicy() == AckPolicy.None; final boolean manualAckEnabled = configuration.isManualAck() && configuration.isJetstreamEnabled(); + // the messages that this handler is processing, and the close of its dispatcher (the consumer stopped) + // that waits until none is: closing a dispatcher interrupts its thread, which runs the route + private final Object inFlightLock = new Object(); + private int messagesInFlight; + private Runnable pendingClose; + + /** + * Closes the dispatcher of this handler now, or once the message in flight is done. + */ + void closeWhenIdle(Runnable close) { + synchronized (inFlightLock) { + if (messagesInFlight > 0) { + // closed by onMessage once the message is done + pendingClose = close; + return; + } + } + close.run(); + } @Override public void onMessage(Message msg) throws InterruptedException { - synchronized (NatsConsumer.this.inFlightLock) { - NatsConsumer.this.messagesInFlight++; + synchronized (inFlightLock) { + messagesInFlight++; } try { processMessage(msg); } finally { Runnable close = null; - synchronized (NatsConsumer.this.inFlightLock) { - if (--NatsConsumer.this.messagesInFlight == 0) { - close = NatsConsumer.this.closeDispatcherWhenIdle; - NatsConsumer.this.closeDispatcherWhenIdle = null; + synchronized (inFlightLock) { + if (--messagesInFlight == 0) { + close = pendingClose; + pendingClose = null; } } if (close != null) { 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 cd021910ed5d7..a847ab7c856a9 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 @@ -42,6 +42,7 @@ import io.nats.client.api.ServerInfo; import io.nats.client.impl.NatsJetStreamMetaData; import io.nats.client.impl.NatsMessage; +import org.apache.camel.Consumer; import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.mock.MockEndpoint; import org.apache.camel.support.service.ServiceHelper; @@ -147,6 +148,29 @@ void stoppingTheRouteDoesNotInterruptTheExchangeInFlight() throws Exception { await().atMost(10, TimeUnit.SECONDS).until(() -> server.closedDispatchers() == 1); } + @Test + void eachRunOfTheConsumerClosesItsOwnDispatcher() throws Exception { + server.publish("slow"); + assertTrue(slowEntered.await(10, TimeUnit.SECONDS), "the exchange is in flight"); + + // a route policy (such as a throttling policy) stops and starts the same consumer, which is not Suspendable, + // while an exchange is in flight: each start makes a new dispatcher + Consumer consumer = context.getRoute("orders").getConsumer(); + ServiceHelper.stopService(consumer); + assertEquals(0, server.closedDispatchers(), "the dispatcher of the exchange in flight is not closed yet"); + ServiceHelper.startService(consumer); + await().atMost(10, TimeUnit.SECONDS).until(() -> server.activeSubscriptions() > 0); + + // the second run has no message in flight, so its dispatcher is closed when it stops + ServiceHelper.stopService(consumer); + assertEquals(1, server.closedDispatchers(), "the dispatcher of the second run is closed when it stops"); + + // the exchange of the first run is done: the dispatcher of the first run is closed too + releaseSlow.countDown(); + await().atMost(10, TimeUnit.SECONDS).until(() -> server.closedDispatchers() == 2); + assertFalse(slowInterrupted.get(), "the step in flight must not be interrupted by the stops"); + } + /** * 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 From 6e44b86caacd327e189a3a0f9fe2ef828445021c Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Fri, 23 Oct 2026 16:22:37 +0530 Subject: [PATCH 3/4] CAMEL-25847: camel-nats - a stop during the subscribe of the consumer is not reported as an error When the consumer stops while its task is still in jetStream().subscribe(...), the stop closes the dispatcher that the task passes to the subscribe, and jnats refuses it ("Dispatcher is closed"). The task reported that to the exception handler as "Error during processing", which logs a WARN for a route that is only being stopped. When the consumer is no longer allowed to run, the task now logs it at DEBUG instead; otherwise it is still reported as before. New test stopWhileSubscribingIsNotReportedAsAnError: the consumer stops from inside the subscribe of the fake server; nothing must reach the exception handler. Without the change it fails with "reported: [java.lang.IllegalStateException: Dispatcher is closed]". Co-Authored-By: Claude Opus 5.5 --- .../camel/component/nats/NatsConsumer.java | 9 +++- .../NatsConsumerJetStreamPushStopTest.java | 51 +++++++++++++++++++ 2 files changed, 59 insertions(+), 1 deletion(-) 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 3546af5248643..88b89b98a638a 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 @@ -246,7 +246,14 @@ public void run() { setupStandardNatsConsumer(topic, queueName, maxMessages); } } catch (final Exception e) { - NatsConsumer.this.getExceptionHandler().handleException("Error during processing", e); + if (!NatsConsumer.this.isRunAllowed()) { + // the consumer stopped meanwhile (its stop closes the dispatcher that a JetStream push subscribe + // uses, which then fails): nothing to report + LOG.debug("Not consuming from topic {} as the consumer is stopping", + getEndpoint().getConfiguration().getTopic(), e); + } else { + NatsConsumer.this.getExceptionHandler().handleException("Error during processing", e); + } } } 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 a847ab7c856a9..ea1f942f82b49 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 @@ -18,6 +18,7 @@ import java.lang.reflect.Proxy; import java.nio.charset.StandardCharsets; +import java.time.Duration; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; @@ -25,6 +26,7 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -43,8 +45,11 @@ import io.nats.client.impl.NatsJetStreamMetaData; import io.nats.client.impl.NatsMessage; import org.apache.camel.Consumer; +import org.apache.camel.Exchange; import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.spi.ExceptionHandler; +import org.apache.camel.support.DefaultConsumer; import org.apache.camel.support.service.ServiceHelper; import org.apache.camel.test.junit6.CamelTestSupport; import org.junit.jupiter.api.Test; @@ -171,6 +176,40 @@ void eachRunOfTheConsumerClosesItsOwnDispatcher() throws Exception { assertFalse(slowInterrupted.get(), "the step in flight must not be interrupted by the stops"); } + @Test + void stopWhileSubscribingIsNotReportedAsAnError() throws Exception { + DefaultConsumer consumer = (DefaultConsumer) context.getRoute("orders").getConsumer(); + ServiceHelper.stopService(consumer); + List 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); + } + }); + + // the consumer stops while its task subscribes: the stop closes the dispatcher that the task passes to the + // subscribe, which then fails with "Dispatcher is closed" + server.beforeSubscribe = () -> ServiceHelper.stopService(consumer); + ServiceHelper.startService(consumer); + await().atMost(10, TimeUnit.SECONDS).until(() -> server.refusedSubscribes() == 1); + + // the default exception handler would log it at WARN, but a stop is not an error + await().during(Duration.ofMillis(500)).atMost(5, TimeUnit.SECONDS) + .untilAsserted(() -> assertTrue(reported.isEmpty(), () -> "reported: " + reported)); + assertEquals(0, server.activeSubscriptions(), "no subscription left once the consumer stopped"); + } + /** * 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 @@ -185,6 +224,9 @@ static final class FakeServer { final List subscriptions = new ArrayList<>(); final Map bySubscription = new IdentityHashMap<>(); final Set closed = Collections.newSetFromMap(new IdentityHashMap<>()); + // runs on the thread of the consumer, before a push subscribe reaches the server + volatile Runnable beforeSubscribe; + int refusedSubscribes; final class FakeSubscription { final Dispatcher dispatcher; @@ -204,6 +246,10 @@ synchronized void publish(String body) { notifyAll(); } + synchronized int refusedSubscribes() { + return refusedSubscribes; + } + synchronized int closedDispatchers() { return closed.size(); } @@ -216,6 +262,10 @@ Connection connection() { JetStreamManagement jsm = proxy(JetStreamManagement.class, (name, args) -> null); JetStream js = proxy(JetStream.class, (name, args) -> { if ("subscribe".equals(name) && args.length == 6 && args[5] instanceof PushSubscribeOptions o) { + Runnable hook = beforeSubscribe; + if (hook != null) { + hook.run(); + } return subscribe(o.getConsumerConfiguration(), (Dispatcher) args[2], (MessageHandler) args[3]); } throw new UnsupportedOperationException(name); @@ -277,6 +327,7 @@ private synchronized void closeDispatcher(Dispatcher d) { private synchronized JetStreamSubscription subscribe( ConsumerConfiguration cc, Dispatcher dispatcher, MessageHandler handler) { if (closed.contains(dispatcher)) { + refusedSubscribes++; throw new IllegalStateException("Dispatcher is closed"); } FakeSubscription f = new FakeSubscription(dispatcher); From 799c2f9629985527cd7b42f0814205faf4f5f52e Mon Sep 17 00:00:00 2001 From: Claus Ibsen Date: Sat, 24 Oct 2026 15:52:27 +0200 Subject: [PATCH 4/4] CAMEL-25847: camel-nats - note why the dispatcher closes outside the lock Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Claus Ibsen --- .../main/java/org/apache/camel/component/nats/NatsConsumer.java | 1 + 1 file changed, 1 insertion(+) 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 88b89b98a638a..7a31579e1d618 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 @@ -625,6 +625,7 @@ void closeWhenIdle(Runnable close) { return; } } + // no message in flight: close now (outside the lock, as the close may block) close.run(); }