Add test with asynchronous processing

To test flow control.

Author: acogoluegnes

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

@@ -15,6 +15,8 @@
 
 import static com.rabbitmq.stream.ConsumerFlowStrategy.creditWhenHalfMessagesProcessed;
 import static com.rabbitmq.stream.impl.TestUtils.*;
+import static com.rabbitmq.stream.impl.TestUtils.CountDownLatchConditions.completed;
+import static java.lang.Runtime.getRuntime;
 import static java.lang.String.format;
 import static java.util.Collections.synchronizedList;
 import static org.assertj.core.api.Assertions.*;
@@ -236,6 +238,35 @@ void consumeWithAsyncConsumerFlowControl() throws Exception {
     consumer.close();
   }
 
+  @Test
+  void asynchronousProcessingWithFlowControl() {
+    int messageCount = 100_000;
+    TestUtils.publishAndWaitForConfirms(cf, messageCount, stream);
+
+    ExecutorService executorService =
+        Executors.newFixedThreadPool(getRuntime().availableProcessors());
+    try {
+      CountDownLatch latch = new CountDownLatch(messageCount);
+      environment.consumerBuilder().stream(stream)
+          .offset(OffsetSpecification.first())
+          .flow()
+          .strategy(creditWhenHalfMessagesProcessed())
+          .builder()
+          .messageHandler(
+              (ctx, message) -> {
+                executorService.submit(
+                    () -> {
+                      latch.countDown();
+                      ctx.processed();
+                    });
+              })
+          .build();
+      assertThat(latch).is(completed());
+    } finally {
+      executorService.shutdownNow();
+    }
+  }
+
   @Test
   void closeOnCondition() throws Exception {
     int messageCount = 50_000;