From c8c69dc3ee68270ee05122289f66e72afc86d4fb Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Tue, 20 Oct 2026 08:35:18 +0530 Subject: [PATCH 1/4] CAMEL-25845: camel-nats - the JetStream push consumer subscribes again when the server deleted its ephemeral consumer With pullSubscription=false and no durableName the JetStream consumer subscribes with an ephemeral push consumer. The NATS server deletes it 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, so the route stayed started and received nothing, without any log. After every reconnect (RESUBSCRIBED) the consumer now asks the server for its consumer and, when its ephemeral consumer is gone, subscribes again. jnats (2.26.3) answers getConsumerInfo() with null when the server has no such consumer (lookupConsumerInfo maps api error 10014 to null), so null counts as gone, as does a 404 error. The new consumer starts at the first message 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. With AckPolicy.None the delivery is the acknowledgement: no message is tracked, and the new consumer starts after the last one delivered. A message delivered maxDeliver times (when set) counts as settled, as the server does not deliver it again, so a poison message does not hold the resume point back. When nothing was delivered yet, a consumer with the deliver policy New (or Last) starts after (or at) the last stream sequence seen at the subscribe, so the messages published meanwhile are not skipped. The new consumer is sent without a start time (a ByStartTime configuration would otherwise carry a start time next to the start sequence, which the server rejects). Before subscribing again the consumer looks the stream up. A stream with WorkQueue retention accepts only the deliver policy All (error 10101 otherwise) and has already removed the acknowledged messages, so the consumer configuration is used as it is. A stream whose create time differs from the one seen at the subscribe was created again (its sequences start at 1 again, a start sequence of the old stream would skip its messages): the tracked messages are cleared and the consumer configuration is used. This follows the review of the pull twin (CAMEL-25815, #28204). A durable consumer is not created again: it is gone only when an administrator deleted it or after its inactiveThreshold, and creating it again from its configuration would replay the stream from its deliver policy (the whole stream with the default DeliverPolicy.All). The keeper logs a WARN and keeps the subscription. A failed check or subscribe is tried again while the route runs: five attempts a second apart, then with a delay that doubles up to 30 seconds, with a WARN after the fifth attempt (JetStream not ready yet after a server restart, or a stream that is gone until it is created again). The first failed subscribe of a reconnect (jnats throws IllegalStateException for some, such as SUB-90007 when the stream is gone) goes to the exception handler, the next ones are logged at DEBUG. A reconnect does not queue a second check while one waits. A subscription made while the consumer was being stopped is ended again, and a subscribe that fails because the consumer stops is not reported. Co-Authored-By: Claude Opus 5.5 --- .../camel/component/nats/NatsConsumer.java | 20 +- .../nats/NatsJetStreamPushConsumerKeeper.java | 374 ++++++++ .../NatsConsumerPushConsumerDeletedTest.java | 823 ++++++++++++++++++ .../pages/camel-4x-upgrade-guide-4_24.adoc | 20 + 4 files changed, 1236 insertions(+), 1 deletion(-) create mode 100644 components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsJetStreamPushConsumerKeeper.java create mode 100644 components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPushConsumerDeletedTest.java diff --git a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java index 88026b0dbb2c2..2d5d3699142ad 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,9 @@ 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; + private NatsJetStreamPushConsumerKeeper pushConsumerKeeper; public NatsConsumer(NatsEndpoint endpoint, Processor processor) { super(endpoint, processor); @@ -103,6 +105,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 +282,24 @@ 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); 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..06cb3407229c0 --- /dev/null +++ b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsJetStreamPushConsumerKeeper.java @@ -0,0 +1,374 @@ +/* + * 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. + */ +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() { + lock.lock(); + 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..f12a46342f3f2 --- /dev/null +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPushConsumerDeletedTest.java @@ -0,0 +1,823 @@ +/* + * 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"); + } + + /** + * 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<>(); + 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" -> 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..0905b1b07d7d9 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,26 @@ 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 `nats` consumer stayed subscribed to the deliver +subject of a JetStream consumer that the server had deleted, and consumed nothing more until the route was restarted. +After every reconnect it now asks the server for its consumer and subscribes again when its ephemeral consumer (no +`durableName`, and no durable name in the `consumerConfiguration`) is gone, which the server deletes after about 5 +seconds without a subscriber, for example while the client is disconnected, or loses on a restart. The new consumer +starts at the first message the route did not acknowledge, so the route can see again the messages after that one +(with `ackPolicy=None`, after the last message 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 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 +consumer logs a WARN and keeps its subscription, and messages arrive again when the durable is created again with the +same deliver subject. + === 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 From 0dca7884490c542bcc81cfcf18b2c24b90901eed Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Fri, 23 Oct 2026 15:00:09 +0530 Subject: [PATCH 2/4] CAMEL-25845: camel-nats - end the keeper of a push consumer that is stopped while it subscribes Review of #28318: - NatsConsumer: when the route is stopped while the consuming task subscribes, doStop does not see the keeper yet, so the keeper stayed registered as a listener of the (shared) connection. After the subscribe the task now stops the keeper when the consumer is no longer allowed to run. pushConsumerKeeper is volatile (written by the task, read by doStop). - NatsJetStreamPushConsumerKeeper javadoc: the consumer is checked only after a reconnect (a consumer deleted while the connection stays up is not detected), and Message.lastAck() can be stale, which makes the consumer resume earlier and the server redeliver, never skip. - Upgrade guide: the entry is shortened to the migration note, with the limitation; the details are in a new section of nats-component.adoc (and its catalog copy). Test: removesItsConnectionListenerWhenTheRouteStopsWhileItSubscribes (the route is stopped from the subscribe of its first subscription) fails without the change: the keeper is left in the connection's listeners. Co-Authored-By: Claude Opus 5.5 --- .../camel/catalog/docs/nats-component.adoc | 30 ++++++++++++++ .../src/main/docs/nats-component.adoc | 30 ++++++++++++++ .../camel/component/nats/NatsConsumer.java | 7 +++- .../nats/NatsJetStreamPushConsumerKeeper.java | 10 +++++ .../NatsConsumerPushConsumerDeletedTest.java | 40 ++++++++++++++++++- .../pages/camel-4x-upgrade-guide-4_24.adoc | 22 +++------- 6 files changed, 120 insertions(+), 19 deletions(-) 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 2d5d3699142ad..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 @@ -67,7 +67,8 @@ public class NatsConsumer extends DefaultConsumer { 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; - private NatsJetStreamPushConsumerKeeper pushConsumerKeeper; + // set by the consuming task and read when the consumer stops + private volatile NatsJetStreamPushConsumerKeeper pushConsumerKeeper; public NatsConsumer(NatsEndpoint endpoint, Processor processor) { super(endpoint, processor); @@ -300,6 +301,10 @@ private void setupJetStreamConsumer(String topic, String queueName) throws IOExc 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 index 06cb3407229c0..d2fe8c80164a6 100644 --- 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 @@ -64,6 +64,16 @@ * 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 { 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 index f12a46342f3f2..e556a041732b2 100644 --- 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 @@ -210,6 +210,39 @@ void endsTheSubscriptionItMadeWhileTheRouteStopped() throws Exception { 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 + await().atMost(20, TimeUnit.SECONDS).until(() -> server.listenersAdded.get() == added + 1); + 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. */ @@ -487,6 +520,8 @@ static final class FakeServer { 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 @@ -599,7 +634,10 @@ Connection connection() { case "jetStreamManagement" -> jsm; case "jetStream" -> js; case "createDispatcher" -> dispatcher(); - case "addConnectionListener" -> listeners.add((ConnectionListener) args[0]); + case "addConnectionListener" -> { + listenersAdded.incrementAndGet(); + yield listeners.add((ConnectionListener) args[0]); + } case "closeDispatcher" -> { synchronized (this) { closedDispatchers.add((Dispatcher) args[0]); 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 0905b1b07d7d9..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 @@ -2171,23 +2171,11 @@ 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 `nats` consumer stayed subscribed to the deliver -subject of a JetStream consumer that the server had deleted, and consumed nothing more until the route was restarted. -After every reconnect it now asks the server for its consumer and subscribes again when its ephemeral consumer (no -`durableName`, and no durable name in the `consumerConfiguration`) is gone, which the server deletes after about 5 -seconds without a subscriber, for example while the client is disconnected, or loses on a restart. The new consumer -starts at the first message the route did not acknowledge, so the route can see again the messages after that one -(with `ackPolicy=None`, after the last message 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 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 -consumer logs a WARN and keeps its subscription, and messages arrive again when the durable is created again with the -same deliver subject. +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 From a7f30df2c4c5f095ce59a6e73772b219a1f6ad0c Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Fri, 23 Oct 2026 15:11:48 +0530 Subject: [PATCH 3/4] CAMEL-25845: camel-nats - the stop-while-subscribing test also passes once the stop closes the dispatcher removesItsConnectionListenerWhenTheRouteStopsWhileItSubscribes waited for the keeper to register its listener. Merged with CAMEL-25847 (#28320), the stop closes the dispatcher, jnats refuses the subscribe on it and no keeper is started, so the wait timed out. The test now waits for either the keeper's listener or the refused subscribe, and then checks that no keeper listens to the connection. Checked on a local merge of both branches: camel-nats 21 tests, 0 failures (the previous version of this test failed there). On this branch alone it still fails without the keeper guard of the previous commit. Co-Authored-By: Claude Opus 5.5 --- .../nats/NatsConsumerPushConsumerDeletedTest.java | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) 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 index e556a041732b2..f663f04fe437b 100644 --- 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 @@ -237,8 +237,13 @@ void removesItsConnectionListenerWhenTheRouteStopsWhileItSubscribes() throws Exc }; consumeWith("exceptionHandler=#handler"); - // the keeper of the new route registered its listener, after the stop - await().atMost(20, TimeUnit.SECONDS).until(() -> server.listenersAdded.get() == added + 1); + // 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")); } From fd79fbd0d67a6258e6cc291414250bd4d820ca11 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Fri, 23 Oct 2026 15:47:40 +0530 Subject: [PATCH 4/4] CAMEL-25845: camel-nats - a queued consumer check leaves when the consumer stops Review of #28318: the keeper runs its consumer check on the executor of the consumer, a fixed pool (poolSize, 10 by default), so a check queued by a reconnect can wait for the lock while an earlier check holds it between attempts (up to 30 seconds). The stop of the consumer calls shutdownNow on that executor, which does not wait for its threads but interrupts them; the check holding the lock leaves on the interrupt, while lock() kept the waiting one blocked until it got the lock. checkConsumer() now uses lockInterruptibly(): on an interrupt it restores the interrupt flag, clears checkQueued and returns without a check. No new test: the stop does not wait for the executor, so the only difference is how long a pool thread of a stopping consumer lingers, which no route behaviour shows. The camel-nats unit tests pass (18, among them the 13 of NatsConsumerPushConsumerDeletedTest). Co-Authored-By: Claude Opus 5.5 --- .../nats/NatsJetStreamPushConsumerKeeper.java | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) 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 index d2fe8c80164a6..25bd9a1eb984d 100644 --- 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 @@ -175,7 +175,16 @@ public void connectionEvent(Connection conn, Events type) { } void checkConsumer() { - lock.lock(); + 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);