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 6bc46f9cffc1 CAMEL-26884: camel-nats - the stop of a consumer or
producer goes on when the flush fails, and a consumer ends a pull subscription
made while it stops (#29788)
6bc46f9cffc1 is described below
commit 6bc46f9cffc112cf25030101ca1e895b5267dc6a
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Nov 13 20:23:14 2026 +0530
CAMEL-26884: camel-nats - the stop of a consumer or producer goes on when
the flush fails, and a consumer ends a pull subscription made while it stops
(#29788)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/component/nats/NatsConsumer.java | 94 +++++--
.../apache/camel/component/nats/NatsProducer.java | 24 +-
.../component/nats/NatsConsumerStopGapsTest.java | 309 +++++++++++++++++++++
3 files changed, 403 insertions(+), 24 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 e5e0d7343996..a346e298b56a 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
@@ -113,6 +113,9 @@ public class NatsConsumer extends DefaultConsumer {
@Override
protected void doStop() throws Exception {
final NatsConfiguration configuration =
this.getEndpoint().getConfiguration();
+ // an interrupt during the flush is kept until the end: jnats drops a
message to the server (an UNSUB) that is
+ // queued on an interrupted thread
+ boolean interrupted = false;
if (this.pushConsumerKeeper != null) {
this.pushConsumerKeeper.stop();
@@ -121,7 +124,19 @@ public class NatsConsumer extends DefaultConsumer {
if (configuration.isFlushConnection() &&
ObjectHelper.isNotEmpty(this.connection)) {
LOG.debug("Flushing Messages before stopping");
-
this.connection.flush(Duration.ofMillis(configuration.getFlushTimeout()));
+ try {
+
this.connection.flush(Duration.ofMillis(configuration.getFlushTimeout()));
+ } catch (InterruptedException e) {
+ interrupted = true;
+ LOG.warn("Interrupted while flushing the NATS connection
before stopping the consumer of topic {}",
+ configuration.getTopic());
+ } catch (Exception e) {
+ // best effort (jnats fails the flush of a connection that is
not connected): the consumer must still
+ // end its subscriptions, its executor and its own connection
+ LOG.warn("Cannot flush the NATS connection before stopping the
consumer of topic {} due to: {}",
+ configuration.getTopic(), e.toString());
+ LOG.debug("Flush failure", e);
+ }
}
final Dispatcher stoppedDispatcher = this.dispatcher;
@@ -159,28 +174,40 @@ public class NatsConsumer extends DefaultConsumer {
this.dispatcherHandler = null;
}
- if (ObjectHelper.isNotEmpty(this.jetStreamSubscription) &&
this.jetStreamSubscription.isActive()) {
+ final JetStreamSubscription stoppedSubscription =
this.jetStreamSubscription;
+ if (ObjectHelper.isNotEmpty(stoppedSubscription) &&
stoppedSubscription.isActive()) {
try {
- this.jetStreamSubscription.unsubscribe();
+ stoppedSubscription.unsubscribe();
} catch (final Exception e) {
- this.getExceptionHandler().handleException("Error during
unsubscribing", e);
+ if (stoppedSubscription.isActive()) {
+ this.getExceptionHandler().handleException("Error during
unsubscribing", e);
+ } else {
+ // the consuming task ended it meanwhile, as it saw the
stop after its subscribe
+ LOG.debug("Not unsubscribing the subscription of the
stopped consumer, as it is ended already", e);
+ }
}
}
LOG.debug("Stopping Nats Consumer");
- if (this.executor != null) {
- if (this.getEndpoint() != null &&
this.getEndpoint().getCamelContext() != null) {
-
this.getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(this.executor);
- } else {
- this.executor.shutdownNow();
+ try {
+ if (this.executor != null) {
+ if (this.getEndpoint() != null &&
this.getEndpoint().getCamelContext() != null) {
+
this.getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(this.executor);
+ } else {
+ this.executor.shutdownNow();
+ }
}
- }
- this.executor = null;
+ this.executor = null;
- if (ObjectHelper.isEmpty(configuration.getConnection()) &&
ObjectHelper.isNotEmpty(this.connection)) {
- LOG.debug("Closing Nats Connection");
- if (!this.connection.getStatus().equals(Connection.Status.CLOSED))
{
- this.connection.close();
+ if (ObjectHelper.isEmpty(configuration.getConnection()) &&
ObjectHelper.isNotEmpty(this.connection)) {
+ LOG.debug("Closing Nats Connection");
+ if
(!this.connection.getStatus().equals(Connection.Status.CLOSED)) {
+ this.connection.close();
+ }
+ }
+ } finally {
+ if (interrupted) {
+ Thread.currentThread().interrupt();
}
}
super.doStop();
@@ -331,9 +358,16 @@ public class NatsConsumer extends DefaultConsumer {
configuration.getTopic(),
configuration.getAckPolicy(), configuration.getAckWait(),
configuration.getNackWait(),
configuration.getPullBatchSize(), configuration.getPullFetchTimeout());
- NatsConsumer.this.jetStreamSubscription =
this.connection.jetStream().subscribe(
+ JetStreamSubscription pullSubscription =
this.connection.jetStream().subscribe(
NatsConsumer.this.getEndpoint().getConfiguration().getTopic(),
pullOptions);
+ NatsConsumer.this.jetStreamSubscription = pullSubscription;
+ if (!NatsConsumer.this.isRunAllowed()) {
+ // stopped while subscribing: doStop may have run before
the subscription was set, and then did
+ // not end it
+ unsubscribeStopped(pullSubscription);
+ return;
+ }
NatsConsumer.this.setActive(true);
runPullFetchLoop(NatsConsumer.this.jetStreamSubscription,
messageHandler, cc);
@@ -385,6 +419,25 @@ public class NatsConsumer extends DefaultConsumer {
}
}
+ private void unsubscribeStopped(JetStreamSubscription subscription) {
+ // the stop interrupts this thread, and jnats drops a message to
the server that is queued on an interrupted
+ // thread: clear the interrupt for the unsubscribe, and keep it
afterwards
+ boolean interrupted = Thread.interrupted();
+ try {
+ if (subscription.isActive()) {
+ subscription.unsubscribe();
+ }
+ } catch (Exception e) {
+ // doStop may have ended it meanwhile
+ LOG.debug("Cannot unsubscribe the pull subscription of the
stopped consumer of topic {}",
+ configuration.getTopic(), e);
+ } finally {
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
private void runPullFetchLoop(
JetStreamSubscription subscription, CamelNatsMessageHandler
messageHandler, ConsumerConfiguration cc) {
JetStreamSubscription sub = subscription;
@@ -576,12 +629,9 @@ public class NatsConsumer extends DefaultConsumer {
unacknowledged.clear();
NatsConsumer.this.jetStreamSubscription = sub;
if (!NatsConsumer.this.isRunAllowed()) {
- // stopping: doStop may have unsubscribed the old subscription
before this one was assigned
- try {
- sub.unsubscribe();
- } catch (Exception e) {
- LOG.debug("Error unsubscribing the new pull subscription
while stopping", e);
- }
+ // stopping: doStop may have unsubscribed the old subscription
before this one was assigned, and the
+ // stop interrupts this thread
+ unsubscribeStopped(sub);
}
try {
old.unsubscribe();
diff --git
a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsProducer.java
b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsProducer.java
index 3c308a7035be..ee4a2493e6fe 100644
---
a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsProducer.java
+++
b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsProducer.java
@@ -191,11 +191,31 @@ public class NatsProducer extends DefaultAsyncProducer {
if
(ObjectHelper.isEmpty(this.getEndpoint().getConfiguration().getConnection())) {
LOG.debug("Closing Nats Connection");
if (this.connection != null &&
!this.connection.getStatus().equals(Status.CLOSED)) {
+ // an interrupt during the flush is kept until the connection
is closed
+ boolean interrupted = false;
if (this.getEndpoint().getConfiguration().isFlushConnection())
{
LOG.debug("Flushing Nats Connection");
-
this.connection.flush(Duration.ofMillis(this.getEndpoint().getConfiguration().getFlushTimeout()));
+ try {
+
this.connection.flush(Duration.ofMillis(this.getEndpoint().getConfiguration().getFlushTimeout()));
+ } catch (InterruptedException e) {
+ interrupted = true;
+ LOG.warn("Interrupted while flushing the NATS
connection before stopping the producer of topic {}",
+
this.getEndpoint().getConfiguration().getTopic());
+ } catch (Exception e) {
+ // best effort (jnats fails the flush of a connection
that is not connected): the connection
+ // of the producer must still be closed
+ LOG.warn("Cannot flush the NATS connection before
stopping the producer of topic {} due to: {}",
+
this.getEndpoint().getConfiguration().getTopic(), e.toString());
+ LOG.debug("Flush failure", e);
+ }
+ }
+ try {
+ this.connection.close();
+ } finally {
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
}
- this.connection.close();
}
}
super.doStop();
diff --git
a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerStopGapsTest.java
b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerStopGapsTest.java
new file mode 100644
index 000000000000..69399ee86c04
--- /dev/null
+++
b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerStopGapsTest.java
@@ -0,0 +1,309 @@
+/*
+ * 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.time.ZonedDateTime;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.nats.client.Connection;
+import io.nats.client.JetStream;
+import io.nats.client.JetStreamManagement;
+import io.nats.client.JetStreamStatusException;
+import io.nats.client.JetStreamSubscription;
+import io.nats.client.PullSubscribeOptions;
+import io.nats.client.api.ServerInfo;
+import io.nats.client.api.StreamInfo;
+import io.nats.client.support.DateTimeUtils;
+import io.nats.client.support.JsonParser;
+import io.nats.client.support.Status;
+import org.apache.camel.Producer;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static
org.apache.camel.component.nats.NatsConsumerPullConsumerDeletedTest.FakeServer.proxy;
+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.assertTrue;
+
+/**
+ * The stop of a NATS consumer when the flush of the connection fails, and
when the consumer stops while its JetStream
+ * pull subscribe, or its subscribe again after the server deleted the
consumer, is in flight.
+ */
+class NatsConsumerStopGapsTest extends CamelTestSupport {
+
+ private final NatsConsumerJetStreamPushStopTest.FakeServer server = new
NatsConsumerJetStreamPushStopTest.FakeServer();
+
+ // the pull subscribe of the second test
+ private final CountDownLatch subscribing = new CountDownLatch(1);
+ private final CountDownLatch subscribeDone = new CountDownLatch(1);
+ private final AtomicInteger pullUnsubscribes = new AtomicInteger();
+ private final AtomicBoolean pullActive = new AtomicBoolean(true);
+ private final AtomicBoolean unsubscribedOnInterruptedThread = new
AtomicBoolean();
+
+ // the pull subscribe again of the third test
+ private final CountDownLatch resubscribing = new CountDownLatch(1);
+ private final CountDownLatch resubscribeDone = new CountDownLatch(1);
+ private final AtomicInteger newPullUnsubscribes = new AtomicInteger();
+ private final AtomicBoolean newPullActive = new AtomicBoolean(true);
+ private final AtomicBoolean newUnsubscribedOnInterruptedThread = new
AtomicBoolean();
+
+ NatsConsumerStopGapsTest() {
+ testConfiguration().withUseRouteBuilder(false);
+ }
+
+ @Test
+ void stopWhenTheFlushFailsEndsTheSubscription() throws Exception {
+ context.getRegistry().bind("conn", server.connection());
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("nats:orders?connection=#conn&jetstreamEnabled=true&jetstreamName=ORDERS&pullSubscription=false")
+ .routeId("orders").to("mock:result");
+ }
+ });
+ context.start();
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
server.activeSubscriptions() == 1);
+
+ // jnats fails the flush of a connection that is not connected (the
server is down)
+ server.onFlush = () -> sneakyThrow(new TimeoutException("Timeout out
waiting for connection before flush."));
+
+ assertDoesNotThrow(() ->
context.getRouteController().stopRoute("orders"));
+
+ // the stop still ends the subscription and closes the dispatcher
+ assertEquals(0, server.activeSubscriptions());
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
server.closedDispatchers() == 1);
+
assertTrue(context.getRouteController().getRouteStatus("orders").isStopped());
+ }
+
+ @Test
+ void pullSubscriptionMadeWhileStoppingIsEnded() throws Exception {
+ context.getRegistry().bind("conn", pullConnection());
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("nats:orders?connection=#conn&jetstreamEnabled=true&jetstreamName=ORDERS&flushConnection=false")
+ .routeId("orders").to("mock:result");
+ }
+ });
+ context.start();
+ assertTrue(subscribing.await(10, TimeUnit.SECONDS));
+
+ // the subscribe completes just before the stop interrupts the task,
and the task stores the subscription after
+ // doStop looked for it
+ context.getRouteController().stopRoute("orders");
+ subscribeDone.countDown();
+
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() ->
assertEquals(1, pullUnsubscribes.get(),
+ "the subscription made for the stopped consumer must be
ended"));
+ assertFalse(unsubscribedOnInterruptedThread.get(), "the unsubscribe
must not run on an interrupted thread");
+ }
+
+ @Test
+ void pullSubscriptionMadeAgainWhileStoppingIsEnded() throws Exception {
+ context.getRegistry().bind("conn", resubscribeConnection());
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("nats:orders?connection=#conn&jetstreamEnabled=true&jetstreamName=ORDERS&flushConnection=false")
+ .routeId("orders").to("mock:result");
+ }
+ });
+ context.start();
+ // the first fetch tells that the server deleted the ephemeral
consumer, and the task subscribes again
+ assertTrue(resubscribing.await(10, TimeUnit.SECONDS));
+
+ // the subscribe again completes just before the stop interrupts the
task, and the task stores the new
+ // subscription after doStop looked for it
+ context.getRouteController().stopRoute("orders");
+ resubscribeDone.countDown();
+
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() ->
assertEquals(1, newPullUnsubscribes.get(),
+ "the subscription made again for the stopped consumer must be
ended"));
+ assertFalse(newUnsubscribedOnInterruptedThread.get(), "the unsubscribe
must not run on an interrupted thread");
+ }
+
+ @Test
+ void producerStopWhenTheFlushFailsClosesItsConnection() throws Exception {
+ AtomicInteger closes = new AtomicInteger();
+ Connection connection = proxy(Connection.class, (name, args) -> switch
(name) {
+ case "flush" -> throw new TimeoutException("Timeout out waiting
for connection before flush.");
+ case "close" -> {
+ closes.incrementAndGet();
+ yield null;
+ }
+ case "getStatus" -> Connection.Status.DISCONNECTED;
+ default -> null;
+ });
+ context.start();
+ NatsComponent component = context.getComponent("nats",
NatsComponent.class);
+ NatsConfiguration configuration = new NatsConfiguration();
+ configuration.setTopic("orders");
+ // the producer creates its own connection (no connection option),
which the stop must close
+ NatsEndpoint endpoint = new NatsEndpoint("nats:orders", component,
configuration) {
+ @Override
+ public Connection getConnection() {
+ return connection;
+ }
+ };
+ endpoint.setCamelContext(context);
+ Producer producer = endpoint.createProducer();
+ producer.start();
+
+ assertDoesNotThrow(producer::stop);
+
+ assertEquals(1, closes.get(), "the connection of the producer must be
closed");
+ assertTrue(((NatsProducer) producer).isStopped());
+ }
+
+ private Connection pullConnection() {
+ StreamInfo streamInfo = new StreamInfo(
+
JsonParser.parseUnchecked("{\"config\":{\"name\":\"ORDERS\",\"subjects\":[\"orders\"]},\"created\":\""
+ +
DateTimeUtils.toRfc3339(ZonedDateTime.now()) + "\",\"state\":{}}"));
+ JetStreamManagement jsm = proxy(JetStreamManagement.class, (name,
args) -> switch (name) {
+ case "getStreamInfo", "addStream" -> streamInfo;
+ default -> null;
+ });
+ JetStreamSubscription subscription =
proxy(JetStreamSubscription.class, (name, args) -> switch (name) {
+ case "isActive" -> pullActive.get();
+ case "unsubscribe" -> {
+ // jnats drops the UNSUB that it queues on an interrupted
thread
+ unsubscribedOnInterruptedThread.compareAndSet(false,
Thread.currentThread().isInterrupted());
+ pullUnsubscribes.incrementAndGet();
+ pullActive.set(false);
+ yield null;
+ }
+ case "fetch" -> List.of();
+ default -> null;
+ });
+ JetStream js = proxy(JetStream.class, (name, args) -> {
+ if ("subscribe".equals(name) && args.length == 2 && args[1]
instanceof PullSubscribeOptions) {
+ subscribing.countDown();
+ // the subscribe completed before the interrupt of the stop
reached it: the task gets the subscription
+ // after doStop looked for it
+ boolean interrupted = false;
+ while (true) {
+ try {
+ assertTrue(subscribeDone.await(20, TimeUnit.SECONDS));
+ break;
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ return subscription;
+ }
+ 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 "getStatus" -> Connection.Status.CONNECTED;
+ default -> null;
+ });
+ }
+
+ private Connection resubscribeConnection() {
+ StreamInfo streamInfo = new StreamInfo(
+
JsonParser.parseUnchecked("{\"config\":{\"name\":\"ORDERS\",\"subjects\":[\"orders\"]},\"created\":\""
+ +
DateTimeUtils.toRfc3339(ZonedDateTime.now()) + "\",\"state\":{}}"));
+ JetStreamManagement jsm = proxy(JetStreamManagement.class, (name,
args) -> switch (name) {
+ case "getStreamInfo", "addStream" -> streamInfo;
+ default -> null;
+ });
+ // the subscription of the first subscribe, whose consumer the server
deleted
+ AtomicBoolean firstActive = new AtomicBoolean(true);
+ AtomicBoolean toldDeleted = new AtomicBoolean();
+ JetStreamSubscription[] first = new JetStreamSubscription[1];
+ first[0] = proxy(JetStreamSubscription.class, (name, args) -> switch
(name) {
+ case "isActive" -> firstActive.get();
+ case "unsubscribe" -> {
+ firstActive.set(false);
+ yield null;
+ }
+ case "fetch" -> {
+ if (toldDeleted.compareAndSet(false, true)) {
+ throw new JetStreamStatusException(new Status(409,
Status.CONSUMER_DELETED), first[0]);
+ }
+ yield List.of();
+ }
+ default -> null;
+ });
+ // the subscription made again
+ JetStreamSubscription again = proxy(JetStreamSubscription.class,
(name, args) -> switch (name) {
+ case "isActive" -> newPullActive.get();
+ case "unsubscribe" -> {
+ // jnats drops the UNSUB that it queues on an interrupted
thread
+ newUnsubscribedOnInterruptedThread.compareAndSet(false,
Thread.currentThread().isInterrupted());
+ newPullUnsubscribes.incrementAndGet();
+ newPullActive.set(false);
+ yield null;
+ }
+ case "fetch" -> List.of();
+ default -> null;
+ });
+ AtomicInteger subscribes = new AtomicInteger();
+ JetStream js = proxy(JetStream.class, (name, args) -> {
+ if ("subscribe".equals(name) && args.length == 2 && args[1]
instanceof PullSubscribeOptions) {
+ if (subscribes.incrementAndGet() == 1) {
+ return first[0];
+ }
+ resubscribing.countDown();
+ // the subscribe again completed before the interrupt of the
stop reached it: the task gets the new
+ // subscription after doStop looked for it
+ boolean interrupted = false;
+ while (true) {
+ try {
+ assertTrue(resubscribeDone.await(20,
TimeUnit.SECONDS));
+ break;
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ return again;
+ }
+ 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 "getStatus" -> Connection.Status.CONNECTED;
+ default -> null;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <E extends Throwable> void sneakyThrow(Throwable e) throws
E {
+ throw (E) e;
+ }
+}
