Handle no offset response in default consumer update handler

References #46, rabbitmq/rabbitmq-server#3753

Author: acogoluegnes

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

@@ -56,8 +56,7 @@ class ConsumersCoordinator {
 
   static final int MAX_SUBSCRIPTIONS_PER_CLIENT = 256;
 
-  private static final OffsetSpecification DEFAULT_OFFSET_SPECIFICATION =
-      OffsetSpecification.next();
+  static final OffsetSpecification DEFAULT_OFFSET_SPECIFICATION = OffsetSpecification.next();
 
   private static final Logger LOGGER = LoggerFactory.getLogger(ConsumersCoordinator.class);
   private final Random random = new Random();

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

@@ -60,6 +60,7 @@ class StreamConsumer implements Consumer {
   private volatile Status status;
   private volatile long lastRequestedStoredOffset = 0;
   private volatile ConsumerUpdateListener.Status sacStatus;
+  private final OffsetSpecification initialOffsetSpecification;
 
   StreamConsumer(
       String stream,
@@ -79,6 +80,10 @@ class StreamConsumer implements Consumer {
       this.name = name;
       this.stream = stream;
       this.environment = environment;
+      this.initialOffsetSpecification =
+          offsetSpecification == null
+              ? ConsumersCoordinator.DEFAULT_OFFSET_SPECIFICATION
+              : offsetSpecification;
 
       AtomicReference<MessageHandler> decoratedMessageHandler = new AtomicReference<>();
       LongSupplier trackingFlushCallback;
@@ -102,7 +107,7 @@ class StreamConsumer implements Consumer {
         }
 
         this.trackingCallback = trackingConsumerRegistration.trackingCallback();
-        trackingFlushCallback = () -> trackingConsumerRegistration.flush();
+        trackingFlushCallback = trackingConsumerRegistration::flush;
       } else {
         trackingClosingCallback = () -> {};
         this.trackingCallback = Utils.NO_OP_LONG_CONSUMER;
@@ -132,11 +137,23 @@ class StreamConsumer implements Consumer {
                           || context.previousStatus() == ConsumerUpdateListener.Status.PASSIVE)
                       && context.status() == ConsumerUpdateListener.Status.ACTIVE) {
                     LOGGER.debug("Looking up offset (stream {})", this.stream);
-                    StreamConsumer consumer = (StreamConsumer) context.consumer();
-                    long offset = consumer.storedOffset();
-                    LOGGER.debug(
-                        "Stored offset is {}, returning the value + 1 to the server", offset);
-                    return OffsetSpecification.offset(offset + 1);
+                    Consumer consumer = context.consumer();
+                    try {
+                      long offset = consumer.storedOffset();
+                      LOGGER.debug(
+                          "Stored offset is {}, returning the value + 1 to the server", offset);
+                      result = OffsetSpecification.offset(offset + 1);
+                    } catch (StreamException e) {
+                      if (e.getCode() == Constants.RESPONSE_CODE_NO_OFFSET) {
+                        LOGGER.debug(
+                            "No stored offset, using initial offset specification: {}",
+                            this.initialOffsetSpecification);
+                        result = initialOffsetSpecification;
+                      } else {
+                        throw e;
+                      }
+                    }
+                    return result;
                   } else if (context.previousStatus() == ConsumerUpdateListener.Status.ACTIVE
                       && context.status() == ConsumerUpdateListener.Status.PASSIVE) {
                     LOGGER.debug(
@@ -262,9 +279,7 @@ void waitForOffsetToBeStored(long expectedStoredOffset) {
           "Offset {} stored (consumer {}, stream {})", expectedStoredOffset, this.id, this.stream);
     } catch (InterruptedException e) {
       Thread.currentThread().interrupt();
-    } catch (ExecutionException e) {
-      LOGGER.warn("Error while checking offset has been stored", e);
-    } catch (TimeoutException e) {
+    } catch (ExecutionException | TimeoutException e) {
       LOGGER.warn("Error while checking offset has been stored", e);
     }
   }
@@ -449,8 +464,8 @@ public long storedOffset() {
       } else {
         throw new StreamException(
             String.format(
-                "QueryOffset for consumer %s on stream %s returned an error",
-                this.name, this.stream),
+                "QueryOffset for consumer %s on stream %s returned an error (%s)",
+                this.name, this.stream, Utils.formatConstant(response.getResponseCode())),
             response.getResponseCode());
       }