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 8d11c415f0878..b83225d25e829 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 8d11c415f0878..b83225d25e829 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 88026b0dbb2c2..4e1903c166030 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 @@ protected void doStart() throws Exception {
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 @@ private void setupJetStreamConsumer(String topic, String queueName) throws IOExc
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 0000000000000..25bd9a1eb984d
--- /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.
+ *
+ * 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.
+ *
+ * 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.
+ *
+ * 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.
+ *
+ * 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.
+ *
+ * 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 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 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 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.
+ *
+ * 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.
+ *
+ * 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 0000000000000..f663f04fe437b
--- /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 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 stream = new ArrayList<>();
+ final Set removed = new HashSet<>();
+ RetentionPolicy retention = RetentionPolicy.Limits;
+ ZonedDateTime created = DateTimeUtils.gmtNow().minusHours(1);
+ final List consumers = new ArrayList<>();
+ final List configurations = new ArrayList<>();
+ final List 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 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 deliveries = new HashMap<>();
+ long next;
+ boolean deleted;
+ boolean unsubscribed;
+ final List 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(
+ 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(
+ 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 proxy(Class type, Handler handler) {
+ return (T) Proxy.newProxyInstance(FakeServer.class.getClassLoader(), new Class[] { type },
+ (p, m, a) -> {
+ if (m.getDeclaringClass() == Object.class) {
+ return switch (m.getName()) {
+ case "hashCode" -> System.identityHashCode(p);
+ case "equals" -> p == a[0];
+ default -> type.getSimpleName() + "@fake";
+ };
+ }
+ Object r = handler.handle(m.getName(), a == null ? new Object[0] : a);
+ if (r == null && m.getReturnType() == boolean.class) {
+ return false;
+ }
+ return r;
+ });
+ }
+ }
+}
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 c513494b38f69..7f47d241db2a0 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
@@ -2169,6 +2169,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