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 dfd02e66a4d0 CAMEL-25845: camel-nats - the JetStream push consumer
subscribes again when the server deleted its ephemeral consumer (#28318)
dfd02e66a4d0 is described below
commit dfd02e66a4d0dfe8e435f680bbd9b544cddba032
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 24 02:55:58 2026 +0530
CAMEL-25845: camel-nats - the JetStream push consumer subscribes again when
the server deleted its ephemeral consumer (#28318)
With pullSubscription=false and no durableName the JetStream push consumer
is ephemeral, and the
NATS server deletes it after its inactive threshold while the client is
disconnected (or on a
server restart). jnats subscribes the deliver subject again after the
reconnect but does not create
the consumer again, so the route received nothing, without any log.
After every reconnect the consumer now checks its consumer and, when an
ephemeral one is gone,
subscribes again from the first delivered and unacknowledged message (or
after the last delivered),
handling AckPolicy.None, maxDeliver, the deliver policies New/Last,
WorkQueue streams and streams
created again. A durable consumer is not created again (WARN). Failed
checks and subscribes are
retried with backoff, and a keeper stopped while subscribing leaves the
connection.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/docs/nats-component.adoc | 30 +
.../camel-nats/src/main/docs/nats-component.adoc | 30 +
.../apache/camel/component/nats/NatsConsumer.java | 25 +-
.../nats/NatsJetStreamPushConsumerKeeper.java | 393 ++++++++++
.../nats/NatsConsumerPushConsumerDeletedTest.java | 866 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 8 +
6 files changed, 1351 insertions(+), 1 deletion(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/nats-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/nats-component.adoc
index 8d11c415f087..b83225d25e82 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/nats-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/nats-component.adoc
@@ -292,3 +292,33 @@
from("nats:mytopic?jetstreamEnabled=true&jetstreamName=mystream&durableName=myco
If the route completes without calling any NatsManualAck method, the message
remains unacknowledged
and NATS will redeliver it after `ackWait` expires (default 30 seconds).
+
+=== JetStream push consumer whose consumer was deleted
+
+With `jetstreamEnabled=true` and `pullSubscription=false`, the server deletes
an ephemeral
+JetStream consumer (no `durableName`, and no durable name in the
`consumerConfiguration`)
+after about 5 seconds without a subscriber, for example while the client is
disconnected, and
+loses it on a restart. After every reconnect the Camel consumer asks the
server for its
+consumer, and subscribes again when its ephemeral consumer is gone:
+
+* 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
+delivered). A message that was not acknowledged and was already delivered
`maxDeliver` times
+does not count, as the server does not deliver it again.
+* When nothing was delivered yet, a consumer with the deliver policy `New` (or
`Last`) starts
+after (or at) the last message of the stream at the subscribe.
+* 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
(from the first
+message of the new stream for `New` and `Last`).
+* When the consumer cannot be checked or created again (for example JetStream
is not ready yet
+after a server restart, or the stream is gone until it is created again), the
Camel consumer
+keeps trying while the route runs, a few times a second apart and then up to
every 30 seconds,
+and logs a WARN after the first attempts.
+
+A durable consumer is not created again: when it is gone (deleted by an
administrator, or
+after its `inactiveThreshold`), the Camel consumer logs a WARN and keeps its
subscription, and
+messages arrive again when the durable is created again with the same deliver
subject.
+
+The consumer is checked only after a reconnect. A consumer that is deleted
while the
+connection stays up (for example with `nats consumer rm`) is not detected, and
the route
+receives nothing more until it is restarted.
diff --git a/components/camel-nats/src/main/docs/nats-component.adoc
b/components/camel-nats/src/main/docs/nats-component.adoc
index 8d11c415f087..b83225d25e82 100644
--- a/components/camel-nats/src/main/docs/nats-component.adoc
+++ b/components/camel-nats/src/main/docs/nats-component.adoc
@@ -292,3 +292,33 @@
from("nats:mytopic?jetstreamEnabled=true&jetstreamName=mystream&durableName=myco
If the route completes without calling any NatsManualAck method, the message
remains unacknowledged
and NATS will redeliver it after `ackWait` expires (default 30 seconds).
+
+=== JetStream push consumer whose consumer was deleted
+
+With `jetstreamEnabled=true` and `pullSubscription=false`, the server deletes
an ephemeral
+JetStream consumer (no `durableName`, and no durable name in the
`consumerConfiguration`)
+after about 5 seconds without a subscriber, for example while the client is
disconnected, and
+loses it on a restart. After every reconnect the Camel consumer asks the
server for its
+consumer, and subscribes again when its ephemeral consumer is gone:
+
+* 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
+delivered). A message that was not acknowledged and was already delivered
`maxDeliver` times
+does not count, as the server does not deliver it again.
+* When nothing was delivered yet, a consumer with the deliver policy `New` (or
`Last`) starts
+after (or at) the last message of the stream at the subscribe.
+* 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
(from the first
+message of the new stream for `New` and `Last`).
+* When the consumer cannot be checked or created again (for example JetStream
is not ready yet
+after a server restart, or the stream is gone until it is created again), the
Camel consumer
+keeps trying while the route runs, a few times a second apart and then up to
every 30 seconds,
+and logs a WARN after the first attempts.
+
+A durable consumer is not created again: when it is gone (deleted by an
administrator, or
+after its `inactiveThreshold`), the Camel consumer logs a WARN and keeps its
subscription, and
+messages arrive again when the durable is created again with the same deliver
subject.
+
+The consumer is checked only after a reconnect. A consumer that is deleted
while the
+connection stays up (for example with `nats consumer rm`) is not detected, and
the route
+receives nothing more until it is restarted.
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 88026b0dbb2c..4e1903c16603 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,7 +65,10 @@ public class NatsConsumer extends DefaultConsumer {
private Connection connection;
private Dispatcher dispatcher;
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;
+ // set by the consuming task and read when the consumer stops
+ private volatile NatsJetStreamPushConsumerKeeper pushConsumerKeeper;
public NatsConsumer(NatsEndpoint endpoint, Processor processor) {
super(endpoint, processor);
@@ -103,6 +106,11 @@ public class NatsConsumer extends DefaultConsumer {
protected void doStop() throws Exception {
final NatsConfiguration configuration =
this.getEndpoint().getConfiguration();
+ if (this.pushConsumerKeeper != null) {
+ this.pushConsumerKeeper.stop();
+ this.pushConsumerKeeper = null;
+ }
+
if (configuration.isFlushConnection() &&
ObjectHelper.isNotEmpty(this.connection)) {
LOG.debug("Flushing Messages before stopping");
this.connection.flush(Duration.ofMillis(configuration.getFlushTimeout()));
@@ -275,13 +283,28 @@ public class NatsConsumer extends DefaultConsumer {
configuration.getTopic(), queueName,
configuration.getAckPolicy(), configuration.getAckWait(),
configuration.getNackWait());
+ // the keeper subscribes again when the server deleted the
ephemeral consumer (after a disconnect of
+ // more than its inactive threshold, server restart); a
durable consumer is not created again
+ NatsJetStreamPushConsumerKeeper keeper = new
NatsJetStreamPushConsumerKeeper(
+ NatsConsumer.this, this.connection, streamName,
configuration.getTopic(), queueName,
+ dispatcher, messageHandler, cc,
NatsConsumer.this.executor,
+ sub -> NatsConsumer.this.jetStreamSubscription = sub);
NatsConsumer.this.jetStreamSubscription =
this.connection.jetStream().subscribe(
NatsConsumer.this.getEndpoint().getConfiguration().getTopic(),
queueName,
dispatcher,
- messageHandler,
+ keeper,
autoAck,
pushOptions);
+ NatsConsumer.this.pushConsumerKeeper = keeper;
+ keeper.start(NatsConsumer.this.jetStreamSubscription,
+ streamInfo != null ? streamInfo.getCreateTime() : null,
+ streamInfo != null && streamInfo.getStreamState() !=
null
+ ?
streamInfo.getStreamState().getLastSequence() : -1);
+ if (!NatsConsumer.this.isRunAllowed()) {
+ // stopped while subscribing: doStop did not see the keeper
+ keeper.stop();
+ }
NatsConsumer.this.setActive(true);
}
diff --git
a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsJetStreamPushConsumerKeeper.java
b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsJetStreamPushConsumerKeeper.java
new file mode 100644
index 000000000000..25bd9a1eb984
--- /dev/null
+++
b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsJetStreamPushConsumerKeeper.java
@@ -0,0 +1,393 @@
+/*
+ * 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.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
+
+import io.nats.client.Connection;
+import io.nats.client.ConnectionListener;
+import io.nats.client.Dispatcher;
+import io.nats.client.JetStreamApiException;
+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.AckPolicy;
+import io.nats.client.api.ConsumerConfiguration;
+import io.nats.client.api.DeliverPolicy;
+import io.nats.client.api.RetentionPolicy;
+import io.nats.client.api.StreamInfo;
+import io.nats.client.impl.AckType;
+import org.apache.camel.spi.ExceptionHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Keeps a JetStream push subscription consuming when the server deleted its
consumer.
+ * <p>
+ * The server deletes an ephemeral push consumer (no durable name) when nobody
is subscribed to its deliver subject 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. jnats subscribes the deliver subject again after
the reconnect, but it does not create the
+ * consumer again, so the subscription would never receive anything again.
After every reconnect this asks the server
+ * for the consumer and, when an ephemeral consumer is gone, subscribes again:
the new consumer starts at the first
+ * message that was delivered to the route and not acknowledged (or after the
last one delivered), so that nothing is
+ * skipped and the stream is not replayed from the deliver policy. 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. A WARN says so, and the
subscription is kept.
+ * <p>
+ * jnats answers the consumer lookup with null (not an exception) when the
server has no such consumer (api error
+ * 10014). On a stream with WorkQueue retention, which accepts only the
deliver policy All and has already removed the
+ * acknowledged messages, and on a stream that was created again since the
subscribe (its sequences started again at 1),
+ * the new consumer uses the consumer configuration as it is.
+ * <p>
+ * When the consumer cannot be checked or created again (JetStream not ready
yet after a server restart, the stream is
+ * gone), this keeps trying while the route runs: a few times one second
apart, then with a growing delay up to 30
+ * seconds, with a WARN after the first attempts.
+ * <p>
+ * The consumer is checked only after a reconnect (when jnats has subscribed
the deliver subject again). A consumer that
+ * is deleted while the connection stays up (for example by an administrator)
is not detected, and the route receives
+ * nothing more: an idle heartbeat would detect it, but it is not allowed with
a queue group, and its alarm goes to the
+ * error listener of the connection, which the route does not own when the
connection is shared.
+ * <p>
+ * The delivered messages that are not settled yet are found with {@link
Message#lastAck()} on each delivery. jnats
+ * writes that field on the thread that acknowledges, without synchronization,
so a read here can miss a recent
+ * acknowledgement: the consumer then resumes at an earlier message and the
server delivers it again, nothing is
+ * skipped.
+ */
+final class NatsJetStreamPushConsumerKeeper implements MessageHandler,
ConnectionListener {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(NatsJetStreamPushConsumerKeeper.class);
+ // the attempts one second apart, before the delay grows up to the maximum
+ private static final int CHECK_ATTEMPTS = 5;
+ private static final long CHECK_RETRY_DELAY_MILLIS = 1000;
+ private static final long MAX_CHECK_RETRY_DELAY_MILLIS = 30000;
+
+ private final NatsConsumer consumer;
+ private final Connection connection;
+ private final String streamName;
+ private final String topic;
+ private final String queueName;
+ private final Dispatcher dispatcher;
+ private final MessageHandler handler;
+ private final ConsumerConfiguration configuration;
+ private final ExecutorService executor;
+ private final Consumer<JetStreamSubscription> onSubscribed;
+ private final Lock lock = new ReentrantLock();
+ // whether a check is submitted and has not started yet, so that
reconnects do not queue several checks
+ private final AtomicBoolean checkQueued = new AtomicBoolean();
+ // the JetStream messages delivered to the route that may still need a
delivery, by stream sequence
+ private final ConcurrentSkipListMap<Long, Message> unacknowledged = new
ConcurrentSkipListMap<>();
+ private final AtomicLong highestDelivered = new AtomicLong();
+ // with AckPolicy.None the delivery is the acknowledgement: no delivered
message waits for an ack
+ private final boolean trackUnacknowledged;
+ // the server delivers a message at most maxDeliver times (when set): a
message delivered that often is settled
+ private final long maxDeliver;
+ private volatile JetStreamSubscription subscription;
+ // 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 volatile ZonedDateTime streamCreated;
+ // the last sequence of the stream at the subscribe (-1 if not known),
where a consumer that starts at the new (or
+ // last) messages started
+ private volatile long streamLastSequence = -1;
+ // why the consumer counts as gone, for the log
+ private volatile String goneReason;
+
+ NatsJetStreamPushConsumerKeeper(NatsConsumer consumer, Connection
connection, String streamName, String topic,
+ String queueName, Dispatcher dispatcher,
MessageHandler handler,
+ ConsumerConfiguration configuration,
ExecutorService executor,
+ Consumer<JetStreamSubscription>
onSubscribed) {
+ this.consumer = consumer;
+ this.connection = connection;
+ this.streamName = streamName;
+ this.topic = topic;
+ this.queueName = queueName;
+ this.dispatcher = dispatcher;
+ this.handler = handler;
+ this.configuration = configuration;
+ this.executor = executor;
+ this.onSubscribed = onSubscribed;
+ this.trackUnacknowledged = configuration.getAckPolicy() !=
AckPolicy.None;
+ this.maxDeliver = configuration.getMaxDeliver();
+ }
+
+ /**
+ * Starts watching the connection for reconnects, once the first
subscription is made.
+ *
+ * @param streamCreated when the stream was created, as seen before
the subscribe (null if not known)
+ * @param streamLastSequence the last sequence of the stream, as seen
before the subscribe (-1 if not known)
+ */
+ void start(JetStreamSubscription subscription, ZonedDateTime
streamCreated, long streamLastSequence) {
+ this.subscription = subscription;
+ this.streamCreated = streamCreated;
+ this.streamLastSequence = streamLastSequence;
+ connection.addConnectionListener(this);
+ }
+
+ void stop() {
+ connection.removeConnectionListener(this);
+ }
+
+ @Override
+ public void onMessage(Message msg) throws InterruptedException {
+ if (msg.isJetStream()) {
+ long seq = msg.metaData().streamSequence();
+ if (trackUnacknowledged) {
+ unacknowledged.values().removeIf(this::settled);
+ unacknowledged.put(seq, msg);
+ }
+ highestDelivered.accumulateAndGet(seq, Math::max);
+ }
+ handler.onMessage(msg);
+ }
+
+ @Override
+ public void connectionEvent(Connection conn, Events type) {
+ // jnats has subscribed the deliver subject again: is the consumer
still there?
+ if (type == Events.RESUBSCRIBED && consumer.isRunAllowed() &&
checkQueued.compareAndSet(false, true)) {
+ try {
+ executor.submit(this::checkConsumer);
+ } catch (Exception e) {
+ checkQueued.set(false);
+ LOG.debug("Cannot check the JetStream consumer of the push
subscription on topic {}", topic, e);
+ }
+ }
+ }
+
+ void checkConsumer() {
+ try {
+ // a check runs on the executor of the consumer (a pool), and
another check may hold the lock while it
+ // waits to try again: the stop of the consumer interrupts both
(shutdownNow), and this one then leaves at
+ // once instead of waiting for the lock
+ lock.lockInterruptibly();
+ } catch (InterruptedException e) {
+ checkQueued.set(false);
+ Thread.currentThread().interrupt();
+ return;
+ }
+ try {
+ // a reconnect from now on queues the next check
+ checkQueued.set(false);
+ long delay = CHECK_RETRY_DELAY_MILLIS;
+ boolean reported = false;
+ for (int attempt = 1; consumer.isRunAllowed(); attempt++) {
+ Boolean gone = consumerGone();
+ if (Boolean.FALSE.equals(gone)) {
+ if (attempt > CHECK_ATTEMPTS) {
+ LOG.info("The JetStream consumer of the push
subscription on topic {} is there again", topic);
+ }
+ return;
+ }
+ if (Boolean.TRUE.equals(gone)) {
+ if (consumerDeleted(!reported)) {
+ return;
+ }
+ // the failure went to the exception handler once, the
next ones are logged at DEBUG
+ reported = true;
+ }
+ // could not check (not connected, JetStream not ready,
timeout), or could not subscribe again: try
+ // again while the route runs, as nothing else makes this
subscription receive again
+ if (attempt == CHECK_ATTEMPTS) {
+ LOG.warn("Cannot check, or create again, the JetStream
consumer of the push subscription on topic {}"
+ + " after {} attempts; the route receives nothing
until it succeeds, trying again every"
+ + " {} seconds at most",
+ topic, attempt, MAX_CHECK_RETRY_DELAY_MILLIS /
1000);
+ }
+ if (attempt >= CHECK_ATTEMPTS) {
+ delay = Math.min(delay * 2, MAX_CHECK_RETRY_DELAY_MILLIS);
+ }
+ Thread.sleep(delay);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * @return whether the server no longer has the consumer, or null if that
could not be checked
+ */
+ private Boolean consumerGone() {
+ 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 (subscription.getConsumerInfo() == null) {
+ goneReason = "consumer not found";
+ return true;
+ }
+ return false;
+ } catch (JetStreamApiException e) {
+ if (e.getErrorCode() == 404) {
+ // jnats answers consumer not found with null, so this is
another 404, such as stream not found
+ goneReason = e.getMessage();
+ return true;
+ }
+ LOG.debug("Cannot check the JetStream consumer of the push
subscription on topic {}", topic, e);
+ } catch (Exception e) {
+ LOG.debug("Cannot check the JetStream consumer of the push
subscription on topic {}", topic, e);
+ }
+ return null;
+ }
+
+ /**
+ * The server no longer has the consumer: subscribes again for an
ephemeral consumer, and only logs a WARN for a
+ * durable one.
+ *
+ * @param report whether a failure goes to the exception handler (else it
is logged at DEBUG)
+ * @return false if subscribing again failed, to try again
+ */
+ private boolean consumerDeleted(boolean report) {
+ if (configuration.getDurable() != null) {
+ LOG.warn("The durable JetStream consumer {} of the push
subscription on topic {} is gone ({}); it is not"
+ + " created again, and messages arrive again when it is
created again with the same deliver"
+ + " subject",
+ configuration.getDurable(), topic, goneReason);
+ return true;
+ }
+ if (!consumer.isRunAllowed()) {
+ return true;
+ }
+ try {
+ subscribeAgain();
+ return true;
+ } catch (Exception e) {
+ if (!consumer.isRunAllowed()) {
+ // the consumer stopped meanwhile (its stop can close the
dispatcher): nothing to report
+ LOG.debug("Not subscribing again as the consumer of topic {}
is stopping", topic, e);
+ return true;
+ }
+ if (report) {
+ // jnats reports some subscribe failures as
IllegalStateException, such as SUB-90007 when the stream
+ // is gone
+ exceptionHandler().handleException(
+ "Error subscribing again after the JetStream consumer
of the push subscription was deleted", e);
+ } else {
+ LOG.debug("Error subscribing again after the JetStream
consumer of the push subscription on topic {}"
+ + " was deleted",
+ topic, e);
+ }
+ return false;
+ }
+ }
+
+ /**
+ * Subscribes again after the server deleted the ephemeral consumer,
starting at the first message delivered to the
+ * route and not acknowledged.
+ * <p>
+ * A message that is not acknowledged stays in the unacknowledged map
until this resubscribe, and the new consumer
+ * starts at its sequence, so the route sees the messages after it again
(at-least-once). A message delivered
+ * maxDeliver times (when set) is settled, as the server does not deliver
it again. With ackPolicy=None no message
+ * is tracked: the new consumer starts after the highest one delivered.
When nothing was delivered yet, a consumer
+ * with the deliver policy New (or Last) starts after (or at) the last
message of the stream at the subscribe, so
+ * that the messages published meanwhile are not skipped.
+ * <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 void subscribeAgain() throws Exception {
+ StreamInfo stream =
connection.jetStreamManagement().getStreamInfo(streamName);
+ ZonedDateTime created = stream != null ? stream.getCreateTime() : null;
+ 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 (from its first message for the deliver policy
New or Last)",
+ streamName);
+ unacknowledged.clear();
+ highestDelivered.set(0);
+ // every message of the new stream came after the subscribe
+ streamLastSequence = 0;
+ }
+ if (created != null) {
+ streamCreated = created;
+ }
+ boolean workQueue = stream != null && stream.getConfiguration() != null
+ && stream.getConfiguration().getRetentionPolicy() ==
RetentionPolicy.WorkQueue;
+ ConsumerConfiguration again = configuration;
+ unacknowledged.values().removeIf(this::settled);
+ long highest = highestDelivered.get();
+ DeliverPolicy policy = configuration.getDeliverPolicy();
+ long lastAtSubscribe = streamLastSequence;
+ // the stream sequence to start at, 0 to use the consumer
configuration as it is
+ long start = 0;
+ if (workQueue) {
+ LOG.debug("The JetStream stream {} has WorkQueue retention:
subscribing with the consumer configuration",
+ streamName);
+ } else if (highest > 0) {
+ start = unacknowledged.isEmpty() ? highest + 1 :
unacknowledged.firstKey();
+ } else if (policy == DeliverPolicy.New && lastAtSubscribe >= 0) {
+ start = lastAtSubscribe + 1;
+ } else if (policy == DeliverPolicy.Last && lastAtSubscribe >= 0) {
+ start = Math.max(lastAtSubscribe, 1);
+ }
+ boolean resume = start > 0;
+ if (resume) {
+ // a start time (DeliverPolicy.ByStartTime) together with a start
sequence is rejected by the server
+ again = ConsumerConfiguration.builder(configuration)
+ .deliverPolicy(DeliverPolicy.ByStartSequence)
+ .startSequence(start)
+ .startTime(null)
+ .build();
+ }
+ LOG.warn("The JetStream consumer of the push subscription on topic {}
is gone ({}); subscribing again{}",
+ topic, goneReason, resume ? " from stream sequence " + start :
"");
+ PushSubscribeOptions options = PushSubscribeOptions.builder()
+ .configuration(again)
+ .deliverGroup(queueName)
+ .build();
+ JetStreamSubscription old = subscription;
+ subscription = connection.jetStream().subscribe(topic, queueName,
dispatcher, this, false, options);
+ // the new consumer delivers those again; a message it delivered
meanwhile stays tracked
+ unacknowledged.values().removeIf(m -> m.getSubscription() == old);
+ onSubscribed.accept(subscription);
+ try {
+ dispatcher.unsubscribe(old);
+ } catch (Exception e) {
+ LOG.debug("Error unsubscribing the push subscription of the
deleted consumer", e);
+ }
+ if (!consumer.isRunAllowed()) {
+ // the consumer was stopped while subscribing: its stop may have
ended the previous subscription only
+ try {
+ dispatcher.unsubscribe(subscription);
+ } catch (Exception e) {
+ LOG.debug("Error unsubscribing the push subscription made
while the consumer stopped", e);
+ }
+ }
+ }
+
+ private ExceptionHandler exceptionHandler() {
+ return consumer.getExceptionHandler();
+ }
+
+ /**
+ * Whether the server does not deliver the message again: it is
acknowledged or terminated, or it was delivered
+ * maxDeliver times.
+ */
+ private boolean settled(Message msg) {
+ AckType ack = msg.lastAck();
+ if (ack == AckType.AckAck || ack == AckType.AckTerm) {
+ return true;
+ }
+ return maxDeliver > 0 && msg.metaData().deliveredCount() >= maxDeliver;
+ }
+}
diff --git
a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPushConsumerDeletedTest.java
b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPushConsumerDeletedTest.java
new file mode 100644
index 000000000000..f663f04fe437
--- /dev/null
+++
b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPushConsumerDeletedTest.java
@@ -0,0 +1,866 @@
+/*
+ * 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.ZonedDateTime;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.nats.client.Connection;
+import io.nats.client.ConnectionListener;
+import io.nats.client.Dispatcher;
+import io.nats.client.JetStream;
+import io.nats.client.JetStreamApiException;
+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.AckPolicy;
+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.Error;
+import io.nats.client.api.RetentionPolicy;
+import io.nats.client.api.ServerInfo;
+import io.nats.client.api.StreamInfo;
+import io.nats.client.impl.AckType;
+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.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.ExceptionHandler;
+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.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * With pullSubscription=false the JetStream consumer uses a push
subscription, with an ephemeral consumer when no
+ * durableName is set. The NATS server deletes an ephemeral push consumer when
nobody is subscribed to its deliver
+ * subject 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. jnats subscribes the deliver
subject again after the reconnect but does not
+ * create the consumer again. The consumer must then subscribe again, starting
at the first message it did not
+ * acknowledge, instead of waiting for ever on a deliver subject nobody
delivers to. A durable consumer that is gone is
+ * not created again (that would replay the stream from its deliver policy).
+ */
+class NatsConsumerPushConsumerDeletedTest extends CamelTestSupport {
+
+ private final FakeServer server = new FakeServer();
+ private final AtomicInteger attempts = new AtomicInteger();
+ private final List<Throwable> handled = new CopyOnWriteArrayList<>();
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ context.getRegistry().bind("conn", server.connection());
+ context.getRegistry().bind("handler", 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);
+ }
+ });
+ // an ephemeral consumer that starts at a time: the server rejects a
start time next to a start sequence
+ context.getRegistry().bind("byStartTime",
ConsumerConfiguration.builder()
+
.deliverPolicy(DeliverPolicy.ByStartTime).startTime(ZonedDateTime.now().minusDays(1)).build());
+ // an ephemeral consumer of the messages published after the subscribe
+ context.getRegistry().bind("deliverNew",
ConsumerConfiguration.builder().deliverPolicy(DeliverPolicy.New).build());
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("nats:orders?connection=#conn&jetstreamEnabled=true&jetstreamName=ORDERS&pullSubscription=false")
+ .convertBodyTo(String.class)
+ .process(e -> {
+ if ("fail-once".equals(e.getMessage().getBody())
&& attempts.incrementAndGet() == 1) {
+ // the client is disconnected while the
message is processed, for longer than the
+ // inactive threshold: the server deletes the
consumer, the nak is lost
+ server.disconnect();
+ throw new IllegalStateException("processing
failed");
+ }
+ })
+ .to("mock:result");
+ }
+ };
+ }
+
+ @Test
+ void consumesAgainAfterTheServerDeletedTheEphemeralConsumer() throws
Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("order-1");
+ server.publish("order-1");
+ mock.assertIsSatisfied();
+
+ // disconnected for longer than the inactive threshold: the server
deletes the consumer
+ server.disconnect();
+ server.publish("order-2");
+ server.publish("order-3");
+ server.reconnect();
+
+ mock.reset();
+ mock.expectedBodiesReceived("order-2", "order-3");
+ mock.setResultWaitTime(20_000);
+ mock.assertIsSatisfied();
+ assertEquals(2, server.consumersCreated, "one consumer created again");
+ }
+
+ @Test
+ void redeliversTheMessageWhoseNakWasLostWithTheConsumer() throws Exception
{
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("order-1");
+ server.publish("order-1");
+ mock.assertIsSatisfied();
+
+ server.publish("fail-once");
+ server.awaitDisconnected();
+ server.publish("order-3");
+ server.reconnect();
+
+ mock.reset();
+ mock.expectedBodiesReceived("fail-once", "order-3");
+ mock.setResultWaitTime(20_000);
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void endsTheSubscriptionItMadeWhileTheRouteStopped() throws Exception {
+ consumeWith("exceptionHandler=#handler");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("order-1");
+ server.publish("order-1");
+ mock.assertIsSatisfied();
+
+ // the route is stopped while the consumer subscribes again: the stop
ends the subscription of the deleted
+ // consumer only, so the consumer must end the one it makes
+ String routeId = context.getRoutes().get(0).getId();
+ server.beforeSubscribe = () -> {
+ server.beforeSubscribe = null;
+ Thread stopper = new Thread(() -> {
+ try {
+ context.getRouteController().stopRoute(routeId);
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ }, "route-stopper");
+ stopper.start();
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(20);
+ while (stopper.isAlive() && System.nanoTime() < deadline) {
+ try {
+ stopper.join(1000);
+ } catch (InterruptedException e) {
+ // the stop interrupts the executor of the consumer, which
runs this
+ }
+ }
+ };
+ int before;
+ synchronized (server) {
+ before = server.consumers.size();
+ }
+ server.disconnect();
+ server.reconnect();
+
+ // the new subscription is ended, or (when the stop closes the
dispatcher) jnats refuses to subscribe on the
+ // closed dispatcher; either way no subscription is left and nothing
is reported during the normal stop
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> {
+ synchronized (server) {
+ FakeServer.FakeConsumer last =
server.consumers.get(server.consumers.size() - 1);
+ assertTrue(server.consumers.size() == before + 1 &&
last.unsubscribed || server.refusedSubscribes > 0,
+ "the subscription made while the route stopped is
ended");
+ }
+ });
+ assertEquals(List.of(), handled, "a subscribe that fails as the route
stops is not reported");
+ }
+
+ @Test
+ void removesItsConnectionListenerWhenTheRouteStopsWhileItSubscribes()
throws Exception {
+ int added = server.listenersAdded.get();
+ // the route is stopped while the consumer makes its first
subscription, so the stop does not see the keeper
+ // yet: the keeper must not stay registered on the (shared) connection
+ server.beforeSubscribe = () -> {
+ server.beforeSubscribe = null;
+ String routeId = context.getRoutes().get(0).getId();
+ Thread stopper = new Thread(() -> {
+ try {
+ context.getRouteController().stopRoute(routeId);
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ }, "route-stopper");
+ stopper.start();
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(20);
+ while (stopper.isAlive() && System.nanoTime() < deadline) {
+ try {
+ stopper.join(1000);
+ } catch (InterruptedException e) {
+ // the stop interrupts the executor of the consumer, which
runs this
+ }
+ }
+ };
+ consumeWith("exceptionHandler=#handler");
+
+ // the keeper of the new route registered its listener after the stop,
or (when the stop closes the dispatcher,
+ // CAMEL-25847) jnats refused the subscribe on the closed dispatcher
and no keeper was started
+ await().atMost(20, TimeUnit.SECONDS).until(() -> {
+ synchronized (server) {
+ return server.listenersAdded.get() == added + 1 ||
server.refusedSubscribes > 0;
+ }
+ });
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(List.of(), server.listeners, "no keeper
listens to the connection after the stop"));
+ }
+
+ /**
+ * Replaces the route of the test by one consuming with the given options.
+ */
+ private void consumeWith(String options) throws Exception {
+ String routeId = context.getRoutes().get(0).getId();
+ context.getRouteController().stopRoute(routeId);
+ context.removeRoute(routeId);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("nats:orders?connection=#conn&jetstreamEnabled=true&jetstreamName=ORDERS&pullSubscription=false&"
+ + options)
+ .convertBodyTo(String.class)
+ .process(e -> {
+ if ("poison".equals(e.getMessage().getBody())) {
+ throw new IllegalStateException("cannot
process the poison message");
+ }
+ })
+ .to("mock:result");
+ }
+ });
+ }
+
+ @Test
+ void doesNotCreateAGoneDurableConsumerAgain() throws Exception {
+ consumeWith("durableName=orders");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("order-1");
+ server.publish("order-1");
+ mock.assertIsSatisfied();
+ int created;
+ synchronized (server) {
+ created = server.consumersCreated;
+ assertEquals("orders-durable",
server.configurations.get(server.configurations.size() - 1).getDurable());
+ }
+
+ // the durable consumer is deleted (as by an administrator) while the
client is disconnected
+ server.disconnect();
+ server.reconnect();
+
+ // creating it again would replay the stream from its deliver policy
(DeliverPolicy.All: order-1 again)
+ await().during(3, TimeUnit.SECONDS).atMost(10,
TimeUnit.SECONDS).untilAsserted(() -> {
+ synchronized (server) {
+ assertEquals(created, server.consumersCreated, "the durable
consumer must not be created again");
+ }
+ });
+ assertEquals(1, mock.getReceivedCounter());
+ }
+
+ @Test
+ void reportsAFailedSubscribeAndTriesAgain() throws Exception {
+ consumeWith("exceptionHandler=#handler");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("order-1");
+ server.publish("order-1");
+ mock.assertIsSatisfied();
+
+ // jnats reports some subscribe failures as IllegalStateException
(SUB-90007 when the stream is gone)
+ server.beforeSubscribe = () -> {
+ server.beforeSubscribe = null;
+ throw new IllegalStateException("[SUB-90007] No matching streams
for subject.");
+ };
+ server.disconnect();
+ server.publish("order-2");
+ server.reconnect();
+
+ mock.reset();
+ mock.expectedBodiesReceived("order-2");
+ mock.setResultWaitTime(20_000);
+ mock.assertIsSatisfied();
+ assertEquals(1, handled.size(), "the failed subscribe goes to the
exception handler");
+ assertInstanceOf(IllegalStateException.class, handled.get(0));
+ }
+
+ @Test
+ void resumesAnEphemeralConsumerThatStartedAtATime() throws Exception {
+ consumeWith("consumerConfiguration=#byStartTime");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("order-1");
+ server.publish("order-1");
+ mock.assertIsSatisfied();
+
+ server.disconnect();
+ server.publish("order-2");
+ server.reconnect();
+
+ mock.reset();
+ mock.expectedBodiesReceived("order-2");
+ mock.setResultWaitTime(20_000);
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void workQueueStreamIsSubscribedAgainWithDeliverPolicyAll() throws
Exception {
+ server.retention(RetentionPolicy.WorkQueue);
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("order-1");
+ server.publish("order-1");
+ mock.assertIsSatisfied();
+ // the route acknowledges order-1 once the exchange is complete: the
stream removes it
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ synchronized (server) {
+ assertTrue(server.removed.contains(1L), "order-1 is
acknowledged and removed");
+ }
+ });
+
+ server.disconnect();
+ server.publish("order-2");
+ server.reconnect();
+
+ // a work queue stream refuses a consumer whose deliver policy is not
All (10101); order-1 was acknowledged
+ // and is removed, so the consumer configuration replays nothing
+ mock.reset();
+ mock.expectedBodiesReceived("order-2");
+ mock.setResultWaitTime(20_000);
+ mock.assertIsSatisfied();
+ synchronized (server) {
+ assertEquals(2, server.consumersCreated, "one consumer created
again");
+ assertEquals(DeliverPolicy.All,
server.configurations.get(1).getDeliverPolicy(),
+ "a work queue stream accepts only the deliver policy All");
+ }
+ }
+
+ @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.disconnect();
+ server.createStreamAgain();
+ server.publish("order-3");
+ server.reconnect();
+
+ mock.reset();
+ mock.expectedBodiesReceived("order-3");
+ mock.setResultWaitTime(20_000);
+ mock.assertIsSatisfied();
+ synchronized (server) {
+ assertEquals(DeliverPolicy.All,
server.configurations.get(1).getDeliverPolicy(),
+ "the consumer configuration, not a start sequence of the
old stream");
+ }
+ }
+
+ @Test
+ void ackPolicyNoneResumesAfterTheHighestDelivered() throws Exception {
+ // the route never calls ack: with ackPolicy=None the delivery is the
acknowledgement
+ consumeWith("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.disconnect();
+ server.publish("order-3");
+ server.reconnect();
+
+ mock.reset();
+ mock.expectedBodiesReceived("order-3");
+ mock.setResultWaitTime(20_000);
+ mock.assertIsSatisfied();
+ synchronized (server) {
+ ConsumerConfiguration again =
server.configurations.get(server.configurations.size() - 1);
+ assertEquals(AckPolicy.None, again.getAckPolicy());
+ assertEquals(DeliverPolicy.ByStartSequence,
again.getDeliverPolicy());
+ assertEquals(3, again.getStartSequence(), "starts after the
highest delivered sequence");
+ }
+ }
+
+ @Test
+ void aMessageDeliveredMaxDeliverTimesDoesNotHoldBackTheResume() throws
Exception {
+ consumeWith("maxDeliver=3");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("order-2", "order-3");
+ // the route fails the poison message: it is nak'ed until the server
stops delivering it (maxDeliver)
+ server.publish("poison");
+ server.publish("order-2");
+ server.publish("order-3");
+ mock.assertIsSatisfied();
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ synchronized (server) {
+ assertEquals(3, server.consumers.get(server.consumers.size() -
1).deliveries.get(1L),
+ "the poison message is delivered maxDeliver times");
+ }
+ });
+
+ server.disconnect();
+ server.publish("order-4");
+ server.reconnect();
+
+ // the server never delivers the poison message again, so the new
consumer starts after the highest delivered
+ mock.reset();
+ mock.expectedBodiesReceived("order-4");
+ mock.setResultWaitTime(20_000);
+ mock.assertIsSatisfied();
+ synchronized (server) {
+ ConsumerConfiguration again =
server.configurations.get(server.configurations.size() - 1);
+ assertEquals(4, again.getStartSequence(), "starts after the
highest delivered sequence");
+ }
+ }
+
+ @Test
+ void keepsCheckingUntilTheStreamIsCreatedAgain() throws Exception {
+ consumeWith("exceptionHandler=#handler");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("order-1");
+ server.publish("order-1");
+ mock.assertIsSatisfied();
+
+ // a memory stream lost on a server restart, and created again only
later
+ server.disconnect();
+ server.deleteStream();
+ server.reconnect();
+ // more checks than the first attempts one second apart
+ await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> {
+ synchronized (server) {
+ assertTrue(server.streamNotFound >= 6, "the consumer keeps
checking, " + server.streamNotFound);
+ }
+ });
+ server.createStreamAgain();
+ server.publish("order-2");
+
+ mock.reset();
+ mock.expectedBodiesReceived("order-2");
+ mock.setResultWaitTime(40_000);
+ mock.assertIsSatisfied();
+ assertEquals(1, handled.size(), "the first failure goes to the
exception handler, the next ones are logged");
+ }
+
+ @Test
+ void
deliverPolicyNewWithNothingDeliveredResumesAfterTheLastMessageAtTheSubscribe()
throws Exception {
+ consumeWith("consumerConfiguration=#deliverNew");
+ // the route subscribes on the executor of the consumer, and then
listens to the connection
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
!server.listeners.isEmpty());
+
+ // nothing was delivered when the consumer is deleted: the messages
published meanwhile must not be skipped
+ server.disconnect();
+ server.publish("order-1");
+ server.reconnect();
+
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("order-1");
+ mock.setResultWaitTime(20_000);
+ mock.assertIsSatisfied();
+ synchronized (server) {
+ ConsumerConfiguration again =
server.configurations.get(server.configurations.size() - 1);
+ assertEquals(DeliverPolicy.ByStartSequence,
again.getDeliverPolicy());
+ assertEquals(1, again.getStartSequence(), "starts after the last
message of the stream at the subscribe");
+ }
+ }
+
+ /**
+ * A JetStream server with one stream, faked at the level of the jnats
interfaces the consumer calls. A push
+ * consumer delivers from its start sequence on its own thread (like the
dispatcher of jnats), one message at a
+ * time; a nak'ed message is delivered again. While the client is
disconnected nothing is delivered and the server
+ * deletes the ephemeral consumers (their inactive threshold passes);
acknowledgements to a deleted consumer are
+ * lost. A reconnect fires RESUBSCRIBED like jnats once it subscribed the
deliver subjects again. getConsumerInfo
+ * answers null for a deleted consumer, as jnats does when the server
answers consumer not found (10014), and the
+ * info of a live one. 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. A message nak'ed is not delivered again once it was
delivered maxDeliver times (the delivered
+ * count is in the reply subject, as on the server). While the stream does
not exist, getStreamInfo and
+ * getConsumerInfo throw 404 stream not found (10059) and subscribe throws
IllegalStateException (SUB-90007), as
+ * jnats does. A closed dispatcher (closeDispatcher) refuses a subscribe
with IllegalStateException, as jnats does.
+ */
+ static final class FakeServer {
+ final List<byte[]> stream = new ArrayList<>();
+ final Set<Long> removed = new HashSet<>();
+ RetentionPolicy retention = RetentionPolicy.Limits;
+ ZonedDateTime created = DateTimeUtils.gmtNow().minusHours(1);
+ final List<FakeConsumer> consumers = new ArrayList<>();
+ final List<ConsumerConfiguration> configurations = new ArrayList<>();
+ final List<ConnectionListener> listeners = new
CopyOnWriteArrayList<>();
+ // the connection listeners added so far
+ final AtomicInteger listenersAdded = new AtomicInteger();
+ boolean connected = true;
+ boolean streamExists = true;
+ // the consumer lookups answered with stream not found
+ int streamNotFound;
+ // the subscribes refused as their dispatcher was closed
+ int refusedSubscribes;
+ final Set<Dispatcher> closedDispatchers = new HashSet<>();
+ int consumersCreated;
+ // runs on the thread that subscribes, before the subscription is made
+ volatile Runnable beforeSubscribe;
+ Connection connection;
+
+ final class FakeConsumer {
+ long maxDeliver;
+ final Map<Long, Integer> deliveries = new HashMap<>();
+ long next;
+ boolean deleted;
+ boolean unsubscribed;
+ final List<Long> redeliver = new ArrayList<>();
+ }
+
+ synchronized void retention(RetentionPolicy retention) {
+ this.retention = retention;
+ }
+
+ /**
+ * The stream is created again, for example a memory stream after a
server restart: its sequences start again at
+ * 1.
+ */
+ synchronized void createStreamAgain() {
+ streamExists = true;
+ stream.clear();
+ removed.clear();
+ created = created.plusMinutes(1);
+ for (FakeConsumer c : consumers) {
+ c.deleted = true;
+ }
+ notifyAll();
+ }
+
+ /**
+ * The stream is deleted, for example a memory stream lost on a server
restart.
+ */
+ synchronized void deleteStream() {
+ streamExists = false;
+ stream.clear();
+ removed.clear();
+ for (FakeConsumer c : consumers) {
+ c.deleted = true;
+ }
+ notifyAll();
+ }
+
+ synchronized void publish(String body) {
+ stream.add(body.getBytes(StandardCharsets.UTF_8));
+ notifyAll();
+ }
+
+ synchronized void disconnect() {
+ connected = false;
+ for (FakeConsumer c : consumers) {
+ c.deleted = true;
+ }
+ notifyAll();
+ }
+
+ synchronized void awaitDisconnected() throws InterruptedException {
+ long deadline = System.currentTimeMillis() + 20_000;
+ while (connected && System.currentTimeMillis() < deadline) {
+ wait(100);
+ }
+ }
+
+ void reconnect() {
+ synchronized (this) {
+ connected = true;
+ notifyAll();
+ }
+ for (ConnectionListener l : listeners) {
+ l.connectionEvent(connection,
ConnectionListener.Events.RECONNECTED);
+ l.connectionEvent(connection,
ConnectionListener.Events.RESUBSCRIBED);
+ }
+ }
+
+ 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 == 6 && args[5]
instanceof PushSubscribeOptions o) {
+ Runnable hook = beforeSubscribe;
+ if (hook != null) {
+ hook.run();
+ }
+ synchronized (this) {
+ if (closedDispatchers.contains((Dispatcher) args[2])) {
+ // as jnats NatsDispatcher.subscribeImplJetStream
(checkBeforeSubImpl)
+ refusedSubscribes++;
+ throw new IllegalStateException("Dispatcher is
closed");
+ }
+ }
+ return subscribe(o.getConsumerConfiguration(),
(MessageHandler) args[3]);
+ }
+ throw new UnsupportedOperationException(name);
+ });
+ ServerInfo info = new ServerInfo("INFO
{\"server_id\":\"fake\",\"jetstream\":true}");
+ connection = proxy(Connection.class, (name, args) -> switch (name)
{
+ case "getServerInfo" -> info;
+ case "jetStreamManagement" -> jsm;
+ case "jetStream" -> js;
+ case "createDispatcher" -> dispatcher();
+ case "addConnectionListener" -> {
+ listenersAdded.incrementAndGet();
+ yield listeners.add((ConnectionListener) args[0]);
+ }
+ case "closeDispatcher" -> {
+ synchronized (this) {
+ closedDispatchers.add((Dispatcher) args[0]);
+ }
+ yield null;
+ }
+ case "removeConnectionListener" ->
listeners.remove((ConnectionListener) args[0]);
+ case "getStatus" -> Connection.Status.CONNECTED;
+ default -> null;
+ });
+ return connection;
+ }
+
+ /**
+ * A new dispatcher, as jnats creates one per createDispatcher call.
+ */
+ private static Dispatcher dispatcher() {
+ return proxy(Dispatcher.class, (name, args) -> {
+ if ("unsubscribe".equals(name) && args.length == 1 && args[0]
instanceof JetStreamSubscription s) {
+ s.unsubscribe();
+ }
+ return null;
+ });
+ }
+
+ private synchronized void requireStream() throws JetStreamApiException
{
+ if (!streamExists) {
+ throw new JetStreamApiException(
+ new ApiResponse<Void>(
+ JsonParser.parseUnchecked(
+
"{\"error\":{\"code\":404,\"err_code\":10059,\"description\":\"stream not
found\"}}")) {
+ });
+ }
+ }
+
+ private synchronized StreamInfo streamInfo() throws
JetStreamApiException {
+ requireStream();
+ return new StreamInfo(
+
JsonParser.parseUnchecked("{\"config\":{\"name\":\"ORDERS\",\"subjects\":[\"orders\"],"
+ + "\"retention\":\"" + retention
+ "\"},\"created\":\""
+ +
DateTimeUtils.toRfc3339(created) + "\",\"state\":{\"last_seq\":" + stream.size()
+ + "}}"));
+ }
+
+ private synchronized JetStreamSubscription
subscribe(ConsumerConfiguration cc, MessageHandler handler)
+ throws JetStreamApiException {
+ if (!streamExists) {
+ throw new IllegalStateException("[SUB-90007] No matching
streams for subject.");
+ }
+ 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\"}}")) {
+ });
+ }
+ if (cc != null && cc.getDeliverPolicy() ==
DeliverPolicy.ByStartSequence && cc.getStartTime() != null) {
+ // as nats-server: a start time is not allowed with
DeliverPolicy.ByStartSequence
+ throw new JetStreamApiException(
+ Error.convert(new Status(
+ 400, "consumer delivery policy is deliver by"
+ + " start sequence, but optional start
time is also set")));
+ }
+ configurations.add(cc);
+ FakeConsumer c = new FakeConsumer();
+ DeliverPolicy policy = cc != null ? cc.getDeliverPolicy() :
DeliverPolicy.All;
+ if (policy == DeliverPolicy.ByStartSequence) {
+ c.next = cc.getStartSequence();
+ } else if (policy == DeliverPolicy.New) {
+ c.next = stream.size() + 1;
+ } else if (policy == DeliverPolicy.Last) {
+ c.next = Math.max(stream.size(), 1);
+ } else {
+ c.next = 1;
+ }
+ c.maxDeliver = cc != null ? cc.getMaxDeliver() : -1;
+ consumers.add(c);
+ consumersCreated++;
+ JetStreamSubscription sub = proxy(JetStreamSubscription.class,
(name, args) -> switch (name) {
+ case "getConsumerInfo" -> consumerInfo(c);
+ case "unsubscribe" -> {
+ synchronized (this) {
+ c.unsubscribed = true;
+ notifyAll();
+ }
+ yield null;
+ }
+ case "isActive" -> true;
+ default -> null;
+ });
+ Thread t = new Thread(() -> deliver(c, sub, handler), "fake-push-"
+ consumersCreated);
+ t.setDaemon(true);
+ t.start();
+ return sub;
+ }
+
+ private synchronized ConsumerInfo consumerInfo(FakeConsumer c) throws
JetStreamApiException {
+ // jnats lookupConsumerInfo maps only consumer not found to null:
stream not found is thrown
+ if (!streamExists) {
+ // counts the checks of a consumer while the stream is gone
+ streamNotFound++;
+ }
+ requireStream();
+ if (c.deleted) {
+ // jnats lookupConsumerInfo maps consumer not found (10014) to
null
+ return null;
+ }
+ return new
ConsumerInfo(JsonParser.parseUnchecked("{\"stream_name\":\"ORDERS\",\"name\":\"c\",\"config\":{}}"));
+ }
+
+ private void deliver(FakeConsumer c, JetStreamSubscription sub,
MessageHandler handler) {
+ while (true) {
+ long seq;
+ int delivered;
+ synchronized (this) {
+ while (!c.deleted && !c.unsubscribed
+ && (!connected || (c.redeliver.isEmpty() && c.next
> stream.size()))) {
+ try {
+ wait();
+ } catch (InterruptedException e) {
+ return;
+ }
+ }
+ if (c.deleted || c.unsubscribed) {
+ return;
+ }
+ seq = c.redeliver.isEmpty() ? c.next++ :
c.redeliver.remove(0);
+ if (removed.contains(seq)) {
+ continue;
+ }
+ delivered = c.deliveries.merge(seq, 1, Integer::sum);
+ }
+ try {
+ handler.onMessage(message(c, sub, seq, delivered));
+ } catch (InterruptedException e) {
+ return;
+ }
+ }
+ }
+
+ private Message message(FakeConsumer c, JetStreamSubscription sub,
long seq, int delivered) {
+ 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." + delivered
+ "." + seq + "." + seq + ".0.0", data) {
+ @Override
+ public boolean isJetStream() {
+ return true;
+ }
+ });
+ AckType[] lastAck = new AckType[1];
+ 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 "lastAck" -> lastAck[0];
+ case "ack" -> {
+ lastAck[0] = AckType.AckAck;
+ removeFromWorkQueue(c, seq);
+ yield null;
+ }
+ case "term" -> {
+ lastAck[0] = AckType.AckTerm;
+ removeFromWorkQueue(c, seq);
+ yield null;
+ }
+ case "nak", "nakWithDelay" -> {
+ lastAck[0] = AckType.AckNak;
+ synchronized (this) {
+ // the server stops delivering a message delivered
maxDeliver times
+ if (!c.deleted && (c.maxDeliver <= 0 || delivered <
c.maxDeliver)) {
+ c.redeliver.add(seq);
+ notifyAll();
+ }
+ }
+ yield null;
+ }
+ default -> null;
+ });
+ }
+
+ private synchronized void removeFromWorkQueue(FakeConsumer c, long
seq) {
+ if (retention == RetentionPolicy.WorkQueue && !c.deleted) {
+ removed.add(seq);
+ }
+ }
+
+ @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/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 ceb34e9b5906..a391291dbba9 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
@@ -2173,6 +2173,14 @@ wraps an `ExecutionException`, with the
`RuntimeCamelException` (status and body
order of the messages is the same. The steps after the producer now run on the
calling thread, and no longer on a
thread of the HTTP client.
+=== camel-nats - the JetStream push consumer subscribes again when its
ephemeral consumer was deleted
+
+With `jetstreamEnabled=true` and `pullSubscription=false`, the consumer now
creates its ephemeral JetStream consumer
+again after a reconnect when the server deleted it, starting at the first
message the route did not acknowledge, so
+the route can see again the messages after that one. A durable consumer is not
created again (a WARN is logged), and a
+consumer deleted while the connection stays up is not detected. See the
+xref:components::nats-component.adoc[NATS component] page for the details.
+
=== camel-vertx-websocket - the client consumer makes one reconnect attempt at
a time
When the WebSocket of a consumer with `consumeAsClient=true` is closed, the
consumer now makes one reconnect attempt
