Repository: smallrye/smallrye-reactive-messaging
Issue: Unordered blocking consumer requests too many messages

# Describe the bug
I made a [discussion post](https://github.com/smallrye/smallrye-reactive-messaging/discussions/2196) describing the problem:

<div type='discussions-op-text'>

<sup>Originally posted by **AndreasPetersen** June 14, 2023</sup>
I'm implementing a connector to Azure Service Bus using SmallRye Reactive Messaging 3.22.1. I want to be able to concurrently process a number of messages. I can do this by:

```
@Incoming("in-history")
@Blocking(ordered = false, value = "in-history")
public CompletionStage<Void> handleMessageFromAzure(Message<History> message) {
    try {
        logger.info("Processing message: " + message.getMetadata(MessageId.class).get().id()); // My connector adds a Azure Service Bus message id as metadata
        Thread.sleep(1000); // Do some blocking work
        return message.ack();
    } catch (InterruptedException e) {
        return message.nack(e);
    }
}
```

with the config:

```
smallrye.messaging.worker.in-history.max-concurrency=2
```

This works almost as expected. Two messages are indeed processed concurrently, but requests are constantly being made to the publisher. My connector looks like this:

```
    @Override
    public PublisherBuilder<? extends Message<?>> getPublisherBuilder(Config config) {
        ServiceBusReceiverAsyncClient serviceBusReceiverAsyncClient = getServiceBusReceiverAsyncClient(config); // This gives us an Azure Service Bus client that we can use to received messages from Azure
        Flux<ServiceBusReceivedMessage> flux = serviceBusReceiverAsyncClient.receiveMessages();

        var multi = Multi.createFrom().publisher(flux)
                .log() // I've added a log to see what's going on
                .onItem().invoke(serviceBusReceivedMessage -> logger.info("Received {}", serviceBusReceivedMessage.getMessageId()))                     
                .onItem().transform(azureMessage -> mapToReactiveMessage(azureMessage))

        return ReactiveStreams.fromPublisher(multi);
    }
```

When starting my Quarkus service I can see in the logs:

```
2023-06-14 17:27:50,196 INFO  [dk.ban.qua.rea.mes.asb.AzureServiceBusConnector] (boundedElastic-1) {} Received 9500cb1f42ed4b9a860ec35c4c89b172
2023-06-14 17:27:50,216 INFO  [io.qua.mut.run.MutinyInfrastructure] (vert.x-eventloop-thread-3) {} Multi..0 | request(1)
2023-06-14 17:27:50,247 INFO  [io.qua.mut.run.MutinyInfrastructure] (boundedElastic-1) {} Multi..0 | onItem(com.azure.messaging.servicebus.ServiceBusReceivedMessage@79ed499b)
2023-06-14 17:27:50,248 INFO  [dk.ban.qua.rea.mes.asb.AzureServiceBusConnector] (boundedElastic-1) {} Received b4f16eb0798c459aaef6bb7f1b6bcaed
2023-06-14 17:27:51,574 INFO  [io.qua.mut.run.MutinyInfrastructure] (pool-1-thread-1) {} Multi..0 | request(1)
2023-06-14 17:27:51,575 INFO  [io.qua.mut.run.MutinyInfrastructure] (pool-1-thread-1) {} Multi..0 | onItem(com.azure.messaging.servicebus.ServiceBusReceivedMessage@337cb7d4)
2023-06-14 17:27:51,575 INFO  [dk.ban.qua.rea.mes.asb.AzureServiceBusConnector] (pool-1-thread-1) {} Received f4149959456041eaa386605c11a5c536
2023-06-14 17:27:51,576 INFO  [io.qua.mut.run.MutinyInfrastructure] (vert.x-eventloop-thread-4) {} Multi..0 | request(1)
2023-06-14 17:27:51,582 INFO  [io.sma.rea.mes.pro.con.WorkerPoolRegistry] (vert.x-eventloop-thread-3) {sampled=true, spanId=edfc2b30352bb593, traceId=8bd823a32007c950151012ec38489f85} Created worker pool named in-history with concurrency of 2
2023-06-14 17:27:51,591 INFO  [dk.ban.drb.ser.GenerateHistoryServiceBusConsumer] (in-history-0) {sampled=true, spanId=9213606bb5f98990, traceId=b2a7e115f36f8208c4a92c82db28fa4a} Processing message: f4149959456041eaa386605c11a5c536
2023-06-14 17:27:51,591 INFO  [dk.ban.drb.ser.GenerateHistoryServiceBusConsumer] (in-history-1) {sampled=true, spanId=edfc2b30352bb593, traceId=8bd823a32007c950151012ec38489f85} Processing message: 9500cb1f42ed4b9a860ec35c4c89b172
2023-06-14 17:27:51,615 INFO  [io.qua.mut.run.MutinyInfrastructure] (boundedElastic-1) {} Multi..0 | onItem(com.azure.messaging.servicebus.ServiceBusReceivedMessage@6b8ff80c)
2023-06-14 17:27:51,616 INFO  [dk.ban.qua.rea.mes.asb.AzureServiceBusConnector] (boundedElastic-1) {} Received bc9e6cbe4cb040d7a3e1a67c3b5faeaf
2023-06-14 17:27:51,617 INFO  [io.qua.mut.run.MutinyInfrastructure] (vert.x-eventloop-thread-3) {} Multi..0 | request(1)
2023-06-14 17:27:51,661 INFO  [io.qua.mut.run.MutinyInfrastructure] (boundedElastic-1) {} Multi..0 | onItem(com.azure.messaging.servicebus.ServiceBusReceivedMessage@48e9a550)
2023-06-14 17:27:51,662 INFO  [dk.ban.qua.rea.mes.asb.AzureServiceBusConnector] (boundedElastic-1) {} Received 43a5c809241b4222bb985d8a0d6baa9c
2023-06-14 17:27:51,663 INFO  [io.qua.mut.run.MutinyInfrastructure] (vert.x-eventloop-thread-3) {} Multi..0 | request(1)
2023-06-14 17:27:51,702 INFO  [io.qua.mut.run.MutinyInfrastructure] (boundedElastic-1) {} Multi..0 | onItem(com.azure.messaging.servicebus.ServiceBusReceivedMessage@336e7e88)
2023-06-14 17:27:51,702 INFO  [dk.ban.qua.rea.mes.asb.AzureServiceBusConnector] (boundedElastic-1) {} Received ecbc17cb38dd44a08b2b3421b7c9eb72
2023-06-14 17:27:51,703 INFO  [io.qua.mut.run.MutinyInfrastructure] (vert.x-eventloop-thread-3) {} Multi..0 | request(1)
2023-06-14 17:27:51,741 INFO  [io.qua.mut.run.MutinyInfrastructure] (boundedElastic-1) {} Multi..0 | onItem(com.azure.messaging.servicebus.ServiceBusReceivedMessage@20dc58ef)
2023-06-14 17:27:51,742 INFO  [dk.ban.qua.rea.mes.asb.AzureServiceBusConnector] (boundedElastic-1) {} Received 03002654344c44638eb78da8c10649d8
2023-06-14 17:27:51,743 INFO  [io.qua.mut.run.MutinyInfrastructure] (vert.x-eventloop-thread-3) {} Multi..0 | request(1)
2023-06-14 17:27:51,784 INFO  [io.qua.mut.run.MutinyInfrastructure] (boundedElastic-1) {} Multi..0 | onItem(com.azure.messaging.servicebus.ServiceBusReceivedMessage@34079b60)
2023-06-14 17:27:51,785 INFO  [dk.ban.qua.rea.mes.asb.AzureServiceBusConnector] (boundedElastic-1) {} Received 24ef34ed572a412fb4f23083bb7df6d7
2023-06-14 17:27:51,788 INFO  [io.qua.mut.run.MutinyInfrastructure] (vert.x-eventloop-thread-3) {} Multi..0 | request(1)
2023-06-14 17:27:51,830 INFO  [io.qua.mut.run.MutinyInfrastructure] (boundedElastic-1) {} Multi..0 | onItem(com.azure.messaging.servicebus.ServiceBusReceivedMessage@31417b95)
2023-06-14 17:27:51,830 INFO  [dk.ban.qua.rea.mes.asb.AzureServiceBusConnector] (boundedElastic-1) {} Received 1787ba34ea924740b31db1fd2a7a8b28
2023-06-14 17:27:51,833 INFO  [io.qua.mut.run.MutinyInfrastructure] (vert.x-eventloop-thread-3) {} Multi..0 | request(1)
2023-06-14 17:27:51,871 INFO  [io.qua.mut.run.MutinyInfrastructure] (boundedElastic-1) {} Multi..0 | onItem(com.azure.messaging.servicebus.ServiceBusReceivedMessage@711bc90b)
2023-06-14 17:27:51,872 INFO  [dk.ban.qua.rea.mes.asb.AzureServiceBusConnector] (boundedElastic-1) {} Received b31a40ac77744e0fb8165c83934b6cf0
2023-06-14 17:27:51,874 INFO  [io.qua.mut.run.MutinyInfrastructure] (vert.x-eventloop-thread-3) {} Multi..0 | request(1)
2023-06-14 17:27:51,912 INFO  [io.qua.mut.run.MutinyInfrastructure] (boundedElastic-1) {} Multi..0 | onItem(com.azure.messaging.servicebus.ServiceBusReceivedMessage@3342fce7)
2023-06-14 17:27:51,913 INFO  [dk.ban.qua.rea.mes.asb.AzureServiceBusConnector] (boundedElastic-1) {} Received 40eb624688c54727a026b74a45c0fe49
2023-06-14 17:27:51,914 INFO  [io.qua.mut.run.MutinyInfrastructure] (vert.x-eventloop-thread-3) {} Multi..0 | request(1)
2023-06-14 17:27:51,953 INFO  [io.qua.mut.run.MutinyInfrastructure] (boundedElastic-1) {} Multi..0 | onItem(com.azure.messaging.servicebus.ServiceBusReceivedMessage@3b96bd74)
2023-06-14 17:27:51,953 INFO  [dk.ban.qua.rea.mes.asb.AzureServiceBusConnector] (boundedElastic-1) {} Received cfe3c9d1739d4342a1a1e379deb4ec0c
2023-06-14 17:27:51,954 INFO  [io.qua.mut.run.MutinyInfrastructure] (vert.x-eventloop-thread-3) {} Multi..0 | request(1)
2023-06-14 17:27:52,000 INFO  [io.qua.mut.run.MutinyInfrastructure] (boundedElastic-1) {} Multi..0 | onItem(com.azure.messaging.servicebus.ServiceBusReceivedMessage@1b1f440)
2023-06-14 17:27:52,001 INFO  [dk.ban.qua.rea.mes.asb.AzureServiceBusConnector] (boundedElastic-1) {} Received 3649dce24a3e45519f9f14730f909ace
2023-06-14 17:27:52,002 INFO  [io.qua.mut.run.MutinyInfrastructure] (vert.x-eventloop-thread-3) {} Multi..0 | request(1)
2023-06-14 17:27:52,143 INFO  [io.qua.mut.run.MutinyInfrastructure] (boundedElastic-1) {} Multi..0 | onItem(com.azure.messaging.servicebus.ServiceBusReceivedMessage@2e4d26ec)
2023-06-14 17:27:52,144 INFO  [dk.ban.qua.rea.mes.asb.AzureServiceBusConnector] (boundedElastic-1) {} Received 0be14e338d30481881c5b58cf02a99e3
2023-06-14 17:27:52,145 INFO  [io.qua.mut.run.MutinyInfrastructure] (vert.x-eventloop-thread-3) {} Multi..0 | request(1)
2023-06-14 17:27:52,189 INFO  [io.qua.mut.run.MutinyInfrastructure] (boundedElastic-1) {} Multi..0 | onItem(com.azure.messaging.servicebus.ServiceBusReceivedMessage@acc4206)
2023-06-14 17:27:52,189 INFO  [dk.ban.qua.rea.mes.asb.AzureServiceBusConnector] (boundedElastic-1) {} Received c0a02c97fdd34075ad7ceb830cb28e52
2023-06-14 17:27:52,191 INFO  [io.qua.mut.run.MutinyInfrastructure] (vert.x-eventloop-thread-3) {} Multi..0 | request(1)
2023-06-14 17:27:52,232 INFO  [io.qua.mut.run.MutinyInfrastructure] (boundedElastic-1) {} Multi..0 | onItem(com.azure.messaging.servicebus.ServiceBusReceivedMessage@43f2336)
2023-06-14 17:27:52,232 INFO  [dk.ban.qua.rea.mes.asb.AzureServiceBusConnector] (boundedElastic-1) {} Received f862c0ffcf284bf2b8357368ba755ba7
2023-06-14 17:27:52,234 INFO  [io.qua.mut.run.MutinyInfrastructure] (vert.x-eventloop-thread-3) {} Multi..0 | request(1)
2023-06-14 17:27:52,280 INFO  [io.qua.mut.run.MutinyInfrastructure] (boundedElastic-1) {} Multi..0 | onItem(com.azure.messaging.servicebus.ServiceBusReceivedMessage@a22d405)
2023-06-14 17:27:52,280 INFO  [dk.ban.qua.rea.mes.asb.AzureServiceBusConnector] (boundedElastic-1) {} Received 1344b27a8dac4ca39420831de2f00d47
2023-06-14 17:27:52,626 INFO  [dk.ban.drb.ser.GenerateHistoryServiceBusConsumer] (in-history-1) {sampled=true, spanId=ebeaacbb82822529, traceId=e7549b9f53135ff66f7621b37c1ab969} Processing message: b4f16eb0798c459aaef6bb7f1b6bcaed
2023-06-14 17:27:52,626 INFO  [dk.ban.drb.ser.GenerateHistoryServiceBusConsumer] (in-history-0) {sampled=true, spanId=490007faaf99b2bd, traceId=8d20cb5d3553d16afc17316635472b88} Processing message: bc9e6cbe4cb040d7a3e1a67c3b5faeaf
2023-06-14 17:27:52,678 INFO  [dk.ban.qua.rea.mes.asb.AzureServiceBusIncomingMessage] (reactor-executor-1) {} Acked 9500cb1f42ed4b9a860ec35c4c89b172
2023-06-14 17:27:52,680 INFO  [dk.ban.qua.rea.mes.asb.AzureServiceBusIncomingMessage] (reactor-executor-1) {} Acked f4149959456041eaa386605c11a5c536
```

How can I limit the subscriber to only request new items, when existing ones have been acked? I would expect the subscriber to request 2 items, process those two items, and then request another 2. Instead, the subscriber just constantly requests 1 item, resulting in my service not being able to keep up and eventually getting a variety of errors.

If the blocking annotation is ordered, then a new item is only requested once the current item has been acked.</div>

# Expected behavior
Messages requests should match configured worker threads concurrency.

# Actual behavior
From my understanding, the problem occurs at [this line](https://github.com/smallrye/smallrye-reactive-messaging/blob/1d2db081d7acf3f1307f445071702e058d6d295e/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/PublisherMediator.java#LL146C28-L146C28).

`transformToUniAndMerge` will merge with the default concurrency of `Queues.BUFFER_S` which is 256. The merge will therefore fill up the buffer with 256 messages, despite the consumer worker thread having a lower concurrency. This becomes a problem with a slow consumer with messages being received in peek-lock mode. The consumer cannot keep up with the requests.

If I set the system property `mutiny.buffer-size.s` to match the concurrency of the worker threads, then messages are only being received as fast as they can be processed. Of course setting this will globally change the buffer size for all merges, which is not exactly great.

# Implementation ideas
Should the merge not match the concurrency of the configured worker threads? Or at the very least be configurable to do so? I'm thinking it should be something along the lines of:

```
this.publisher = decorate(MultiUtils.createFromGenerator(this::invokeBlocking)
                        .onItem().transformToUni(u -> u)
                        .withRequests(configuredRequestSize) // I'm not sure if this should be changed from the default 1. Maybe it could make sense to match the configured concurrency here as well, or allow it to be configured separately? 
                        .merge(configuredConcurrency) // This will ensure the consumer can keep up
                        .onItem().transform(Message::of));
```

at [this ](https://github.com/smallrye/smallrye-reactive-messaging/blob/1d2db081d7acf3f1307f445071702e058d6d295e/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/PublisherMediator.java#LL146C28-L146C28) and [this ](https://github.com/smallrye/smallrye-reactive-messaging/blob/1d2db081d7acf3f1307f445071702e058d6d295e/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/PublisherMediator.java#LL166C1-L167C1) location.

# Version
I've tested this with Quarkus 2.16.7 with JDK17 and SmallRye Reactive Messaging 3.22.1. I haven't tested the latest version, but the code in question seems to be the same as in 3.22.1.

Changed files:
smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/ProcessorMediator.java
smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/extension/MediatorManager.java
smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/connectors/WorkerPoolRegistry.java
smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/AbstractMediator.java
smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/SubscriberMediator.java
smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/PublisherMediator.java

diff --git a/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/AbstractMediator.java b/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/AbstractMediator.java
index 8bb6a5a57..0f252e74f 100644
--- a/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/AbstractMediator.java
+++ b/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/AbstractMediator.java
@@ -44,6 +44,7 @@ public abstract class AbstractMediator {
     protected HealthCenter health;
     private Instance<MessageConverter> converters;
     private Instance<KeyValueExtractor> extractors;
+    private int maxConcurrency;
 
     public AbstractMediator(MediatorConfiguration configuration) {
         this.configuration = configuration;
@@ -99,6 +100,10 @@ public abstract class AbstractMediator {
         this.workerPoolRegistry = workerPoolRegistry;
     }
 
+    public void setMaxConcurrency(int maxConcurrency) {
+        this.maxConcurrency = maxConcurrency;
+    }
+
     public void run() {
         // Do nothing by default.
     }
@@ -259,8 +264,11 @@ public abstract class AbstractMediator {
         return extractors;
     }
 
+    public int maxConcurrency() {
+        return maxConcurrency;
+    }
+
     public void terminate() {
         // Do nothing by default.
     }
-
 }
diff --git a/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/ProcessorMediator.java b/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/ProcessorMediator.java
index 822d76ccd..cde6e8afe 100644
--- a/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/ProcessorMediator.java
+++ b/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/ProcessorMediator.java
@@ -323,10 +323,11 @@ public class ProcessorMediator extends AbstractMediator {
                 this.mapper = upstream -> {
                     Multi<? extends Message<?>> multi = MultiUtils.handlePreProcessingAcknowledgement(upstream, configuration);
                     return multi
-                            .onItem().transformToMultiAndMerge(message -> invokeBlocking(message, getArguments(message))
+                            .onItem().transformToMulti(message -> invokeBlocking(message, getArguments(message))
                                     .onItemOrFailure()
                                     .transformToUni((o, t) -> this.handlePostInvocationWithMessage((Message<?>) o, t))
-                                    .onItem().transformToMulti(this::handleSkip));
+                                    .onItem().transformToMulti(this::handleSkip))
+                            .merge(maxConcurrency());
                 };
             }
 
@@ -358,9 +359,10 @@ public class ProcessorMediator extends AbstractMediator {
                                 .onItem().transformToMulti(this::handleSkip));
             } else {
                 this.mapper = upstream -> MultiUtils.handlePreProcessingAcknowledgement(upstream, configuration)
-                        .onItem().transformToMultiAndMerge(message -> invokeBlocking(message, getArguments(message))
+                        .onItem().transformToMulti(message -> invokeBlocking(message, getArguments(message))
                                 .onItemOrFailure().transformToUni((r, f) -> handlePostInvocation(message, r, f))
-                                .onItem().transformToMulti(this::handleSkip));
+                                .onItem().transformToMulti(this::handleSkip))
+                        .merge(maxConcurrency());
             }
 
         } else {
diff --git a/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/PublisherMediator.java b/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/PublisherMediator.java
index c64807b4a..7b4d69beb 100644
--- a/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/PublisherMediator.java
+++ b/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/PublisherMediator.java
@@ -143,7 +143,7 @@ public class PublisherMediator extends AbstractMediator {
                         .onItem().transform(o -> (Message<?>) o));
             } else {
                 this.publisher = decorate(MultiUtils.createFromGenerator(this::invokeBlocking)
-                        .onItem().transformToUniAndMerge(u -> u)
+                        .onItem().transformToUni(u -> u).merge(maxConcurrency())
                         .onItem().transform(o -> (Message<?>) o));
             }
         } else {
@@ -163,7 +163,7 @@ public class PublisherMediator extends AbstractMediator {
                         .onItem().transform(Message::of));
             } else {
                 this.publisher = decorate(MultiUtils.createFromGenerator(this::invokeBlocking)
-                        .onItem().transformToUniAndMerge(u -> u)
+                        .onItem().transformToUni(u -> u).merge(maxConcurrency())
                         .onItem().transform(Message::of));
             }
         } else {
diff --git a/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/SubscriberMediator.java b/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/SubscriberMediator.java
index 12f4fd6e0..090d24fad 100644
--- a/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/SubscriberMediator.java
+++ b/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/SubscriberMediator.java
@@ -165,8 +165,9 @@ public class SubscriberMediator extends AbstractMediator {
                         .invoke(failure -> health.reportApplicationFailure(configuration.methodAsString(), failure));
             } else {
                 this.function = upstream -> MultiUtils.handlePreProcessingAcknowledgement(upstream, configuration)
-                        .onItem().transformToUniAndMerge(msg -> invokeBlocking(msg, getArguments(msg))
+                        .onItem().transformToUni(msg -> invokeBlocking(msg, getArguments(msg))
                                 .onItemOrFailure().transformToUni(handleInvocationResult(msg)))
+                        .merge(maxConcurrency())
                         .onFailure()
                         .invoke(failure -> health.reportApplicationFailure(configuration.methodAsString(), failure));
             }
@@ -210,7 +211,7 @@ public class SubscriberMediator extends AbstractMediator {
                         .onFailure().invoke(this::reportFailure);
             } else {
                 this.function = upstream -> MultiUtils.handlePreProcessingAcknowledgement(upstream, configuration)
-                        .onItem().transformToUniAndMerge(this::invokeBlockingAndHandleOutcome)
+                        .onItem().transformToUni(this::invokeBlockingAndHandleOutcome).merge(maxConcurrency())
                         .onFailure().invoke(this::reportFailure);
             }
         } else {
@@ -243,7 +244,7 @@ public class SubscriberMediator extends AbstractMediator {
                         .onFailure().invoke(this::reportFailure);
             } else {
                 this.function = upstream -> MultiUtils.handlePreProcessingAcknowledgement(upstream, configuration)
-                        .onItem().transformToUniAndMerge(this::invokeBlockingAndHandleOutcome)
+                        .onItem().transformToUni(this::invokeBlockingAndHandleOutcome).merge(maxConcurrency())
                         .onFailure().invoke(this::reportFailure);
             }
         } else {
diff --git a/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/connectors/WorkerPoolRegistry.java b/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/connectors/WorkerPoolRegistry.java
index 067db3403..13bdf4567 100644
--- a/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/connectors/WorkerPoolRegistry.java
+++ b/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/connectors/WorkerPoolRegistry.java
@@ -35,8 +35,8 @@ import io.vertx.mutiny.core.WorkerExecutor;
 
 @ApplicationScoped
 public class WorkerPoolRegistry {
-    private static final String WORKER_CONFIG_PREFIX = "smallrye.messaging.worker";
-    private static final String WORKER_CONCURRENCY = "max-concurrency";
+    public static final String WORKER_CONFIG_PREFIX = "smallrye.messaging.worker";
+    public static final String WORKER_CONCURRENCY = "max-concurrency";
 
     @Inject
     Instance<ExecutionHolder> executionHolder;
diff --git a/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/extension/MediatorManager.java b/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/extension/MediatorManager.java
index 20f2042ca..1d502baef 100644
--- a/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/extension/MediatorManager.java
+++ b/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/extension/MediatorManager.java
@@ -1,5 +1,7 @@
 package io.smallrye.reactive.messaging.providers.extension;
 
+import static io.smallrye.reactive.messaging.providers.connectors.WorkerPoolRegistry.WORKER_CONCURRENCY;
+import static io.smallrye.reactive.messaging.providers.connectors.WorkerPoolRegistry.WORKER_CONFIG_PREFIX;
 import static io.smallrye.reactive.messaging.providers.i18n.ProviderLogging.log;
 
 import java.lang.reflect.Constructor;
@@ -11,10 +13,12 @@ import jakarta.enterprise.inject.Instance;
 import jakarta.enterprise.inject.spi.*;
 import jakarta.inject.Inject;
 
+import org.eclipse.microprofile.config.Config;
 import org.eclipse.microprofile.config.inject.ConfigProperty;
 import org.eclipse.microprofile.reactive.messaging.Incoming;
 import org.eclipse.microprofile.reactive.messaging.Outgoing;
 
+import io.smallrye.mutiny.helpers.queues.Queues;
 import io.smallrye.reactive.messaging.*;
 import io.smallrye.reactive.messaging.EmitterConfiguration;
 import io.smallrye.reactive.messaging.PublisherDecorator;
@@ -79,6 +83,9 @@ public class MediatorManager {
     @ConfigProperty(name = STRICT_MODE_PROPERTY, defaultValue = "false")
     boolean strictMode;
 
+    @Inject
+    Instance<Config> configInstance;
+
     public <T> void analyze(AnnotatedType<T> annotatedType, Bean<T> bean) {
 
         if (strictMode) {
@@ -97,6 +104,22 @@ public class MediatorManager {
                 });
     }
 
+    private int getWorkerMaxConcurrency(MediatorConfiguration configuration) {
+        // max concurrency is not relevant if not blocking
+        if (!configuration.isBlocking()) {
+            return -1;
+        }
+        String poolName = configuration.getWorkerPoolName();
+        // if the poll name is null we are on the default worker pool, set the default concurrent requests
+        if (poolName == null) {
+            return Queues.BUFFER_S;
+        }
+        String concurrencyConfigKey = WORKER_CONFIG_PREFIX + "." + poolName + "." + WORKER_CONCURRENCY;
+        Optional<Integer> concurrency = configInstance.get().getOptionalValue(concurrencyConfigKey, Integer.class);
+        // Fallback to the default concurrent requests if setting is not found
+        return concurrency.orElse(Queues.BUFFER_S);
+    }
+
     /**
      * This method is used in the Quarkus extension.
      *
@@ -177,6 +200,7 @@ public class MediatorManager {
         mediator.setExtractors(extractors);
         mediator.setHealth(health);
         mediator.setWorkerPoolRegistry(workerPoolRegistry);
+        mediator.setMaxConcurrency(getWorkerMaxConcurrency(configuration));
 
         try {
             Object beanInstance = beanManager.getReference(configuration.getBean(), Object.class,