Limiting requests to publisher on unordered blocking consumer

[AndreasPetersen]

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.