Merge branch 'main' into kv-purge-per-message-ttl

Author: scottf

src/examples/java/io/nats/examples/jetstream/simple/OrderedMessageConsumerExample.java

@@ -0,0 +1,155 @@
+// Copyright 2025 The NATS Authors
+// Licensed 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 io.nats.examples.jetstream.simple;
+
+import io.nats.client.*;
+import io.nats.client.api.OrderedConsumerConfiguration;
+import io.nats.client.api.StorageType;
+import io.nats.client.impl.ErrorListenerConsoleImpl;
+import io.nats.examples.jetstream.ResilientPublisher;
+
+import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static io.nats.examples.jetstream.NatsJsUtils.createOrReplaceStream;
+
+/**
+ * This example will demonstrate simplified ordered consumer with a handler
+ * To run, start a server and then start the example.
+ * To test resiliency, kill the server, wait a couple seconds, then restart it.
+ */
+public class OrderedMessageConsumerExample {
+    private static final String STREAM = "ordered-stream";
+    private static final String SUBJECT = "ordered-subject";
+    private static final String CONSUMER_PREFIX = "prefix";
+    private static final String MESSAGE_PREFIX = "ordered";
+    private static final int STOP_COUNT = 1_000_000;
+    private static final int REPORT_EVERY = 500;
+
+    private static final String SERVER = "nats://localhost:4222";
+
+    public static void main(String[] args) {
+        Options options = Options.builder()
+            .server(SERVER)
+            .errorListener(new ErrorListenerConsoleImpl())
+            .build();
+        try (Connection nc = Nats.connect(options)) {
+            JetStreamManagement jsm = nc.jetStreamManagement();
+            createOrReplaceStream(jsm, STREAM, StorageType.File, SUBJECT); // file is important, memory won't survive a restart
+
+            System.out.println("Starting publish...");
+            ResilientPublisher publisher = new ResilientPublisher(nc, jsm, STREAM, SUBJECT).basicDataPrefix(MESSAGE_PREFIX).jitter(10);
+            Thread pubThread = new Thread(publisher);
+            pubThread.start();
+
+            // get stream context, create consumer and get the consumer context
+            StreamContext streamContext;
+            OrderedConsumerContext orderedConsumerContext;
+            try {
+                OrderedConsumerConfiguration ocConfig = new OrderedConsumerConfiguration()
+                    .consumerNamePrefix(CONSUMER_PREFIX)
+                    .filterSubjects(SUBJECT);
+                streamContext = nc.getStreamContext(STREAM);
+                orderedConsumerContext = streamContext.createOrderedConsumer(ocConfig);
+            }
+            catch (JetStreamApiException | IOException e) {
+                // JetStreamApiException:
+                //      the stream or consumer did not exist
+                // IOException:
+                //      likely a connection problem
+                return;
+            }
+
+            CountDownLatch latch = new CountDownLatch(1);
+            AtomicInteger atomicCount = new AtomicInteger();
+            AtomicLong nextExpectedSequence = new AtomicLong(0);
+            long start = System.nanoTime();
+            MessageHandler handler = msg -> {
+                if (msg.metaData().streamSequence() != nextExpectedSequence.incrementAndGet()) {
+                    System.out.println("MESSAGE RECEIVED OUT OF ORDER!");
+                    System.exit(-1);
+                }
+                msg.ack();
+                int count = atomicCount.incrementAndGet();
+                if (count % REPORT_EVERY == 0) {
+                    report("Handler", start, count);
+                }
+                if (count == STOP_COUNT) {
+                    latch.countDown();
+                }
+            };
+
+            // create the consumer then use it
+            // the expires option affects 2 things
+            // 1. A pull request expiry
+            // 2. The heartbeat checking
+            // The default expiry is 30 seconds which means the idle heartbeat is 15 seconds.
+            // It takes 3 times the heartbeat to trip the alarm, so the subscription does
+            // not restart for 45 seconds since the disconnect.
+            // If your messages come in slowly this is fine, but if your messages are coming in fast
+            // set the expiresIn much lower. Minimum expires is 1 second (1000ms)
+            ConsumeOptions consumeOptions = ConsumeOptions.builder()
+                .expiresIn(ConsumeOptions.MIN_EXPIRES_MILLS)
+                .build();
+            try (MessageConsumer consumer = orderedConsumerContext.consume(consumeOptions, handler)) {
+                latch.await();
+
+                // The consumer has at least 1 pull request active. When stop is called,
+                // no more pull requests will be made, but messages already requested
+                // will still come across the wire to the client.
+                System.out.println("Stop the consumer...");
+                consumer.stop();
+
+                // wait until the consumer is finished
+                while (!consumer.isFinished()) {
+                    //noinspection BusyWait
+                    Thread.sleep(10);
+                }
+            }
+            catch (JetStreamApiException | IOException e) {
+                // JetStreamApiException:
+                //      1. the stream or consumer did not exist
+                //      2. api calls under the covers theoretically this could fail, but practically it won't.
+                // IOException:
+                //      likely a connection problem
+                System.err.println("Exception should be handled properly, just exiting here.");
+                System.exit(-1);
+            }
+            catch (Exception e) {
+                // this is from the FetchConsumer being AutoCloseable, but should never be called
+                // as work inside the close is already guarded by try/catch
+                System.err.println("Exception should be handled properly, just exiting here.");
+                System.exit(-1);
+            }
+
+            report("Final", start, atomicCount.get());
+
+            publisher.stop(); // otherwise it will complain when the connection goes away
+            pubThread.join();
+        }
+        catch (IOException | InterruptedException ioe) {
+            // IOException:
+            //      problem making the connection
+            // InterruptedException:
+            //      thread interruption in the body of the example
+        }
+    }
+
+    private static void report(String label, long start, int count) {
+        long ms = (System.nanoTime() - start) / 1_000_000;
+        System.out.println(label + ": Received " + count + " messages in " + ms + "ms.");
+    }
+}

src/main/java/io/nats/client/impl/MessageManager.java

@@ -47,7 +47,7 @@ public enum ManageResult {MESSAGE, STATUS_HANDLED, STATUS_TERMINUS, STATUS_ERROR
     protected final AtomicBoolean hb;
     protected final AtomicLong idleHeartbeatSettingMillis;
     protected final AtomicLong alarmPeriodSettingNanos;
-    protected final AtomicReference<ScheduledTask> heartbeatTask;
+    protected final AtomicReference<ScheduledTask> heartbeatTaskRef;
 
     protected MessageManager(NatsConnection conn, SubscribeOptions so, boolean syncMode) {
         stateChangeLock = new ReentrantLock();
@@ -62,7 +62,7 @@ protected MessageManager(NatsConnection conn, SubscribeOptions so, boolean syncM
         idleHeartbeatSettingMillis = new AtomicLong();
         alarmPeriodSettingNanos = new AtomicLong();
         lastMsgReceivedNanoTime = new AtomicLong(NatsSystemClock.nanoTime());
-        heartbeatTask = new AtomicReference<>();
+        heartbeatTaskRef = new AtomicReference<>();
     }
 
     protected boolean isSyncMode()              { return syncMode; }
@@ -139,20 +139,25 @@ protected void updateLastMessageReceived() {
     protected void initOrResetHeartbeatTimer() {
         stateChangeLock.lock();
         try {
-            ScheduledTask hbTask = heartbeatTask.get();
+            ScheduledTask hbTask = heartbeatTaskRef.get();
             if (hbTask != null) {
+                // we always want a fresh schedule because it will have the initial delay
                 hbTask.shutdown();
             }
 
-            // replacement or new comes here
-            heartbeatTask.set(new ScheduledTask(conn.getScheduledExecutor(), alarmPeriodSettingNanos.get(), TimeUnit.NANOSECONDS,
-                () -> {
-                    long sinceLast = NatsSystemClock.nanoTime() - lastMsgReceivedNanoTime.get();
-                    if (sinceLast > alarmPeriodSettingNanos.get()) {
-                        shutdownHeartbeatTimer(); // a new one will get started when needed.
-                        handleHeartbeatError();
-                    }
-                }));
+            // set the ref with a new ScheduledTask
+            // reminder that ScheduledTask schedules itself, which is why we pass the executor
+            heartbeatTaskRef.set(
+                new ScheduledTask(conn.getScheduledExecutor(), alarmPeriodSettingNanos.get(), TimeUnit.NANOSECONDS,
+                    () -> {
+                        long sinceLast = NatsSystemClock.nanoTime() - lastMsgReceivedNanoTime.get();
+                        if (sinceLast > alarmPeriodSettingNanos.get()) {
+                            handleHeartbeatError();
+                        }
+                    })
+            );
+
+            // since we just scheduled, reset this otherwise it may alarm too soon
             updateLastMessageReceived();
         }
         finally {
@@ -163,10 +168,10 @@ protected void initOrResetHeartbeatTimer() {
     protected void shutdownHeartbeatTimer() {
         stateChangeLock.lock();
         try {
-            ScheduledTask hbTask = heartbeatTask.get();
+            ScheduledTask hbTask = heartbeatTaskRef.get();
             if (hbTask != null) {
                 hbTask.shutdown();
-                heartbeatTask.set(null);
+                heartbeatTaskRef.set(null);
             }
         }
         finally {

src/main/java/io/nats/client/impl/NatsConsumerContext.java

@@ -129,7 +129,7 @@ private void checkState() throws IOException {
                 }
             }
             if (lastCon.finished.get() && !lastCon.stopped.get()) {
-                lastCon.lenientClose(); // finished, might as well make sure the sub is closed.
+                lastCon.shutdownSub(); // finished, might as well make sure the sub is closed.
             }
         }
     }

src/main/java/io/nats/client/impl/NatsFetchConsumer.java

@@ -67,7 +67,7 @@ public void pendingUpdated() {}
 
     @Override
     public void heartbeatError() {
-        finishAndClose();
+        fullClose();
     }
 
     @Override
@@ -85,7 +85,7 @@ public Message nextMessage() throws InterruptedException, JetStreamStatusChecked
                 if (m == null) {
                     // if there are no messages in the internal cache AND there are no more pending,
                     // they all have been read and we can go ahead and finish
-                    finishAndClose();
+                    fullClose();
                 }
                 return m;
             }
@@ -103,15 +103,15 @@ public Message nextMessage() throws InterruptedException, JetStreamStatusChecked
                 Message m = sub._nextUnmanagedNoWait(pullSubject);
                 if (m == null) {
                     // no message and no time left, go ahead and finish
-                    finishAndClose();
+                    fullClose();
                 }
                 return m;
             }
 
             Message m = sub._nextUnmanaged(timeLeftNanos, pullSubject);
             if (m == null && isNoWaitNoExpires) {
                 // no message and no wait, go ahead and finish
-                finishAndClose();
+                fullClose();
             }
             return m;
         }

src/main/java/io/nats/client/impl/NatsMessageConsumer.java

@@ -52,40 +52,47 @@ class NatsMessageConsumer extends NatsMessageConsumerBase implements PullManager
     @Override
     public void heartbeatError() {
         try {
-            // just close the current sub and make another one.
-            // this could go on endlessly - unless the user had called stop
             if (stopped.get()) {
-                finishAndClose();
+                fullClose();
             }
             else {
-                lenientClose();
+                shutdownSub();
                 doSub();
             }
         }
         catch (JetStreamApiException | IOException e) {
-            pmm.resetTracking();
-            pmm.initOrResetHeartbeatTimer();
+            setupHbAlarmToTrigger();
         }
     }
 
     void doSub() throws JetStreamApiException, IOException {
         MessageHandler mh = userMessageHandler == null ? null : msg -> {
             userMessageHandler.onMessage(msg);
             if (stopped.get() && pmm.noMorePending()) {
-                finishAndClose();
+                finished.set(true);
             }
         };
-        super.initSub(subscriptionMaker.subscribe(mh, userDispatcher, pmm, null));
-        repull();
-        stopped.set(false);
-        finished.set(false);
+        try {
+            stopped.set(false);
+            finished.set(false);
+            super.initSub(subscriptionMaker.subscribe(mh, userDispatcher, pmm, null));
+            repull();
+        }
+        catch (JetStreamApiException | IOException e) {
+            setupHbAlarmToTrigger();
+        }
+    }
+
+    private void setupHbAlarmToTrigger() {
+        pmm.resetTracking();
+        pmm.initOrResetHeartbeatTimer();
     }
 
     @Override
     public void pendingUpdated() {
         if (stopped.get()) {
             if (pmm.noMorePending()) {
-                finishAndClose();
+                fullClose();
             }
         }
         else if (pmm.pendingMessages <= thresholdMessages || (pmm.trackingBytes && pmm.pendingBytes <= thresholdBytes))

src/main/java/io/nats/client/impl/NatsMessageConsumerBase.java

@@ -101,21 +101,19 @@ public void stop() {
 
     @Override
     public void close() throws Exception {
-        lenientClose();
+        stopped.set(true);
+        shutdownSub();
     }
 
-    protected void finishAndClose() {
-        if (pmm != null) {
-            pmm.shutdownHeartbeatTimer();
-        }
+    protected void fullClose() {
+        stopped.set(true);
         finished.set(true);
-        lenientClose();
+        shutdownSub();
     }
 
-    protected void lenientClose() {
+    protected void shutdownSub() {
         try {
-            if (!stopped.get() || sub.isActive()) {
-                stopped.set(true);
+            if (sub.isActive()) {
                 if (sub.getNatsDispatcher() != null) {
                     sub.getDispatcher().unsubscribe(sub);
                 }
@@ -127,5 +125,13 @@ protected void lenientClose() {
         catch (Throwable ignore) {
             // nothing to do
         }
+        if (pmm != null) {
+            try {
+                pmm.shutdownHeartbeatTimer();
+            }
+            catch (Throwable ignore) {
+                // nothing to do
+            }
+        }
     }
 }

src/main/java/io/nats/client/impl/PullOrderedMessageManager.java

@@ -44,6 +44,7 @@ protected PullOrderedMessageManager(NatsConnection conn,
 
     @Override
     protected void startup(NatsJetStreamSubscription sub) {
+        expectedExternalConsumerSeq = 1; // consumer always starts with consumer sequence 1
         super.startup(sub);
         targetSid.set(sub.getSID());
     }