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 9aee81df989a CAMEL-25815: camel-nats - the JetStream pull consumer
subscribes again when the server deleted its consumer (#28204)
9aee81df989a is described below
commit 9aee81df989a1d96790204b21baef703662cd192
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Oct 23 22:03:07 2026 +0530
CAMEL-25815: camel-nats - the JetStream pull consumer subscribes again when
the server deleted its consumer (#28204)
* CAMEL-25815: camel-nats - the JetStream pull consumer subscribes again
when the server deleted its consumer
With jetstreamEnabled=true and the default pull subscription, and without a
durableName, the consumer subscribes with an ephemeral JetStream consumer.
The NATS server deletes an ephemeral consumer that had no waiting pull
request for its inactive threshold (5 seconds by default) once no ack is
pending, for example while the client was disconnected for a few seconds,
and a server restart loses it. jnats does not give a pull subscription a new
consumer: every fetch then ends with an empty list (the pull request has no
responder, status 503), or once with 409 Consumer Deleted, which the pull
loop
logged and continued. The route stayed started and consumed nothing again
until it was restarted.
The pull loop now asks the server for its consumer when a fetch returned
nothing (at most every 5 seconds) and handles 409 Consumer Deleted. When the
consumer is gone it subscribes again: an ephemeral consumer is created again
starting at the first stream sequence the route did not acknowledge (or
after
the last one fetched), so that nothing is skipped and the stream is not
replayed from the deliver policy; a durable consumer is created again from
its
configuration.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25815: camel-nats - address review: subscribe again only for an
ephemeral consumer
A durable consumer is gone only when an administrator deleted it or after
its
inactiveThreshold; creating it again from its configuration replayed the
stream from its deliver policy (the whole stream with the default
DeliverPolicy.All). The pull loop now creates only an ephemeral consumer
again; for a durable it logs a WARN once and keeps the subscription, so the
fetches work again when the durable is created again with the same name.
The empty-fetch path and the 409 path share one helper, which reports a
failed subscribe (jnats throws IllegalStateException for some, such as
SUB-90007 when the stream is gone) through the exception handler and tries
again at the next consumer check, instead of ending the fetch loop. The new
consumer is sent without a start time (ByStartTime together with
ByStartSequence), a subscription made while the route stops is unsubscribed,
the unused DefaultNatsManualAck constructor is removed, and the upgrade
guide
entry covers the ephemeral consumer only.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25815: camel-nats - address review: wait for the rest of the fetch
timeout after an early empty fetch
While the consumer of a pull subscription is gone, jnats ends each fetch
at once (status 503, no responders), so the fetch loop sent pull
requests back to back: up to the next consumer check for an ephemeral
consumer, and for good for a durable that is not created again. After an
empty fetch that ended before its timeout, the loop now waits for the
rest of pullFetchTimeout; a stop interrupts the wait.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25815: camel-nats - address review: treat a null consumer info as
gone, resume only where the stream allows it
jnats 2.26.4 answers getConsumerInfo() with null when the server has no
such consumer (lookupConsumerInfo maps api error 10014 to null), so the
consumer check never found a deleted consumer: it now treats null as
gone, and the fake server answers like jnats (null for a deleted
consumer, the info for a live one).
subscribeAgain now looks the stream up first. A stream with WorkQueue
retention accepts only the deliver policy All and has already removed
the acknowledged messages, so the consumer configuration is used as it
is. A stream whose create time differs from the one seen at the
subscribe was created again (its sequences start at 1 again): the
tracked sequences are cleared and the consumer configuration is used.
The consumer check runs only after an empty fetch that ended before half
of pullFetchTimeout (a gone consumer ends the fetch at once with 503),
so an idle healthy consumer no longer asks for CONSUMER.INFO every 5
seconds. Other error statuses than 409 Consumer Deleted wait for the
rest of the fetch timeout instead of fetching again back to back. With
AckPolicy.None no sequence waits for an ack: a resubscribe starts after
the highest fetched one. The tests wait with setResultWaitTime, and a
JetStream IT deletes the consumer while the route is busy.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25815: camel-nats - address review: wait for the rest of the fetch
timeout after an unexpected fetch failure too
From gnodet-bot's review of #28204: the generic catch of runPullFetchLoop
sent the next pull request at
once after an unexpected exception (for example while the connection
recovers), unlike the error
statuses, which wait for the rest of pullFetchTimeout since the previous
commit. It now waits as well.
New test unexpectedFetchFailureIsNotFetchedAgainBackToBack: every fetch
fails with a RuntimeException.
Without the change the loop made 96635 fetches in one second.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25815: camel-nats - address review: no Thread.sleep in the
back-to-back fetch tests
Both tests now check the bound with Awaitility during the whole second.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../camel/component/nats/DefaultNatsManualAck.java | 6 +-
.../apache/camel/component/nats/NatsConsumer.java | 251 +++++++++++-
.../nats/NatsConsumerPullConsumerDeletedTest.java | 421 +++++++++++++++++++++
.../nats/NatsConsumerPullConsumerGoneTest.java | 263 +++++++++++++
.../NatsJetstreamConsumerPullDeletedIT.java | 107 ++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 14 +
6 files changed, 1046 insertions(+), 16 deletions(-)
diff --git
a/components/camel-nats/src/main/java/org/apache/camel/component/nats/DefaultNatsManualAck.java
b/components/camel-nats/src/main/java/org/apache/camel/component/nats/DefaultNatsManualAck.java
index 7d3b011663ba..9b0859919817 100644
---
a/components/camel-nats/src/main/java/org/apache/camel/component/nats/DefaultNatsManualAck.java
+++
b/components/camel-nats/src/main/java/org/apache/camel/component/nats/DefaultNatsManualAck.java
@@ -23,14 +23,17 @@ import io.nats.client.Message;
class DefaultNatsManualAck implements NatsManualAck {
private final Message message;
+ private final Runnable onAcknowledged;
- DefaultNatsManualAck(Message message) {
+ DefaultNatsManualAck(Message message, Runnable onAcknowledged) {
this.message = message;
+ this.onAcknowledged = onAcknowledged;
}
@Override
public void ack() {
message.ack();
+ onAcknowledged.run();
}
@Override
@@ -46,6 +49,7 @@ class DefaultNatsManualAck implements NatsManualAck {
@Override
public void term() {
message.term();
+ onAcknowledged.run();
}
@Override
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 af23bed3aac9..88026b0dbb2c 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
@@ -18,11 +18,15 @@ package org.apache.camel.component.nats;
import java.io.IOException;
import java.time.Duration;
+import java.time.ZonedDateTime;
import java.util.List;
+import java.util.NavigableSet;
+import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
import io.nats.client.Connection;
-import io.nats.client.Connection.Status;
import io.nats.client.Dispatcher;
import io.nats.client.JetStreamApiException;
import io.nats.client.JetStreamManagement;
@@ -34,7 +38,11 @@ import io.nats.client.PullSubscribeOptions;
import io.nats.client.PushSubscribeOptions;
import io.nats.client.api.AckPolicy;
import io.nats.client.api.ConsumerConfiguration;
+import io.nats.client.api.DeliverPolicy;
+import io.nats.client.api.RetentionPolicy;
import io.nats.client.api.StreamConfiguration;
+import io.nats.client.api.StreamInfo;
+import io.nats.client.support.Status;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.spi.HeaderFilterStrategy;
@@ -49,13 +57,15 @@ import org.slf4j.LoggerFactory;
public class NatsConsumer extends DefaultConsumer {
private static final Logger LOG =
LoggerFactory.getLogger(NatsConsumer.class);
+ // an idle JetStream pull loop asks the server for its consumer at most
this often
+ private static final long CONSUMER_CHECK_INTERVAL_NANOS = 5_000_000_000L;
private final Processor processor;
private ExecutorService executor;
private Connection connection;
private Dispatcher dispatcher;
private boolean active;
- private JetStreamSubscription jetStreamSubscription;
+ private volatile JetStreamSubscription jetStreamSubscription;
public NatsConsumer(NatsEndpoint endpoint, Processor processor) {
super(endpoint, processor);
@@ -126,7 +136,7 @@ public class NatsConsumer extends DefaultConsumer {
if (ObjectHelper.isEmpty(configuration.getConnection()) &&
ObjectHelper.isNotEmpty(this.connection)) {
LOG.debug("Closing Nats Connection");
- if (!this.connection.getStatus().equals(Status.CLOSED)) {
+ if (!this.connection.getStatus().equals(Connection.Status.CLOSED))
{
this.connection.close();
}
}
@@ -145,6 +155,18 @@ public class NatsConsumer extends DefaultConsumer {
private final Connection connection;
private final NatsConfiguration configuration;
+ // stream sequences of the JetStream messages fetched by the pull loop
and not acknowledged yet, and the highest
+ // fetched: where a new ephemeral consumer must start when the server
deleted the one this consumer used
+ private final NavigableSet<Long> unacknowledged = new
ConcurrentSkipListSet<>();
+ private final AtomicLong highestFetched = new AtomicLong();
+ // with AckPolicy.None the delivery is the acknowledgement: no fetched
sequence waits for an ack
+ private boolean trackUnacknowledged = true;
+ // when the stream was created, as seen at the subscribe: a different
time later means that the stream was
+ // created again and its sequences started again at 1
+ private ZonedDateTime streamCreated;
+ private long lastConsumerCheck = System.nanoTime();
+ // whether the WARN that the durable consumer is gone was logged, so
that it is logged once until it is back
+ private boolean durableGoneReported;
NatsConsumingTask(Connection connection, NatsConfiguration
configuration) {
this.connection = connection;
@@ -187,7 +209,8 @@ public class NatsConsumer extends DefaultConsumer {
this.configuration.getTopic());
JetStreamManagement jsm = connection.jetStreamManagement();
- if (streamExists(jsm, streamName)) {
+ StreamInfo streamInfo = lookupStream(jsm, streamName);
+ if (streamInfo != null) {
LOG.debug("JetStream stream '{}' already exists. Skipping
creation.", streamName);
} else {
LOG.debug("Creating JetStream {}", streamName);
@@ -195,9 +218,10 @@ public class NatsConsumer extends DefaultConsumer {
.name(streamName)
.subjects(topic)
.build();
- jsm.addStream(streamConfig);
+ streamInfo = jsm.addStream(streamConfig);
LOG.info("JetStream stream '{}' created successfully.",
streamName);
}
+ streamCreated = streamInfo != null ? streamInfo.getCreateTime() :
null;
ConsumerConfiguration cc =
configuration.getConsumerConfiguration();
if (cc == null) {
@@ -217,6 +241,7 @@ public class NatsConsumer extends DefaultConsumer {
}
cc = ccBuilder.build();
}
+ trackUnacknowledged = cc.getAckPolicy() != AckPolicy.None;
CamelNatsMessageHandler messageHandler = new
CamelNatsMessageHandler();
@@ -235,7 +260,7 @@ public class NatsConsumer extends DefaultConsumer {
pullOptions);
NatsConsumer.this.setActive(true);
- runPullFetchLoop(NatsConsumer.this.jetStreamSubscription,
messageHandler);
+ runPullFetchLoop(NatsConsumer.this.jetStreamSubscription,
messageHandler, cc);
} else {
NatsConsumer.this.dispatcher =
this.connection.createDispatcher(messageHandler);
@@ -262,17 +287,37 @@ public class NatsConsumer extends DefaultConsumer {
}
}
- private void runPullFetchLoop(JetStreamSubscription sub,
CamelNatsMessageHandler messageHandler) {
+ private void runPullFetchLoop(
+ JetStreamSubscription subscription, CamelNatsMessageHandler
messageHandler, ConsumerConfiguration cc) {
+ JetStreamSubscription sub = subscription;
int batchSize = configuration.getPullBatchSize();
Duration fetchTimeout =
Duration.ofMillis(configuration.getPullFetchTimeout());
while (NatsConsumer.this.isRunAllowed() &&
!Thread.currentThread().isInterrupted()) {
+ long fetchStarted = System.nanoTime();
try {
List<Message> messages = sub.fetch(batchSize,
fetchTimeout);
+ if (messages.isEmpty()) {
+ // jnats ends a fetch at once when its pull request
has no responder (status 503), for example
+ // while the consumer is gone, whereas an idle healthy
consumer ends it at its expiry, shortly
+ // before the fetch timeout: only an empty fetch that
ended early asks the server for the
+ // consumer, so that an idle healthy consumer does not
send that request every few seconds
+ boolean endedEarly = System.nanoTime() - fetchStarted
< fetchTimeout.toNanos() / 2;
+ if (endedEarly && consumerCheckDue() &&
consumerGone(sub)) {
+ sub = consumerDeleted(sub, cc);
+ continue;
+ }
+ // wait for the rest of the fetch timeout, so that the
loop does not send pull requests back to
+ // back
+ waitRestOfFetch(fetchStarted, fetchTimeout);
+ continue;
+ }
+ durableGoneReported = false;
for (Message msg : messages) {
if (!NatsConsumer.this.isRunAllowed()) {
break;
}
try {
+ track(msg);
messageHandler.onMessage(msg);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
@@ -280,17 +325,187 @@ public class NatsConsumer extends DefaultConsumer {
}
}
} catch (JetStreamStatusException e) {
- // Synchronous pull calls may throw
JetStreamStatusException for transient
- // statuses such as 409 (consumer deleted/push-based). Log
and continue so the
- // consumer can recover when the consumer state is
restored.
+ // Synchronous pull calls may throw
JetStreamStatusException for statuses such as 409. Report it and
+ // continue; for 409 Consumer Deleted an ephemeral
consumer is created again (see consumerDeleted).
NatsConsumer.this.getExceptionHandler().handleException("JetStream status
during pull fetch", e);
+ Status status = e.getStatus();
+ if (status != null &&
Status.CONSUMER_DELETED.equals(status.getMessage())) {
+ sub = consumerDeleted(sub, cc);
+ } else {
+ // for example 409 Consumer is push based: do not send
pull requests back to back
+ try {
+ waitRestOfFetch(fetchStarted, fetchTimeout);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
} catch (IllegalStateException e) {
// Subscription was closed (e.g. consumer is stopping),
exit the loop
LOG.debug("Pull subscription closed, stopping fetch loop",
e);
return;
+ } catch (InterruptedException e) {
+ // the consumer is stopping
+ Thread.currentThread().interrupt();
+ return;
} catch (Exception e) {
NatsConsumer.this.getExceptionHandler().handleException("Error during JetStream
pull fetch", e);
+ // as for the other error statuses: do not send the next
pull request at once
+ try {
+ waitRestOfFetch(fetchStarted, fetchTimeout);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+ }
+ }
+
+ private void waitRestOfFetch(long fetchStarted, Duration fetchTimeout)
throws InterruptedException {
+ long left = fetchTimeout.toNanos() - (System.nanoTime() -
fetchStarted);
+ if (left > 0) {
+ TimeUnit.NANOSECONDS.sleep(left);
+ }
+ }
+
+ /**
+ * Whether the server no longer has the consumer of this pull
subscription. The server deletes an ephemeral
+ * consumer (no durable name) that saw no pull request for its
inactive threshold (5 seconds by default), for
+ * example while the connection was down for a few seconds, and a
server restart loses it. jnats then answers
+ * every fetch with an empty list (the pull request has no responder),
and the subscription never receives
+ * anything again.
+ */
+ private boolean consumerCheckDue() {
+ long now = System.nanoTime();
+ if (now - lastConsumerCheck < CONSUMER_CHECK_INTERVAL_NANOS) {
+ return false;
+ }
+ lastConsumerCheck = now;
+ return true;
+ }
+
+ private boolean consumerGone(JetStreamSubscription sub) {
+ try {
+ // jnats answers null when the server has no such consumer
(api error 10014), and throws for any other
+ // error, for example 404 stream not found (10059)
+ if (sub.getConsumerInfo() == null) {
+ return true;
}
+ durableGoneReported = false;
+ return false;
+ } catch (JetStreamApiException e) {
+ return e.getErrorCode() == 404;
+ } catch (Exception e) {
+ // not connected, timeout: the consumer may well still be
there, the next fetch tells
+ LOG.debug("Cannot check the JetStream consumer of the pull
subscription", e);
+ return false;
+ }
+ }
+
+ /**
+ * The server no longer has the consumer of the pull subscription. An
ephemeral consumer is created again (see
+ * {@link #subscribeAgain}); when that fails, the failure goes to the
exception handler and the next consumer
+ * check tries again. A durable consumer is not created again: it is
gone only when an administrator deleted it
+ * or after its inactive threshold, and creating it again from its
configuration would replay the stream from
+ * its deliver policy. Its subscription is kept, so the fetches work
again when the durable is created again
+ * with the same name.
+ *
+ * @return the subscription to fetch from
+ */
+ private JetStreamSubscription consumerDeleted(JetStreamSubscription
sub, ConsumerConfiguration cc) {
+ if (cc.getDurable() != null) {
+ if (!durableGoneReported) {
+ durableGoneReported = true;
+ LOG.warn("The durable JetStream consumer {} of the pull
subscription on topic {} is gone; it is not"
+ + " created again, and the fetches resume when it
is created again with the same name",
+ cc.getDurable(), configuration.getTopic());
+ }
+ return sub;
+ }
+ try {
+ return subscribeAgain(sub, cc);
+ } catch (Exception e) {
+ NatsConsumer.this.getExceptionHandler().handleException(
+ "Error subscribing again after the JetStream consumer
was deleted", e);
+ return sub;
+ }
+ }
+
+ /**
+ * Subscribes again after the server deleted the ephemeral consumer,
starting at the first message this consumer
+ * has not acknowledged, so that nothing is skipped and the stream is
not replayed from its deliver policy.
+ * <p>
+ * A sequence that is never acknowledged stays in the unacknowledged
set until this resubscribe, for example a
+ * message nak'd until maxDeliver is used up. The new consumer then
starts at that old sequence, and the route
+ * sees the messages after it again (at-least-once). With
ackPolicy=None no sequence is tracked: the new
+ * consumer starts after the highest one fetched.
+ * <p>
+ * The consumer configuration is used as it is (no start sequence) for
a stream with WorkQueue retention, which
+ * accepts only the deliver policy All and has already removed the
acknowledged messages, and for a stream that
+ * was created again since the subscribe (its sequences started again
at 1).
+ */
+ private JetStreamSubscription subscribeAgain(JetStreamSubscription
old, ConsumerConfiguration cc)
+ throws IOException, JetStreamApiException {
+ StreamInfo stream =
this.connection.jetStreamManagement().getStreamInfo(configuration.getJetstreamName());
+ ZonedDateTime created = stream.getCreateTime();
+ if (created != null && streamCreated != null &&
!created.isEqual(streamCreated)) {
+ LOG.warn("The JetStream stream {} was created again since the
subscribe; consuming it from its deliver"
+ + " policy",
+ configuration.getJetstreamName());
+ unacknowledged.clear();
+ highestFetched.set(0);
+ }
+ if (created != null) {
+ streamCreated = created;
+ }
+ boolean workQueue = stream.getConfiguration() != null
+ && stream.getConfiguration().getRetentionPolicy() ==
RetentionPolicy.WorkQueue;
+ ConsumerConfiguration again = cc;
+ long start = unacknowledged.isEmpty() ? highestFetched.get() + 1 :
unacknowledged.first();
+ boolean resume = !workQueue && highestFetched.get() > 0;
+ if (resume) {
+ // a start time (DeliverPolicy.ByStartTime) together with a
start sequence is rejected by the server
+ again = ConsumerConfiguration.builder(cc)
+ .deliverPolicy(DeliverPolicy.ByStartSequence)
+ .startSequence(start)
+ .startTime(null)
+ .build();
+ }
+ LOG.warn("The JetStream consumer of the pull subscription on topic
{} is gone; subscribing again{}",
+ configuration.getTopic(), resume ? " from stream sequence
" + start : "");
+ JetStreamSubscription sub =
this.connection.jetStream().subscribe(configuration.getTopic(),
+
PullSubscribeOptions.builder().configuration(again).build());
+ 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);
+ }
+ }
+ try {
+ old.unsubscribe();
+ } catch (Exception e) {
+ LOG.debug("Error unsubscribing the pull subscription of the
deleted consumer", e);
+ }
+ return sub;
+ }
+
+ private void track(Message msg) {
+ if (msg.isJetStream()) {
+ long seq = msg.metaData().streamSequence();
+ if (trackUnacknowledged) {
+ unacknowledged.add(seq);
+ }
+ highestFetched.accumulateAndGet(seq, Math::max);
+ }
+ }
+
+ private void acknowledged(Message msg) {
+ if (msg.isJetStream()) {
+ unacknowledged.remove(msg.metaData().streamSequence());
}
}
@@ -325,6 +540,7 @@ public class NatsConsumer extends DefaultConsumer {
public void onComplete(Exchange exchange) {
LOG.debug("ACK");
msg.ack();
+ acknowledged(msg);
}
@Override
@@ -333,6 +549,7 @@ public class NatsConsumer extends DefaultConsumer {
// ACK policy is none which means that we
should auto ACK even if the message processed failed in Camel
LOG.debug("ACK");
msg.ack();
+ acknowledged(msg);
} else {
LOG.debug("NACK (delay:{})",
configuration.getNackWait());
if (configuration.getNackWait() <= 0) {
@@ -379,7 +596,8 @@ public class NatsConsumer extends DefaultConsumer {
evalRedeliveryMessageTrait(msg, exchange));
if (manualAckEnabled) {
-
exchange.getIn().setHeader(NatsConstants.NATS_MANUAL_ACK, new
DefaultNatsManualAck(msg));
+
exchange.getIn().setHeader(NatsConstants.NATS_MANUAL_ACK,
+ new DefaultNatsManualAck(msg, () ->
acknowledged(msg)));
}
NatsConsumer.this.processor.process(exchange);
@@ -407,13 +625,16 @@ public class NatsConsumer extends DefaultConsumer {
}
}
- private boolean streamExists(JetStreamManagement jsm, String streamName)
throws IOException, JetStreamApiException {
+ /**
+ * @return the stream, or null when the server does not have it
+ */
+ private static StreamInfo lookupStream(JetStreamManagement jsm, String
streamName)
+ throws IOException, JetStreamApiException {
try {
- jsm.getStreamInfo(streamName);
- return true;
+ return jsm.getStreamInfo(streamName);
} catch (JetStreamApiException jsae) {
if (jsae.getErrorCode() == 404) {
- return false;
+ return null;
}
throw jsae;
}
diff --git
a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java
b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java
new file mode 100644
index 000000000000..ce09e17c508b
--- /dev/null
+++
b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java
@@ -0,0 +1,421 @@
+/*
+ * 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.time.Duration;
+import java.time.ZonedDateTime;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import io.nats.client.Connection;
+import io.nats.client.JetStream;
+import io.nats.client.JetStreamApiException;
+import io.nats.client.JetStreamManagement;
+import io.nats.client.JetStreamStatusException;
+import io.nats.client.JetStreamSubscription;
+import io.nats.client.Message;
+import io.nats.client.PullSubscribeOptions;
+import io.nats.client.api.ApiResponse;
+import io.nats.client.api.ConsumerConfiguration;
+import io.nats.client.api.ConsumerInfo;
+import io.nats.client.api.DeliverPolicy;
+import io.nats.client.api.RetentionPolicy;
+import io.nats.client.api.ServerInfo;
+import io.nats.client.api.StreamInfo;
+import io.nats.client.impl.NatsJetStreamMetaData;
+import io.nats.client.impl.NatsMessage;
+import io.nats.client.support.DateTimeUtils;
+import io.nats.client.support.JsonParser;
+import io.nats.client.support.Status;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * The JetStream pull consumer (the default with jetstreamEnabled=true) uses
an ephemeral consumer when no durableName
+ * is set. The NATS server deletes an ephemeral consumer that saw no pull
request for its inactive threshold (5 seconds
+ * by default), for example while the client was disconnected for a few
seconds, and a server restart loses it. The
+ * consumer must then subscribe again, starting at the first message it did
not acknowledge, instead of fetching from
+ * the deleted consumer for ever.
+ */
+class NatsConsumerPullConsumerDeletedTest extends CamelTestSupport {
+
+ private final FakeServer server = new FakeServer();
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ context.getRegistry().bind("conn", server.connection());
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("nats:orders?connection=#conn&jetstreamEnabled=true&jetstreamName=ORDERS&pullFetchTimeout=100")
+ .convertBodyTo(String.class)
+ .to("mock:result");
+ }
+ };
+ }
+
+ @Test
+ void consumesAgainAfterTheServerDeletedTheEphemeralConsumer() throws
Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("order-1");
+ server.publish("order-1");
+ mock.assertIsSatisfied();
+
+ // the client was disconnected for longer than the inactive threshold:
the server deleted the consumer
+ server.deleteConsumers(false);
+
+ mock.reset();
+ mock.expectedBodiesReceived("order-2", "order-3");
+ server.publish("order-2");
+ server.publish("order-3");
+ mock.setResultWaitTime(20_000);
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void consumesAgainAfterAWaitingPullGotConsumerDeleted() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("order-1");
+ server.publish("order-1");
+ mock.assertIsSatisfied();
+
+ // the consumer is deleted while a pull request waits: that request
gets 409 Consumer Deleted
+ server.deleteConsumers(true);
+
+ mock.reset();
+ mock.expectedBodiesReceived("order-2");
+ server.publish("order-2");
+ mock.setResultWaitTime(20_000);
+ mock.assertIsSatisfied();
+ assertEquals(2, server.consumersCreated(), "one consumer created
again");
+ }
+
+ @Test
+ void consumesAStreamCreatedAgainFromItsStart() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("order-1", "order-2");
+ server.publish("order-1");
+ server.publish("order-2");
+ mock.assertIsSatisfied();
+
+ // a memory stream after a server restart: the consumer is gone and
order-3 is sequence 1 of the new stream
+ server.createStreamAgain();
+
+ mock.reset();
+ mock.expectedBodiesReceived("order-3");
+ server.publish("order-3");
+ mock.setResultWaitTime(20_000);
+ mock.assertIsSatisfied();
+ ConsumerConfiguration again = server.subscribed().get(1);
+ assertEquals(DeliverPolicy.All, again.getDeliverPolicy(), "the
consumer configuration, not a start sequence");
+ }
+
+ @Test
+ void workQueueStreamIsSubscribedAgainWithDeliverPolicyAll() throws
Exception {
+ server.retention(RetentionPolicy.WorkQueue);
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("order-1");
+ server.publish("order-1");
+ mock.assertIsSatisfied();
+
+ server.deleteConsumers(false);
+
+ mock.reset();
+ mock.expectedBodiesReceived("order-2");
+ server.publish("order-2");
+ mock.setResultWaitTime(20_000);
+ mock.assertIsSatisfied();
+ assertEquals(2, server.consumersCreated(), "one consumer created
again");
+ assertEquals(DeliverPolicy.All,
server.subscribed().get(1).getDeliverPolicy(),
+ "a work queue stream accepts only the deliver policy All");
+ }
+
+ /**
+ * A JetStream server with one stream, faked at the level of the jnats
interfaces the consumer calls. A consumer
+ * delivers from its start sequence; once deleted, a fetch on it gets
nothing at once (jnats ends the fetch on the
+ * 503 "no responders" status), and getConsumerInfo answers null, as jnats
does when the server answers consumer not
+ * found (10014). A live consumer answers its info. A stream with
WorkQueue retention removes the acknowledged
+ * messages and refuses a consumer whose deliver policy is not All (400,
10101), and a stream created again starts
+ * its sequences at 1 again with a new create time.
+ */
+ static final class FakeServer {
+ final List<byte[]> stream = new ArrayList<>();
+ final Set<Long> removed = new HashSet<>();
+ final List<FakeConsumer> consumers = new ArrayList<>();
+ final List<ConsumerConfiguration> subscribed = new ArrayList<>();
+ RetentionPolicy retention = RetentionPolicy.Limits;
+ ZonedDateTime created = DateTimeUtils.gmtNow().minusHours(1);
+ int consumersCreated;
+ int fetches;
+ int consumerInfoRequests;
+ RuntimeException failNextSubscribe;
+ Status fetchStatus;
+ RuntimeException fetchFailure;
+
+ final class FakeConsumer {
+ long next;
+ boolean deleted;
+ boolean tellDeleted;
+ final List<Long> redeliver = new ArrayList<>();
+ }
+
+ synchronized int consumersCreated() {
+ return consumersCreated;
+ }
+
+ synchronized int fetches() {
+ return fetches;
+ }
+
+ synchronized int consumerInfoRequests() {
+ return consumerInfoRequests;
+ }
+
+ synchronized void retention(RetentionPolicy retention) {
+ this.retention = retention;
+ }
+
+ /**
+ * Every fetch fails with this status, for example 409 Consumer is
push based.
+ */
+ synchronized void failFetches(Status status) {
+ fetchStatus = status;
+ }
+
+ /**
+ * Every fetch fails with this exception, for example an error of the
connection while it recovers.
+ */
+ synchronized void failFetches(RuntimeException failure) {
+ fetchFailure = failure;
+ }
+
+ /**
+ * The stream is created again, for example a memory stream after a
server restart: its consumers are gone and
+ * its sequences start again at 1.
+ */
+ synchronized void createStreamAgain() {
+ stream.clear();
+ removed.clear();
+ created = created.plusMinutes(1);
+ deleteConsumers(false);
+ }
+
+ synchronized void publish(String body) {
+ stream.add(body.getBytes(StandardCharsets.UTF_8));
+ notifyAll();
+ }
+
+ synchronized void deleteConsumers(boolean waitingRequest) {
+ for (FakeConsumer c : consumers) {
+ c.deleted = true;
+ c.tellDeleted = waitingRequest;
+ }
+ notifyAll();
+ }
+
+ /**
+ * The deleted consumers are created again with the same name (a
durable created again by an administrator).
+ */
+ synchronized void restoreConsumers() {
+ for (FakeConsumer c : consumers) {
+ c.deleted = false;
+ }
+ notifyAll();
+ }
+
+ synchronized void failNextSubscribe(RuntimeException e) {
+ failNextSubscribe = e;
+ }
+
+ synchronized List<ConsumerConfiguration> subscribed() {
+ return new ArrayList<>(subscribed);
+ }
+
+ Connection connection() {
+ JetStreamManagement jsm = proxy(JetStreamManagement.class, (name,
args) -> switch (name) {
+ case "getStreamInfo", "addStream" -> streamInfo();
+ default -> null;
+ });
+ JetStream js = proxy(JetStream.class, (name, args) -> {
+ if ("subscribe".equals(name) && args.length == 2 && args[1]
instanceof PullSubscribeOptions o) {
+ return subscribe(o.getConsumerConfiguration());
+ }
+ 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 synchronized StreamInfo streamInfo() {
+ return new StreamInfo(
+
JsonParser.parseUnchecked("{\"config\":{\"name\":\"ORDERS\",\"subjects\":[\"orders\"],"
+ + "\"retention\":\"" + retention
+ "\"},\"created\":\""
+ +
DateTimeUtils.toRfc3339(created) + "\",\"state\":{}}"));
+ }
+
+ private synchronized JetStreamSubscription
subscribe(ConsumerConfiguration cc) throws JetStreamApiException {
+ if (failNextSubscribe != null) {
+ RuntimeException e = failNextSubscribe;
+ failNextSubscribe = null;
+ throw e;
+ }
+ if (retention == RetentionPolicy.WorkQueue &&
cc.getDeliverPolicy() != DeliverPolicy.All) {
+ // the server's answer as jnats reads it (ApiResponse is
abstract)
+ throw new JetStreamApiException(
+ new ApiResponse<Void>(
+ JsonParser.parseUnchecked(
+
"{\"error\":{\"code\":400,\"err_code\":10101,"
+ +
"\"description\":\"consumer must be deliver all on workqueue stream\"}}")) {
+ });
+ }
+ subscribed.add(cc);
+ FakeConsumer c = new FakeConsumer();
+ c.next = cc != null && cc.getDeliverPolicy() ==
DeliverPolicy.ByStartSequence ? cc.getStartSequence() : 1;
+ consumers.add(c);
+ consumersCreated++;
+ JetStreamSubscription[] self = new JetStreamSubscription[1];
+ self[0] = proxy(JetStreamSubscription.class, (name, args) ->
switch (name) {
+ case "fetch" -> fetch(c, self[0], (Integer) args[0],
(Duration) args[1]);
+ case "getConsumerInfo" -> consumerInfo(c);
+ case "isActive" -> true;
+ default -> null;
+ });
+ return self[0];
+ }
+
+ private synchronized ConsumerInfo consumerInfo(FakeConsumer c) {
+ consumerInfoRequests++;
+ if (c.deleted) {
+ return null;
+ }
+ return new
ConsumerInfo(JsonParser.parseUnchecked("{\"stream_name\":\"ORDERS\",\"name\":\"c\",\"config\":{}}"));
+ }
+
+ private synchronized List<Message> fetch(FakeConsumer c,
JetStreamSubscription sub, int batch, Duration timeout) {
+ fetches++;
+ long deadline = System.nanoTime() + timeout.toNanos();
+ List<Message> out = new ArrayList<>();
+ if (fetchStatus != null) {
+ throw new JetStreamStatusException(fetchStatus, sub);
+ }
+ if (fetchFailure != null) {
+ throw fetchFailure;
+ }
+ while (true) {
+ if (c.deleted) {
+ if (c.tellDeleted) {
+ c.tellDeleted = false;
+ throw new JetStreamStatusException(new Status(409,
Status.CONSUMER_DELETED), sub);
+ }
+ return out;
+ }
+ while (out.size() < batch && (!c.redeliver.isEmpty() || c.next
<= stream.size())) {
+ long seq = c.redeliver.isEmpty() ? c.next++ :
c.redeliver.remove(0);
+ if (!removed.contains(seq)) {
+ out.add(message(c, sub, seq));
+ }
+ }
+ long left = deadline - System.nanoTime();
+ if (!out.isEmpty() || left <= 0) {
+ return out;
+ }
+ try {
+ TimeUnit.NANOSECONDS.timedWait(this, left);
+ } catch (InterruptedException e) {
+ // as jnats: an interrupted fetch keeps the interrupt and
returns what it has
+ Thread.currentThread().interrupt();
+ return out;
+ }
+ }
+ }
+
+ private Message message(FakeConsumer c, JetStreamSubscription sub,
long seq) {
+ byte[] 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" -> {
+ synchronized (this) {
+ c.redeliver.add(seq);
+ }
+ yield null;
+ }
+ case "ack", "term" -> {
+ synchronized (this) {
+ if (retention == RetentionPolicy.WorkQueue) {
+ removed.add(seq);
+ }
+ }
+ yield null;
+ }
+ default -> null;
+ });
+ }
+
+ @FunctionalInterface
+ interface Handler {
+ Object handle(String method, Object[] args) throws Exception;
+ }
+
+ @SuppressWarnings("unchecked")
+ static <T> T proxy(Class<T> 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;
+ });
+ }
+ }
+}
diff --git
a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java
b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java
new file mode 100644
index 000000000000..72456842c1f3
--- /dev/null
+++
b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java
@@ -0,0 +1,263 @@
+/*
+ * 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.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+
+import io.nats.client.api.AckPolicy;
+import io.nats.client.api.ConsumerConfiguration;
+import io.nats.client.api.DeliverPolicy;
+import io.nats.client.support.Status;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import
org.apache.camel.component.nats.NatsConsumerPullConsumerDeletedTest.FakeServer;
+import org.apache.camel.spi.ExceptionHandler;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.Logger;
+import org.apache.logging.log4j.core.appender.AbstractAppender;
+import org.apache.logging.log4j.core.config.Property;
+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.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * What the JetStream pull consumer does when the server no longer has its
consumer, besides creating an ephemeral
+ * consumer again (see {@link NatsConsumerPullConsumerDeletedTest}): a durable
consumer is not created again, a failed
+ * subscribe is tried again, and a start time is not sent together with the
start sequence.
+ */
+class NatsConsumerPullConsumerGoneTest extends CamelTestSupport {
+
+ private final FakeServer server = new FakeServer();
+ private final List<Throwable> handled = new CopyOnWriteArrayList<>();
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ context.getRegistry().bind("conn", server.connection());
+ context.getRegistry().bind("eh", new ExceptionHandler() {
+ @Override
+ public void handleException(Throwable exception) {
+ handled.add(exception);
+ }
+
+ @Override
+ public void handleException(String message, Throwable exception) {
+ handled.add(exception);
+ }
+
+ @Override
+ public void handleException(String message, Exchange exchange,
Throwable exception) {
+ handled.add(exception);
+ }
+ });
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // each test adds its route
+ }
+ };
+ }
+
+ private void addRoute(String options) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("nats:orders?connection=#conn&jetstreamEnabled=true&jetstreamName=ORDERS&pullFetchTimeout=100"
+ + "&exceptionHandler=#eh" + options)
+ .convertBodyTo(String.class)
+ .to("mock:result");
+ }
+ });
+ }
+
+ @Test
+ void durableConsumerIsNotCreatedAgain() throws Exception {
+ List<String> warnings = new CopyOnWriteArrayList<>();
+ AbstractAppender appender = new AbstractAppender("CaptureWarn", null,
null, true, Property.EMPTY_ARRAY) {
+ @Override
+ public void append(LogEvent event) {
+ if (event.getLevel() == Level.WARN
+ &&
event.getMessage().getFormattedMessage().contains("durable JetStream
consumer")) {
+ warnings.add(event.getMessage().getFormattedMessage());
+ }
+ }
+ };
+ appender.start();
+ Logger logger = (Logger) LogManager.getLogger(NatsConsumer.class);
+ logger.addAppender(appender);
+ try {
+ addRoute("&durableName=orders");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("order-1");
+ server.publish("order-1");
+ mock.assertIsSatisfied();
+
+ // an administrator deleted the durable: a waiting pull gets 409
Consumer Deleted, then the fetches get
+ // nothing and the consumer check (every 5 seconds) finds it gone
+ server.deleteConsumers(true);
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() ->
assertEquals(1, warnings.size()));
+ int fetchesBefore = server.fetches();
+ long waitStarted = System.nanoTime();
+ await().during(6, TimeUnit.SECONDS).atMost(8,
TimeUnit.SECONDS).untilAsserted(() -> {
+ assertEquals(1, server.consumersCreated(), "the durable is not
created again");
+ assertEquals(1, warnings.size(), "the WARN is logged once");
+ });
+ // while the durable is gone each fetch ends at once: the loop
waits pullFetchTimeout (100 ms) between them
+ long waitedMillis =
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - waitStarted);
+ int fetches = server.fetches() - fetchesBefore;
+ assertTrue(fetches <= waitedMillis / 100 + 5,
+ "no pull requests back to back: " + fetches + " fetches in
" + waitedMillis + " ms");
+
+ // the durable is created again with the same name: the same
subscription fetches again
+ mock.reset();
+ mock.expectedBodiesReceived("order-2");
+ server.restoreConsumers();
+ server.publish("order-2");
+ mock.setResultWaitTime(20_000);
+ mock.assertIsSatisfied();
+ assertEquals(1, server.consumersCreated());
+ } finally {
+ logger.removeAppender(appender);
+ appender.stop();
+ }
+ }
+
+ @Test
+ void failedSubscribeIsTriedAgainAtTheNextCheck() throws Exception {
+ addRoute("");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("order-1");
+ server.publish("order-1");
+ mock.assertIsSatisfied();
+
+ // the subscribe fails, for example while the stream itself is gone:
jnats reports that failure as an
+ // IllegalStateException
+ server.failNextSubscribe(new IllegalStateException("[SUB-90007] No
matching streams for subject."));
+ server.deleteConsumers(false);
+
+ mock.reset();
+ mock.expectedBodiesReceived("order-2");
+ server.publish("order-2");
+ mock.setResultWaitTime(30_000);
+ mock.assertIsSatisfied();
+ assertEquals(2, server.consumersCreated(), "created again at the check
after the failed one");
+
assertTrue(handled.stream().anyMatch(IllegalStateException.class::isInstance),
+ "the failed subscribe goes to the exception handler");
+ }
+
+ @Test
+ void idleHealthyConsumerIsNotAskedForItsInfo() throws Exception {
+ addRoute("");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("order-1");
+ server.publish("order-1");
+ mock.assertIsSatisfied();
+
+ // each empty fetch of a healthy consumer lasts the whole fetch
timeout: no consumer check after it
+ await().during(6, TimeUnit.SECONDS).atMost(8, TimeUnit.SECONDS)
+ .untilAsserted(() -> assertEquals(0,
server.consumerInfoRequests()));
+ }
+
+ @Test
+ void otherErrorStatusIsNotFetchedAgainBackToBack() throws Exception {
+ server.failFetches(new Status(409, "Consumer is push based"));
+ addRoute("");
+ await().atMost(5, TimeUnit.SECONDS).until(() -> !handled.isEmpty());
+ int fetchesBefore = server.fetches();
+ long waitStarted = System.nanoTime();
+ // the loop waits for the rest of pullFetchTimeout (100 ms) after each
failed fetch
+ await().during(1, TimeUnit.SECONDS).atMost(2,
TimeUnit.SECONDS).untilAsserted(() -> {
+ long waitedMillis =
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - waitStarted);
+ int fetches = server.fetches() - fetchesBefore;
+ assertTrue(fetches <= waitedMillis / 100 + 5,
+ "no pull requests back to back: " + fetches + " fetches in
" + waitedMillis + " ms");
+ });
+ }
+
+ @Test
+ void unexpectedFetchFailureIsNotFetchedAgainBackToBack() throws Exception {
+ server.failFetches(new RuntimeException("connection is recovering"));
+ addRoute("");
+ await().atMost(5, TimeUnit.SECONDS).until(() -> !handled.isEmpty());
+ int fetchesBefore = server.fetches();
+ long waitStarted = System.nanoTime();
+ // the loop waits for the rest of pullFetchTimeout (100 ms) after each
failed fetch
+ await().during(1, TimeUnit.SECONDS).atMost(2,
TimeUnit.SECONDS).untilAsserted(() -> {
+ long waitedMillis =
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - waitStarted);
+ int fetches = server.fetches() - fetchesBefore;
+ assertTrue(fetches <= waitedMillis / 100 + 5,
+ "no pull requests back to back: " + fetches + " fetches in
" + waitedMillis + " ms");
+ });
+ }
+
+ @Test
+ void ackPolicyNoneResumesAfterTheHighestFetched() throws Exception {
+ // the route never calls ack: with ackPolicy=None the delivery is the
acknowledgement
+ addRoute("&manualAck=true&ackPolicy=None");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("order-1", "order-2");
+ server.publish("order-1");
+ server.publish("order-2");
+ mock.assertIsSatisfied();
+
+ server.deleteConsumers(false);
+
+ mock.reset();
+ mock.expectedBodiesReceived("order-3");
+ server.publish("order-3");
+ mock.setResultWaitTime(20_000);
+ mock.assertIsSatisfied();
+ ConsumerConfiguration again = server.subscribed().get(1);
+ assertEquals(AckPolicy.None, again.getAckPolicy());
+ assertEquals(DeliverPolicy.ByStartSequence, again.getDeliverPolicy());
+ assertEquals(3, again.getStartSequence(), "starts after the highest
fetched sequence");
+ }
+
+ @Test
+ void startTimeIsNotSentWithTheStartSequence() throws Exception {
+ context.getRegistry().bind("cc", ConsumerConfiguration.builder()
+ .deliverPolicy(DeliverPolicy.ByStartTime)
+ .startTime(ZonedDateTime.now().minusHours(1))
+ .build());
+ addRoute("&consumerConfiguration=#cc");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("order-1");
+ server.publish("order-1");
+ mock.assertIsSatisfied();
+
+ server.deleteConsumers(true);
+
+ mock.reset();
+ mock.expectedBodiesReceived("order-2");
+ server.publish("order-2");
+ mock.setResultWaitTime(20_000);
+ mock.assertIsSatisfied();
+ ConsumerConfiguration again = server.subscribed().get(1);
+ assertEquals(DeliverPolicy.ByStartSequence, again.getDeliverPolicy());
+ assertEquals(2, again.getStartSequence());
+ assertNull(again.getStartTime());
+ }
+}
diff --git
a/components/camel-nats/src/test/java/org/apache/camel/component/nats/jetstream/NatsJetstreamConsumerPullDeletedIT.java
b/components/camel-nats/src/test/java/org/apache/camel/component/nats/jetstream/NatsJetstreamConsumerPullDeletedIT.java
new file mode 100644
index 000000000000..9d747783652e
--- /dev/null
+++
b/components/camel-nats/src/test/java/org/apache/camel/component/nats/jetstream/NatsJetstreamConsumerPullDeletedIT.java
@@ -0,0 +1,107 @@
+/*
+ * 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.jetstream;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import io.nats.client.Connection;
+import io.nats.client.JetStreamManagement;
+import io.nats.client.Nats;
+import io.nats.client.Options;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.component.nats.NatsComponent;
+import org.apache.camel.component.nats.integration.NatsITSupport;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.Isolated;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The server deletes the ephemeral consumer of a pull subscription while the
route is busy with a message, so that no
+ * pull request is waiting and no 409 Consumer Deleted arrives: the consumer
must find out that its consumer is gone and
+ * subscribe again after the message it routed.
+ */
+@Isolated
+class NatsJetstreamConsumerPullDeletedIT extends NatsITSupport {
+
+ private static final String STREAM = "mystream-pull-deleted";
+
+ private final CountDownLatch busy = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+
+ @EndpointInject("mock:result")
+ MockEndpoint mockResultEndpoint;
+
+ @Test
+ void consumesAgainAfterTheEphemeralConsumerWasDeleted() throws Exception {
+ waitForNatsConsumers(1);
+ mockResultEndpoint.expectedBodiesReceived("order-1", "order-2");
+
+ template.sendBody("direct:send", "order-1");
+ assertTrue(busy.await(20, TimeUnit.SECONDS), "order-1 is routed");
+
+ String[] servers = Arrays.stream(
+ context.getComponent("nats",
NatsComponent.class).getConfiguration().getServers().split(","))
+ .map(server -> server.contains("://") ? server : "nats://" +
server)
+ .toArray(String[]::new);
+ try (Connection connection = Nats.connect(new
Options.Builder().servers(servers).build())) {
+ JetStreamManagement jsm = connection.jetStreamManagement();
+ List<String> consumers = jsm.getConsumerNames(STREAM);
+ assertEquals(1, consumers.size(), "the ephemeral consumer of the
route");
+ jsm.deleteConsumer(STREAM, consumers.get(0));
+ }
+ release.countDown();
+
+ template.sendBody("direct:send", "order-2");
+ mockResultEndpoint.setResultWaitTime(30_000);
+ mockResultEndpoint.assertIsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ String producerUri =
"nats:mytopic-pull-deleted?jetstreamEnabled=true&jetstreamName=" + STREAM
+ + "&jetstreamAsync=false";
+ String consumerUri =
"nats:mytopic-pull-deleted?jetstreamEnabled=true&jetstreamName=" + STREAM
+ +
"&pullSubscription=true&pullFetchTimeout=500";
+
+ from("direct:send")
+ // when running full test suite then send can fail due
to nats server setup/teardown
+
.errorHandler(defaultErrorHandler().maximumRedeliveries(5))
+ .to(producerUri);
+
+ from(consumerUri)
+ .convertBodyTo(String.class)
+ .process(exchange -> {
+ if
("order-1".equals(exchange.getIn().getBody(String.class))) {
+ busy.countDown();
+ release.await(30, TimeUnit.SECONDS);
+ }
+ })
+ .to(mockResultEndpoint);
+ }
+ };
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 55758c2545c6..982ce7102bdc 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -2143,6 +2143,20 @@ in full and rejected by the broker.
A settlement request the broker rejects is now reported through the consumer's
exception handler.
+=== camel-nats - the JetStream pull consumer subscribes again when its
ephemeral consumer was deleted
+
+With `jetstreamEnabled=true` and a pull subscription (the default), the `nats`
consumer kept fetching from a JetStream
+consumer that the server had deleted, and consumed nothing more until the
route was restarted. It now subscribes again
+when the server no longer has its ephemeral consumer (no `durableName`, and no
durable name in the
+`consumerConfiguration`), which the server deletes after about 5 seconds
without a pull request, for example while the
+client is disconnected, or loses on a restart. The new consumer starts at the
first message the route did not
+acknowledge, so the route can see again the messages after that one (with
`ackPolicy=None`, after the last message
+fetched). On a stream with `WorkQueue` retention, and on a stream that was
created again since the route subscribed, the
+new consumer uses the consumer configuration as it is. The consumer asks the
server for its consumer only after a fetch
+that ended empty well before `pullFetchTimeout`, at most every 5 seconds. A
durable consumer is not created again: when
+it is gone (deleted by an administrator, or after its `inactiveThreshold`),
the consumer logs a WARN and keeps fetching,
+and the fetches work again when the durable is created again with the same
name.
+
=== camel-telegram - a failed send fails the exchange once
When Telegram answers a message with an error status (for example `400` for an
unknown chat), or with a body that
