Merge pull request #881 from rabbitmq/chunk-id-in-flow-strategy-context

Add chunk ID to flow strategy context

Author: acogoluegnes

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

@@ -67,7 +67,8 @@ interface Context {
     /**
      * Provide credits for the subscription.
      *
-     * <p>{@link ConsumerFlowStrategy} implementation should always provide 1 credit a given chunk.
+     * <p>{@link ConsumerFlowStrategy} implementation should always provide 1 credit for a given
+     * chunk.
      *
      * @param credits the number of credits provided, usually 1
      */
@@ -79,6 +80,13 @@ interface Context {
      * @return number of messages in the chunk
      */
     long messageCount();
+
+    /**
+     * The offset of the first message in the chunk, aka chunk ID.
+     *
+     * @return offset of the first message in the chunk (chunk ID)
+     */
+    long chunkId();
   }
 
   /** Behavior for {@link MessageHandler.Context#processed()} calls. */

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

@@ -665,7 +665,8 @@ private ClientSubscriptionsManager(
             if (subscriptionTracker != null && subscriptionTracker.consumer.isOpen()) {
               processCallback =
                   subscriptionTracker.flowStrategy.start(
-                      new DefaultConsumerFlowStrategyContext(subscriptionId, client, messageCount));
+                      new DefaultConsumerFlowStrategyContext(
+                          subscriptionId, client, messageCount, offset));
             } else {
               LOGGER.debug(
                   "Could not find stream subscription {} or subscription closing, not providing credits",
@@ -1391,12 +1392,14 @@ private static class DefaultConsumerFlowStrategyContext implements ConsumerFlowS
     private final byte subscriptionId;
     private final Client client;
     private final long messageCount;
+    private final long chunkId;
 
     private DefaultConsumerFlowStrategyContext(
-        byte subscriptionId, Client client, long messageCount) {
+        byte subscriptionId, Client client, long messageCount, long chunkId) {
       this.subscriptionId = subscriptionId;
       this.client = client;
       this.messageCount = messageCount;
+      this.chunkId = chunkId;
     }
 
     @Override
@@ -1414,7 +1417,12 @@ public void credits(int credits) {
 
     @Override
     public long messageCount() {
-      return messageCount;
+      return this.messageCount;
+    }
+
+    @Override
+    public long chunkId() {
+      return this.chunkId;
     }
   }
 

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

@@ -66,6 +66,11 @@ public void credits(int credits) {
       public long messageCount() {
         throw new UnsupportedOperationException();
       }
+
+      @Override
+      public long chunkId() {
+        throw new UnsupportedOperationException();
+      }
     };
   }
 }

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

@@ -70,6 +70,11 @@ public void credits(int credits) {
       public long messageCount() {
         return messageCount;
       }
+
+      @Override
+      public long chunkId() {
+        return 0;
+      }
     };
   }
 }