Reactive Messaging (RabbitMQ) - Consumers execution is sequential

[samodadela]

Describe the bug

SRRM with RabbitMQ connector:

• a topic exchange bound to 3 queues
• all queues use the same routing key ("all")
• 3 consumer methods annotated with Incoming and Blocking

When sending to the exchange using the "all" routing key, I'd expect the consumers to be run concurrently. What happens is that they get called sequentially.

Reproducer attached.

conc.zip

Expected behavior

Expecting Incoming methods to be called concurrently.

Actual behavior

Methods are called sequentially even though they listen on separate channels.
Even tried to use separate worker pool with max-concurrency, does not help.

How to Reproduce?

Run reproducer.

GET http://localhost:8080/x/generate

Check output.

Output of uname -a or ver

Linux a 5.15.65-1-MANJARO #1 SMP PREEMPT Mon Sep 5 10:15:47 UTC 2022 x86_64 GNU/Linux

Output of java -version

/opt/jdk-11.0.14.1+1/bin/java -version
openjdk version "11.0.14.1" 2022-02-08
OpenJDK Runtime Environment Temurin-11.0.14.1+1 (build 11.0.14.1+1)
OpenJDK 64-Bit Server VM Temurin-11.0.14.1+1 (build 11.0.14.1+1, mixed mode)

GraalVM version (if different from Java)

No response

Quarkus version or git rev

2.13.0.Final

Build tool (ie. output of mvnw --version or gradlew --version)

Apache Maven 3.8.6 (84538c9988a25aec085021c365c560670ad80f63)
Maven home: /opt/apache-maven-3.8.6
Java version: 11.0.14.1, vendor: Eclipse Adoptium, runtime: /opt/jdk-11.0.14.1+1
Default locale: en_US, platform encoding: UTF-8
OS name: "linux", version: "5.15.65-1-manjaro", arch: "amd64", family: "unix"

Additional information

Can't figure out how to invoke methods concurrently.

[quarkus-bot]

/cc @cescoffier, @ozangunalp

[ozangunalp]

@samodadela Thanks for the reproducer.

By default, messages in reactive messaging are invoked sequentially, inside the Vert.x event loop.
For the RabbitMQ connector, messages from different channels are dispatched on the same Vert.x event loop.
Even using @Blocking, which offloads the execution to the a worker pool, sequential execution is the default.
You'd need to pass ordered = false to the @Blocking annotation for concurrently running message invocations.

[samodadela]

@ozangunalp

Will ordered = false change also the order within the same channel? I need sequential ordering within a channel, but I also want to achieve parallelism between channels - since they are independent.

Imagine that in one channel messages are processed slowly, but in other channels processing can be fast. As it is, the 'slowest' channel will dictate the pace of the whole system.

[samodadela]

My use case is an event sourced system with multiple projections. Each event is fed to multiple projections. Projections are independent, the only requirement is that within a projections events are processed in order.

I choose RabbitMQ because it has the single-active-cluster feature - that is RabbitMQ will ensure that there's only one node in the whole cluster that will process the events for one projection. So if I had multiple nodes, projections would run in parallel, right? I'd like to achieve some level of parallelism on a single node, too.

[samodadela]

An interesting consequence of the messages being processed sequentially is that if worker is running for more than 30 minutes and someone sends a message (on any queue), rabbitmq will hit the consumer_timeout and restart.

The error is:

<0.914.0> operation none caused a channel exception precondition_failed: delivery acknowledgement on channel 1 timed out. Timeout value used: 1800000 ms. This timeout value can be configured, see consumers doc guide to learn more�[0m

The workaround is to set consumer_timeout to a greater value or undefined.

[cescoffier]

@ozangunalp didn't you add the concurrency support to the rabbitmq connector?

[ozangunalp]

Indeed, concurrency config may be an option. But it dispatches messages on (non-blocking) event loop threads.
For @Blocking execution I think you'd need different worker pools per channel.

[ozangunalp]

You can use concurrency config along with @Blocking and it'll work as you expect but it is a single consuming method.

That being said both rabbitmq and amqp connectors use the default Vert.x context for dispatching messages, and that's the reason for your issue.
I'll keep this issue open and investigate changing this for both these connectors.

[ozangunalp]

Resolved in Reactive Messaging 4.22.0 integrated with #41851