Refine some consumer flow control details

Author: acogoluegnes

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

@@ -30,6 +30,8 @@
  * ideal solution, it depends on the use cases and several parameters (processing time, network,
  * etc).
  *
+ * <p>This is an experimental API, subject to change.
+ *
  * @since 0.12.0
  * @see MessageHandler.Context#processed()
  * @see ConsumerBuilder#flow()

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

@@ -562,19 +562,9 @@ private ClientSubscriptionsManager(Broker node, Client.ClientParameters clientPa
                 subscriptionTrackers.get(subscriptionId & 0xFF);
             ConsumerFlowStrategy.MessageProcessedCallback processCallback;
             if (subscriptionTracker != null && subscriptionTracker.consumer.isOpen()) {
-              ConsumerFlowStrategy.Context chunkContext =
-                  new ConsumerFlowStrategy.Context() {
-                    @Override
-                    public void credits(int credits) {
-                      client.credit(subscriptionId, 1);
-                    }
-
-                    @Override
-                    public long messageCount() {
-                      return messageCount;
-                    }
-                  };
-              processCallback = subscriptionTracker.flowStrategy.start(chunkContext);
+              processCallback =
+                  subscriptionTracker.flowStrategy.start(
+                      new DefaultConsumerFlowStrategyContext(subscriptionId, client, messageCount));
             } else {
               LOGGER.debug(
                   "Could not find stream subscription {} or subscription closing, not providing credits",
@@ -1208,4 +1198,36 @@ public ClientClosedException() {
       super("Client already closed");
     }
   }
+
+  private static class DefaultConsumerFlowStrategyContext implements ConsumerFlowStrategy.Context {
+
+    private final byte subscriptionId;
+    private final Client client;
+    private final long messageCount;
+
+    private DefaultConsumerFlowStrategyContext(
+        byte subscriptionId, Client client, long messageCount) {
+      this.subscriptionId = subscriptionId;
+      this.client = client;
+      this.messageCount = messageCount;
+    }
+
+    @Override
+    public void credits(int credits) {
+      try {
+        client.credit(subscriptionId, credits);
+      } catch (Exception e) {
+        LOGGER.info(
+            "Error while providing {} credit(s) to subscription {}: {}",
+            credits,
+            subscriptionId,
+            e.getMessage());
+      }
+    }
+
+    @Override
+    public long messageCount() {
+      return messageCount;
+    }
+  }
 }

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

@@ -134,7 +134,7 @@ void nameShouldBeSetIfTrackingStrategyIsSet() {
   @BrokerVersionAtLeast(BrokerVersion.RABBITMQ_3_11)
   void committedOffsetShouldBeSet() throws Exception {
     int messageCount = 20_000;
-    TestUtils.publishAndWaitForConfirms(cf, messageCount, this.stream);
+    publishAndWaitForConfirms(cf, messageCount, this.stream);
 
     CountDownLatch consumeLatch = new CountDownLatch(messageCount);
     AtomicLong committedOffset = new AtomicLong();
@@ -196,7 +196,7 @@ void consume() throws Exception {
   @Test
   void consumeWithAsyncConsumerFlowControl() throws Exception {
     int messageCount = 100_000;
-    TestUtils.publishAndWaitForConfirms(cf, messageCount, stream);
+    publishAndWaitForConfirms(cf, messageCount, stream);
 
     ConsumerBuilder consumerBuilder =
         environment.consumerBuilder().stream(stream)
@@ -241,7 +241,7 @@ void consumeWithAsyncConsumerFlowControl() throws Exception {
   @Test
   void asynchronousProcessingWithFlowControl() {
     int messageCount = 100_000;
-    TestUtils.publishAndWaitForConfirms(cf, messageCount, stream);
+    publishAndWaitForConfirms(cf, messageCount, stream);
 
     ExecutorService executorService =
         Executors.newFixedThreadPool(getRuntime().availableProcessors());
@@ -253,13 +253,12 @@ void asynchronousProcessingWithFlowControl() {
           .strategy(creditWhenHalfMessagesProcessed())
           .builder()
           .messageHandler(
-              (ctx, message) -> {
-                executorService.submit(
-                    () -> {
-                      latch.countDown();
-                      ctx.processed();
-                    });
-              })
+              (ctx, message) ->
+                  executorService.submit(
+                      () -> {
+                        latch.countDown();
+                        ctx.processed();
+                      }))
           .build();
       assertThat(latch).is(completed());
     } finally {