Handle "no offset" (19) response code (query offset)

References rabbitmq/rabbitmq-server#3783
Fixes #48

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/Constants.java

@@ -34,6 +34,7 @@ public final class Constants {
   public static final short RESPONSE_CODE_ACCESS_REFUSED = 16;
   public static final short RESPONSE_CODE_PRECONDITION_FAILED = 17;
   public static final short RESPONSE_CODE_PUBLISHER_DOES_NOT_EXIST = 18;
+  public static final short RESPONSE_CODE_NO_OFFSET = 19;
 
   public static final short CODE_MESSAGE_ENQUEUEING_FAILED = 10_001;
   public static final short CODE_PRODUCER_NOT_AVAILABLE = 10_002;

src/main/java/com/rabbitmq/stream/impl/OffsetTrackingCoordinator.java

@@ -13,7 +13,9 @@
 // [email protected].
 package com.rabbitmq.stream.impl;
 
+import com.rabbitmq.stream.Constants;
 import com.rabbitmq.stream.MessageHandler.Context;
+import com.rabbitmq.stream.StreamException;
 import com.rabbitmq.stream.impl.StreamConsumerBuilder.TrackingConfiguration;
 import java.time.Duration;
 import java.util.Collection;
@@ -91,7 +93,11 @@ Registration registerTrackingConsumer(
                           }
                           Tracker t = iterator.next();
                           if (t.consumer().isOpen()) {
-                            t.flushIfNecessary();
+                            try {
+                              t.flushIfNecessary();
+                            } catch (Exception e) {
+                              LOGGER.info("Error while flushing tracker: {}", e.getMessage());
+                            }
                           } else {
                             iterator.remove();
                           }
@@ -194,10 +200,18 @@ public Consumer<Context> postProcessingCallback() {
     public void flushIfNecessary() {
       if (this.count > 0) {
         if (this.clock.time() - this.lastTrackingActivity > this.flushIntervalInNs) {
-          long lastStoredOffset = consumer.lastStoredOffset();
-          if (Long.compareUnsigned(lastStoredOffset, lastProcessedOffset) < 0) {
-            this.consumer.store(this.lastProcessedOffset);
-            this.lastTrackingActivity = clock.time();
+          try {
+            long lastStoredOffset = consumer.lastStoredOffset();
+            if (Long.compareUnsigned(lastStoredOffset, lastProcessedOffset) < 0) {
+              this.consumer.store(this.lastProcessedOffset);
+              this.lastTrackingActivity = clock.time();
+            }
+          } catch (StreamException e) {
+            if (e.getCode() == Constants.RESPONSE_CODE_NO_OFFSET) {
+              // probably nothing stored yet, let it go
+            } else {
+              throw e;
+            }
           }
         }
       }
@@ -216,15 +230,23 @@ public LongConsumer trackingCallback() {
     @Override
     public Runnable closingCallback() {
       return () -> {
-        long lastStoredOffset = consumer.lastStoredOffset();
-        if (Long.compareUnsigned(lastStoredOffset, lastProcessedOffset) < 0) {
-          LOGGER.debug("Storing offset before closing");
-          this.consumer.store(this.lastProcessedOffset);
-        } else {
-          LOGGER.debug(
-              "Not storing offset before closing because last stored offset after last processed offset: {} > {}",
-              lastStoredOffset,
-              lastProcessedOffset);
+        try {
+          long lastStoredOffset = consumer.lastStoredOffset();
+          if (Long.compareUnsigned(lastStoredOffset, lastProcessedOffset) < 0) {
+            LOGGER.debug("Storing offset before closing");
+            this.consumer.store(this.lastProcessedOffset);
+          } else {
+            LOGGER.debug(
+                "Not storing offset before closing because last stored offset after last processed offset: {} > {}",
+                lastStoredOffset,
+                lastProcessedOffset);
+          }
+        } catch (StreamException e) {
+          if (e.getCode() == Constants.RESPONSE_CODE_NO_OFFSET) {
+            // probably nothing stored yet, let it go
+          } else {
+            throw e;
+          }
         }
       };
     }
@@ -253,10 +275,18 @@ public Consumer<Context> postProcessingCallback() {
     @Override
     public void flushIfNecessary() {
       if (this.clock.time() - this.lastTrackingActivity > this.checkIntervalInNs) {
-        long lastStoredOffset = consumer.lastStoredOffset();
-        if (Long.compareUnsigned(lastStoredOffset, lastRequestedOffset) < 0) {
-          this.consumer.store(this.lastRequestedOffset);
-          this.lastTrackingActivity = clock.time();
+        try {
+          long lastStoredOffset = consumer.lastStoredOffset();
+          if (Long.compareUnsigned(lastStoredOffset, lastRequestedOffset) < 0) {
+            this.consumer.store(this.lastRequestedOffset);
+            this.lastTrackingActivity = clock.time();
+          }
+        } catch (StreamException e) {
+          if (e.getCode() == Constants.RESPONSE_CODE_NO_OFFSET) {
+            // probably nothing stored yet, let it go
+          } else {
+            throw e;
+          }
         }
       }
     }

src/test/java/com/rabbitmq/stream/impl/OffsetTrackingTest.java

@@ -16,14 +16,17 @@
 import static com.rabbitmq.stream.impl.TestUtils.b;
 import static com.rabbitmq.stream.impl.TestUtils.forEach;
 import static com.rabbitmq.stream.impl.TestUtils.latchAssert;
+import static com.rabbitmq.stream.impl.TestUtils.responseCode;
 import static com.rabbitmq.stream.impl.TestUtils.streamName;
 import static com.rabbitmq.stream.impl.TestUtils.waitAtMost;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatNoException;
 
+import com.rabbitmq.stream.Constants;
 import com.rabbitmq.stream.OffsetSpecification;
 import com.rabbitmq.stream.impl.Client.ClientParameters;
 import com.rabbitmq.stream.impl.Client.MessageListener;
+import com.rabbitmq.stream.impl.Client.QueryOffsetResponse;
 import com.rabbitmq.stream.impl.Client.Response;
 import com.rabbitmq.stream.impl.Client.StreamParametersBuilder;
 import io.vavr.Tuple;
@@ -101,6 +104,14 @@ void trackAndQuery(
     assertThat(offset).as(message).isEqualTo(expectedOffset);
   }
 
+  @Test
+  void shouldReturnNoOffsetIfNothingStoredForReference() {
+    QueryOffsetResponse response = cf.get().queryOffset(UUID.randomUUID().toString(), stream);
+    assertThat(response.isOk()).isFalse();
+    assertThat(response.getResponseCode()).isEqualTo(Constants.RESPONSE_CODE_NO_OFFSET);
+    assertThat(response.getOffset()).isEqualTo(0);
+  }
+
   @ParameterizedTest
   @MethodSource
   void consumeAndStore(BiConsumer<String, Client> streamCreator, TestInfo info) throws Exception {