Merge pull request #1313 from nats-io/leadership-change

Better workflow for leadership change while simplified consuming.

Author: scottf

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

@@ -130,19 +130,20 @@ protected Message _nextUnmanagedNoWait(String expectedPullSubject) throws Interr
                 case MESSAGE:
                     return msg;
                 case STATUS_TERMINUS:
-                    // if the status applies return null, otherwise it's ignored, fall through
+                    // if the status applies, return null, otherwise it's ignored, fall through
                     if (expectedPullSubject == null || expectedPullSubject.equals(msg.getSubject())) {
                         return null;
                     }
                     break;
                 case STATUS_ERROR:
-                    // if the status applies throw exception, otherwise it's ignored, fall through
+                    // if the status applies, throw exception, otherwise it's ignored, fall through
                     if (expectedPullSubject == null || expectedPullSubject.equals(msg.getSubject())) {
                         throw new JetStreamStatusException(msg.getStatus(), this);
                     }
                     break;
             }
-            // Check again when, regular messages might have arrived
+            // These statuses don't apply to the message that came in,
+            // so we just loop and move on to the next message.
             // 1. Any STATUS_HANDLED
             // 2. STATUS_TERMINUS or STATUS_ERRORS that aren't for expected pullSubject
         }

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

@@ -53,7 +53,7 @@ protected void startup(NatsJetStreamSubscription sub) {
     @Override
     protected ManageResult manage(Message msg) {
         if (!msg.getSID().equals(targetSid.get())) {
-            return STATUS_HANDLED; // wrong sid is throwaway from previous consumer that errored
+            return STATUS_HANDLED; // wrong sid. message is a throwaway from previous consumer that errored
         }
 
         if (msg.isJetStream()) {
@@ -90,14 +90,14 @@ private void handleErrorCondition() {
             }
             catch (Exception ignore) {}
 
-            // 2. re-subscribe. This means kill the sub then make a new one
+            // 2. re-subscribe. This means killing the sub then making a new one.
             //    New sub needs a new deliverSubject
             String newDeliverSubject = sub.connection.createInbox();
             sub.reSubscribe(newDeliverSubject);
             targetSid.set(sub.getSID());
 
-            // 3. make a new consumer using the same deliver subject but
-            //    with a new starting point
+            // 3. make a new consumer using the same "deliver" subject
+            //    but with a new starting point
             ConsumerConfiguration userCC = js.consumerConfigurationForOrdered(originalCc, lastStreamSeq, newDeliverSubject, actualConsumerName, null);
             ConsumerInfo ci = js._createConsumer(stream, userCC, ConsumerCreateRequest.Action.Create); // this can fail when a server is down.
             sub.setConsumerName(ci.getName());

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

@@ -162,18 +162,17 @@ protected ManageResult manageStatus(Message msg) {
             case CONFLICT_CODE:
                 // sometimes just a warning
                 String statMsg = status.getMessage();
-                if (statMsg.startsWith("Exceeded Max")
-                    || statMsg.equals(SERVER_SHUTDOWN)
-                    || statMsg.equals(LEADERSHIP_CHANGE)
-                ) {
+                if (statMsg.startsWith(EXCEEDED_MAX_PREFIX) || statMsg.equals(SERVER_SHUTDOWN))
+                {
                     if (raiseStatusWarnings) {
                         conn.executeCallback((c, el) -> el.pullStatusWarning(c, sub, status));
                     }
                     return STATUS_HANDLED;
                 }
 
-                if (statMsg.equals(BATCH_COMPLETED) ||
-                    statMsg.equals(MESSAGE_SIZE_EXCEEDS_MAX_BYTES))
+                if (statMsg.equals(BATCH_COMPLETED)
+                    || statMsg.equals(LEADERSHIP_CHANGE)
+                    || statMsg.equals(MESSAGE_SIZE_EXCEEDS_MAX_BYTES))
                 {
                     return STATUS_TERMINUS;
                 }

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

@@ -51,7 +51,7 @@ protected void startup(NatsJetStreamSubscription sub) {
     @Override
     protected ManageResult manage(Message msg) {
         if (!msg.getSID().equals(targetSid.get())) {
-            return STATUS_HANDLED; // wrong sid is throwaway from previous consumer that errored
+            return STATUS_HANDLED; // wrong sid. message is a throwaway from previous consumer that errored
         }
 
         if (msg.isJetStream()) {

src/main/java/io/nats/client/support/Status.java

@@ -37,6 +37,7 @@ public class Status {
     public static String CONSUMER_IS_PUSH_BASED         = "Consumer is push based"; // 409
 
     public static String MESSAGE_SIZE_EXCEEDS_MAX_BYTES = "Message Size Exceeds MaxBytes"; // 409
+    public static String EXCEEDED_MAX_PREFIX            = "Exceeded Max";
     public static String EXCEEDED_MAX_WAITING           = "Exceeded MaxWaiting"; // 409
     public static String EXCEEDED_MAX_REQUEST_BATCH     = "Exceeded MaxRequestBatch"; // 409
     public static String EXCEEDED_MAX_REQUEST_EXPIRES   = "Exceeded MaxRequestExpires"; // 409