statuses with code 409 are known statuses (#643)

nats-io · May 17, 2022 · 17dc90d · 17dc90d
1 parent b87bfff
commit 17dc90d
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 11 deletions.
diff --git a/src/main/java/io/nats/client/impl/PullStatusMessageManager.java b/src/main/java/io/nats/client/impl/PullStatusMessageManager.java
@@ -21,7 +21,7 @@
 
 class PullStatusMessageManager extends MessageManager {
 
-    private static final List<Integer> PULL_KNOWN_STATUS_CODES = Arrays.asList(404, 408);
+    private static final List<Integer> PULL_KNOWN_STATUS_CODES = Arrays.asList(404, 408, 409);
 
     private int lastStatusCode = -1;
 

diff --git a/src/main/java/io/nats/client/impl/PushStatusMessageManager.java b/src/main/java/io/nats/client/impl/PushStatusMessageManager.java
@@ -20,6 +20,8 @@
 import io.nats.client.api.ConsumerConfiguration;
 import io.nats.client.support.Status;
 
+import java.util.Arrays;
+import java.util.List;
 import java.util.Timer;
 import java.util.TimerTask;
 import java.util.concurrent.atomic.AtomicLong;
@@ -28,6 +30,8 @@
 
 class PushStatusMessageManager extends MessageManager {
 
+    private static final List<Integer> PUSH_KNOWN_STATUS_CODES = Arrays.asList(409);
+
     private static final int THRESHOLD = 3;
 
     private final NatsConnection conn;
@@ -175,22 +179,20 @@ boolean manage(Message msg) {
                 if (fc) {
                     _processFlowControl(msg.getReplyTo(), ErrorListener.FlowControlSource.FLOW_CONTROL);
                 }
-                return true;
             }
-
-            if (status.isHeartbeat()) {
+            else if (status.isHeartbeat()) {
                 if (fc) {
                     // status flowControlSubject is set in the beforeQueueProcessor
                     _processFlowControl(extractFcSubject(msg), ErrorListener.FlowControlSource.HEARTBEAT);
                 }
-                return true;
             }
-
-            // this status is unknown to us, always use the error handler.
-            // If it's a sync call, also throw an exception
-            conn.getOptions().getErrorListener().unhandledStatus(conn, sub, status);
-            if (syncMode) {
-                throw new JetStreamStatusException(sub, status);
+            else if (!PUSH_KNOWN_STATUS_CODES.contains(status.getCode())) {
+                // If this status is unknown to us, always use the error handler.
+                // If it's a sync call, also throw an exception
+                conn.getOptions().getErrorListener().unhandledStatus(conn, sub, status);
+                if (syncMode) {
+                    throw new JetStreamStatusException(sub, status);
+                }
             }
             return true;
         }