CAMEL-25845: camel-nats - the JetStream push consumer subscribes again when the server deleted its ephemeral consumer

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.

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.

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);
             }