Byte-based flow control

.github/workflows/test-pr.yml

@@ -30,6 +30,8 @@ jobs:
           cache: 'maven'
       - name: Start broker
         run: ci/start-broker.sh
+        env:
+          RABBITMQ_IMAGE: pivotalrabbitmq/rabbitmq:pr-17185-otp28
       - name: Test
         run: |
           ./mvnw test -Drabbitmqctl.bin=DOCKER:rabbitmq \

src/docs/asciidoc/api.adoc

@@ -888,6 +888,7 @@ Useful when using an external store for offset tracking.
 |`flow#initialCredits`
 |Number of credits when the subscription is created.
 Increase for higher throughput at the expense of memory usage.
+Accepts a `ByteCapacity` instead of an `int` to use a <<byte-based-flow-control,byte window>> rather than a number of chunks (requires broker support).
 |10
 
 |`flow#strategy`
@@ -1184,6 +1185,46 @@ No calling it will stop the dispatching of messages.
 Whether the method is idempotent depends on the flow strategy implementation.
 Apart from the default one, the implementations the library provides does not make `processed()` idempotent.
 
+[[byte-based-flow-control]]
+===== Byte-Based Flow Control
+
+WARNING: Byte-based flow control requires a broker new enough to support the corresponding
+version of the `Subscribe` and `Credit` commands.
+The client fails fast with an exception when creating a consumer with a byte-based strategy
+against a broker that does not support it, it does not fall back to chunk-based flow control.
+
+The credit unit for the strategies described above is the chunk: 1 credit lets the broker send 1
+more chunk. Chunks can vary widely in size, so a chunk-based window does not directly translate
+into a memory bound on the consumer side. `creditOnChunkArrival`, `creditWhenHalfMessagesProcessed`,
+and `creditOnProcessedMessageCount` all accept a `ByteCapacity` window instead of a number of
+chunks, so the client can express the same arrival-versus-feedback trade-off in terms of bytes:
+
+.Setting a byte-based consumer flow control strategy
+[source,java,indent=0]
+--------
+include::{test-examples}/ConsumerUsage.java[tag=flow-control-byte-based]
+--------
+<1> Set a byte-based flow control strategy with a 512 KB window
+<2> Make sure to call `Context#processed()`
+
+* `creditOnChunkArrival(ByteCapacity)` grants the whole window back as soon as a chunk arrives, so
+the number of bytes in flight is bounded, but the number of bytes received and not yet processed
+is unbounded if processing falls behind.
+* `creditWhenHalfMessagesProcessed(ByteCapacity)` and `creditOnProcessedMessageCount(ByteCapacity,
+double)` grant credit as messages are processed, so the number of bytes received and not yet
+processed stays bounded by roughly the window plus the size of the largest chunk received so far.
+This is the feedback property that distinguishes them from `creditOnChunkArrival`.
+* the broker may deliver a chunk larger than the window: a consumer never stalls just because its
+window is smaller than the next chunk, it only ever falls behind by that one chunk.
+* a byte-based strategy must eventually grant credit for every byte the broker charged for a
+chunk (`Context#chunkByteCount()`), whether on the chunk's arrival or on the processing of its
+messages. Granting less would make the subscription's credit drift down and eventually stall it.
+This only matters for a custom `ConsumerFlowStrategy`, the strategies described above already do
+it.
+* the broker resumes sending only once about half the window has been granted back as credit, so
+size the window to roughly twice the amount of data the application is willing to hold in memory
+at once.
+
 [[single-active-consumer]]
 ==== Single Active Consumer
 

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2025 Broadcom. All Rights Reserved.
+// Copyright (c) 2020-2026 Broadcom. All Rights Reserved.
 // The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
@@ -274,6 +274,23 @@ interface FlowConfiguration {
      */
     FlowConfiguration initialCredits(int initialCredits);
 
+    /**
+     * The initial credits for the subscription, as a window in bytes.
+     *
+     * <p>This calls uses {@link ConsumerFlowStrategy#creditOnChunkArrival(ByteCapacity)}.
+     *
+     * <p>The broker may exceed the window by up to one chunk: a chunk larger than the window is
+     * delivered anyway, so that a consumer with a small window still makes forward progress.
+     *
+     * <p>Requires a broker supporting {@code Subscribe} version 2.
+     *
+     * @param initialCredits the initial credit window, in bytes
+     * @return this configuration instance
+     * @since 1.11.0
+     * @see ConsumerFlowStrategy#creditOnChunkArrival(ByteCapacity)
+     */
+    FlowConfiguration initialCredits(ByteCapacity initialCredits);
+
     /**
      * Flow strategy to use
      *
@@ -283,9 +300,12 @@ interface FlowConfiguration {
      * @see ConsumerFlowStrategy
      * @see ConsumerFlowStrategy#creditOnChunkArrival()
      * @see ConsumerFlowStrategy#creditOnChunkArrival(int)
+     * @see ConsumerFlowStrategy#creditOnChunkArrival(ByteCapacity)
      * @see ConsumerFlowStrategy#creditWhenHalfMessagesProcessed()
      * @see ConsumerFlowStrategy#creditWhenHalfMessagesProcessed(int)
+     * @see ConsumerFlowStrategy#creditWhenHalfMessagesProcessed(ByteCapacity)
      * @see ConsumerFlowStrategy#creditOnProcessedMessageCount(int, double)
+     * @see ConsumerFlowStrategy#creditOnProcessedMessageCount(ByteCapacity, double)
      */
     FlowConfiguration strategy(ConsumerFlowStrategy strategy);
 

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2023-2025 Broadcom. All Rights Reserved.
+// Copyright (c) 2023-2026 Broadcom. All Rights Reserved.
 // The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
@@ -31,6 +31,17 @@
  * ideal solution, it depends on the use cases and several parameters (processing time, network,
  * etc).
  *
+ * <p>Custom implementations must never make a credit release depend on the arrival of a
+ * <em>later</em> chunk. Releasing when a chunk arrives, or when its own messages are processed, is
+ * safe: the trigger has already happened by the time the credit is due. Waiting for a chunk that
+ * has not arrived yet is not safe, because the broker may be blocked precisely because that chunk
+ * was never sent.
+ *
+ * <p>Credit is usually expressed in chunks, but a strategy can express it in bytes instead, see
+ * {@link CreditUnit}. Byte-based credit still follows a chunk granularity: a chunk is delivered in
+ * full even if it exceeds the outstanding byte credit, so a consumer never stalls just because its
+ * window is smaller than the next chunk.
+ *
  * <p>This is an experimental API, subject to change.
  *
  * @since 0.12.0
@@ -61,6 +72,27 @@ public interface ConsumerFlowStrategy {
    */
   MessageProcessedCallback start(Context context);
 
+  /** The unit a subscription's credit is expressed in. */
+  enum CreditUnit {
+    /** 1 credit lets the broker send 1 more chunk, whatever its size. */
+    CHUNK,
+    /**
+     * Credit is a number of bytes; the client must eventually grant back every byte the broker
+     * charged for a chunk. Requires broker support, see {@link
+     * com.rabbitmq.stream.ConsumerBuilder.FlowConfiguration#initialCredits(ByteCapacity)}.
+     */
+    BYTE
+  }
+
+  /**
+   * The unit this strategy's credit is expressed in.
+   *
+   * <p>Defaults to {@link CreditUnit#CHUNK}.
+   *
+   * @return the credit unit
+   */
+  CreditUnit unit();
+
   /** Chunk context. */
   interface Context {
 
@@ -70,6 +102,14 @@ interface Context {
      * <p>{@link ConsumerFlowStrategy} implementation should always provide 1 credit for a given
      * chunk.
      *
+     * <p><code>credits</code> counts chunks in both units. For a byte-based subscription, the
+     * client grants the bytes of the corresponding chunks as credit, not the raw <code>credits
+     * </code> value.
+     *
+     * <p>Implementations must never call this method for a chunk based on the arrival of a
+     * <em>later</em> chunk, only on the arrival of the chunk itself or the processing of its own
+     * messages, see {@link ConsumerFlowStrategy}.
+     *
      * @param credits the number of credits provided, usually 1
      */
     void credits(int credits);
@@ -87,6 +127,15 @@ interface Context {
      * @return offset of the first message in the chunk (chunk ID)
      */
     long chunkId();
+
+    /**
+     * The cost the broker charged for the chunk, in bytes.
+     *
+     * <p>This is what a byte-based subscription must eventually grant back as credit.
+     *
+     * @return the chunk cost, in bytes
+     */
+    long chunkByteCount();
   }
 
   /** Behavior for {@link MessageHandler.Context#processed()} calls. */
@@ -134,6 +183,22 @@ static ConsumerFlowStrategy creditOnChunkArrival(int initialCredits) {
     return new CreditOnChunkArrivalConsumerFlowStrategy(initialCredits);
   }
 
+  /**
+   * Strategy that provides a byte window as initial credits and a credit on each new chunk.
+   *
+   * <p>Calls to {@link MessageHandler.Context#processed()} are ignored.
+   *
+   * <p>Requires a broker supporting {@code Subscribe} version 2.
+   *
+   * @param window initial credit window, in bytes
+   * @return flow strategy
+   * @see com.rabbitmq.stream.ConsumerBuilder.FlowConfiguration#initialCredits(ByteCapacity)
+   */
+  static ConsumerFlowStrategy creditOnChunkArrival(ByteCapacity window) {
+    return new CreditOnChunkArrivalConsumerFlowStrategy(
+        windowToInitialCredits(window), CreditUnit.BYTE);
+  }
+
   /**
    * Strategy that provides 10 initial credits and a credit when half of the chunk messages are
    * processed.
@@ -162,6 +227,23 @@ static ConsumerFlowStrategy creditWhenHalfMessagesProcessed(int initialCredits)
     return creditOnProcessedMessageCount(initialCredits, 0.5);
   }
 
+  /**
+   * Strategy that provides a byte window as initial credits and a credit when half of the chunk
+   * messages are processed.
+   *
+   * <p>Make sure to call {@link MessageHandler.Context#processed()} on every message when using
+   * this strategy, otherwise the broker may stop sending messages to the consumer.
+   *
+   * <p>Requires a broker supporting {@code Subscribe} version 2.
+   *
+   * @param window initial credit window, in bytes
+   * @return flow strategy
+   * @see com.rabbitmq.stream.ConsumerBuilder.FlowConfiguration#initialCredits(ByteCapacity)
+   */
+  static ConsumerFlowStrategy creditWhenHalfMessagesProcessed(ByteCapacity window) {
+    return creditOnProcessedMessageCount(window, 0.5);
+  }
+
   /**
    * Strategy that provides the specified number of initial credits and a credit when the specified
    * ratio of the chunk messages are processed.
@@ -176,6 +258,36 @@ static ConsumerFlowStrategy creditOnProcessedMessageCount(int initialCredits, do
     return new MessageCountConsumerFlowStrategy(initialCredits, ratio);
   }
 
+  /**
+   * Strategy that provides a byte window as initial credits and a credit when the specified ratio
+   * of the chunk messages are processed.
+   *
+   * <p>Make sure to call {@link MessageHandler.Context#processed()} on every message when using
+   * this strategy, otherwise the broker may stop sending messages to the consumer.
+   *
+   * <p>Requires a broker supporting {@code Subscribe} version 2.
+   *
+   * @param window initial credit window, in bytes
+   * @param ratio ratio of messages to process before providing credits
+   * @return flow strategy
+   * @see com.rabbitmq.stream.ConsumerBuilder.FlowConfiguration#initialCredits(ByteCapacity)
+   */
+  static ConsumerFlowStrategy creditOnProcessedMessageCount(ByteCapacity window, double ratio) {
+    return new MessageCountConsumerFlowStrategy(
+        windowToInitialCredits(window), ratio, CreditUnit.BYTE);
+  }
+
+  private static int windowToInitialCredits(ByteCapacity window) {
+    if (window == null || window.compareTo(ByteCapacity.B(0)) <= 0) {
+      throw new IllegalArgumentException("The window must be positive");
+    }
+    if (window.compareTo(ByteCapacity.B(Integer.MAX_VALUE)) > 0) {
+      throw new IllegalArgumentException(
+          "The window must be at most " + Integer.MAX_VALUE + " bytes");
+    }
+    return (int) window.toBytes();
+  }
+
   /**
    * Strategy that provides the specified number of initial credits and <code>n</code> credits every
    * <code>n</code> chunks.
@@ -189,6 +301,14 @@ static ConsumerFlowStrategy creditOnProcessedMessageCount(int initialCredits, do
    *
    * <p>Calls to {@link MessageHandler.Context#processed()} are ignored.
    *
+   * <p>This strategy has no byte-based variant and always uses {@link CreditUnit#CHUNK}: it
+   * releases credit once <code>n</code> chunks have arrived, which makes the release of the last
+   * <code>n - 1</code> chunks depend on the arrival of a chunk that has not happened yet, breaking
+   * the contract described in {@link ConsumerFlowStrategy}. In chunk mode the residual is bounded
+   * by <code>n</code> chunks, which the constructor check below keeps under control; in byte mode
+   * the residual would be an unbounded number of bytes, since chunk sizes are not known when the
+   * consumer is built.
+   *
    * @param initialCredits number of initial credits
    * @param n number of chunks and number of credits
    * @return flow strategy
@@ -234,6 +354,11 @@ public int initialCredits() {
       return this.initialCredits;
     }
 
+    @Override
+    public CreditUnit unit() {
+      return CreditUnit.CHUNK;
+    }
+
     @Override
     public MessageProcessedCallback start(Context context) {
       if (chunkCount.incrementAndGet() % n == 0) {
@@ -253,16 +378,27 @@ final class CreditOnChunkArrivalConsumerFlowStrategy implements ConsumerFlowStra
     private static final MessageProcessedCallback CALLBACK = v -> {};
 
     private final int initialCredits;
+    private final CreditUnit unit;
 
     private CreditOnChunkArrivalConsumerFlowStrategy(int initialCredits) {
+      this(initialCredits, CreditUnit.CHUNK);
+    }
+
+    CreditOnChunkArrivalConsumerFlowStrategy(int initialCredits, CreditUnit unit) {
       this.initialCredits = initialCredits;
+      this.unit = unit;
     }
 
     @Override
     public int initialCredits() {
       return this.initialCredits;
     }
 
+    @Override
+    public CreditUnit unit() {
+      return this.unit;
+    }
+
     @Override
     public MessageProcessedCallback start(Context context) {
       context.credits(1);
@@ -281,17 +417,28 @@ final class MessageCountConsumerFlowStrategy implements ConsumerFlowStrategy {
 
     private final int initialCredits;
     private final double ratio;
+    private final CreditUnit unit;
 
     private MessageCountConsumerFlowStrategy(int initialCredits, double ratio) {
+      this(initialCredits, ratio, CreditUnit.CHUNK);
+    }
+
+    MessageCountConsumerFlowStrategy(int initialCredits, double ratio, CreditUnit unit) {
       this.initialCredits = initialCredits;
       this.ratio = ratio;
+      this.unit = unit;
     }
 
     @Override
     public int initialCredits() {
       return this.initialCredits;
     }
 
+    @Override
+    public CreditUnit unit() {
+      return this.unit;
+    }
+
     @Override
     public MessageProcessedCallback start(Context context) {
       long l = (long) (context.messageCount() * ratio);