Merge pull request #1454 from nats-io/pull-status-warnings-1194

Demonstrate Pull Status Warning Handling

Author: scottf

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

@@ -186,6 +186,9 @@ protected ManageResult manageStatus(Message msg) {
                     || statMsg.equals(LEADERSHIP_CHANGE)
                     || statMsg.equals(MESSAGE_SIZE_EXCEEDS_MAX_BYTES))
                 {
+                    if (raiseStatusWarnings) {
+                        conn.executeCallback((c, el) -> el.pullStatusWarning(c, sub, status));
+                    }
                     return STATUS_TERMINUS;
                 }
                 break;

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

@@ -19,7 +19,7 @@
 import static io.nats.client.support.NatsConstants.*;
 import static io.nats.client.support.NatsJetStreamConstants.*;
 import static io.nats.client.support.Status.*;
-import static java.nio.charset.StandardCharsets.US_ASCII;
+import static java.nio.charset.StandardCharsets.UTF_8;
 
 public class Token {
     private final byte[] serialized;
@@ -117,7 +117,7 @@ public String getValueOrNull() {
     }
 
     private String valueAsString() {
-        return new String(serialized, start, valueLength, US_ASCII).trim();
+        return new String(serialized, start, valueLength, UTF_8).trim();
     }
 
     @NonNull
@@ -274,7 +274,6 @@ private boolean valueStartsWithExceededMax() {
         return true;
     }
 
-
     private boolean endsMatch(byte @NonNull [] checkBytes, int compareStartIndex) {
         if (valueLength != checkBytes.length) {
             return false;

src/test/java/io/nats/client/impl/JetStreamConsumerTests.java

@@ -410,4 +410,51 @@ private static void validateMultipleSubjectFilterSub(JetStreamSubscription sub,
         assertEquals(5, count2);
     }
 
+    @Test
+    public void testRaiseStatusWarnings1194() throws Exception {
+        ListenerForTesting listenerForTesting = new ListenerForTesting(true, true);
+        runInJsServer(listenerForTesting, nc -> {
+            // Setup
+            JetStreamManagement jsm = nc.jetStreamManagement();
+            TestingStreamContainer tsc = new TestingStreamContainer(jsm);
+            StreamContext streamContext = nc.getStreamContext(tsc.stream);
+
+            // Setting maxBatch=1, so we shouldn't allow fetching more messages at once.
+            ConsumerConfiguration consumerConfig = ConsumerConfiguration.builder().filterSubject(tsc.subject()).maxBatch(1).build();
+            ConsumerContext consumerContext = streamContext.createOrUpdateConsumer(consumerConfig);
+
+            int count = 0;
+
+            // Fetching a batch of 100 messages is not allowed, so we rightfully don't get any messages and wait for timeout.
+            // But we don't get informed about the status message.
+            FetchConsumeOptions fco = FetchConsumeOptions.builder()
+                .maxMessages(100)
+                .expiresIn(1000)
+                .build();
+            try (FetchConsumer fetchConsumer = consumerContext.fetch(fco)) {
+                Message msg;
+                while ((msg = fetchConsumer.nextMessage()) != null) {
+                    msg.ack();
+                    count++;
+                }
+            }
+            assertEquals(0, count);
+            assertEquals(0, listenerForTesting.getPullStatusWarnings().size());
+
+            fco = FetchConsumeOptions.builder()
+                .maxMessages(100)
+                .expiresIn(1000)
+                .raiseStatusWarnings()
+                .build();
+            try (FetchConsumer fetchConsumer = consumerContext.fetch(fco)) {
+                Message msg;
+                while ((msg = fetchConsumer.nextMessage()) != null) {
+                    msg.ack();
+                    count++;
+                }
+            }
+            assertEquals(0, count);
+            assertEquals(1, listenerForTesting.getPullStatusWarnings().size());
+        });
+    }
 }

src/test/java/io/nats/client/impl/MessageManagerTests.java

@@ -185,8 +185,8 @@ private void testPullBqpAndManage(NatsJetStreamSubscription sub, ListenerForTest
         sleep(100);
 
         List<ListenerForTesting.StatusEvent> list = listener.getPullStatusWarnings();
-        int[] codes = new int[]{NOT_FOUND_CODE, REQUEST_TIMEOUT_CODE, CONFLICT_CODE, CONFLICT_CODE, CONFLICT_CODE, CONFLICT_CODE};
-        assertEquals(6, list.size());
+        int[] codes = new int[]{NOT_FOUND_CODE, REQUEST_TIMEOUT_CODE, CONFLICT_CODE, CONFLICT_CODE, CONFLICT_CODE, CONFLICT_CODE, CONFLICT_CODE, CONFLICT_CODE};
+        assertEquals(8, list.size());
         for (int x = 0; x < list.size(); x++) {
             ListenerForTesting.StatusEvent se = list.get(x);
             assertSame(sub.getSID(), se.sid);