Jobs plugin (rabbitmq): is it possible to have one and only one consumer per queue?

[shieldz80]

Hello,

In my projects at work, I have some scenarios where processing messages in order is critical. From my research RabbitMQ guarantees that messages are processed in order only if there is one and only one consumer per queue (https://www.rabbitmq.com/docs/semantics#ordering)

With release 2.7.0 and later it is still possible for individual consumers to observe messages out of order if the queue has multiple subscribers. This is due to the actions of other subscribers who may requeue messages. From the perspective of the queue the messages are always held in the publication order.

I would like to use RoadRunner and the jobs plugin to implement RabbitMQ workers, but I was wondering if RoadRunner supports restricting one consumer per queue. I was also wondering how it works in general in RoadRunner if I configure let's say 4 queues and a pool with 4 workers. Will that configuration guarantee that I will only have one consumer per queue? How about scenarios where for one queue I need to have only one consumer but for other queues I may need more than one because I don't care about the message processing order?

Also from my research I've read about RabbitMQ features like exclusive consumer (https://www.rabbitmq.com/docs/consumers#exclusivity) and single active consumer (https://www.rabbitmq.com/docs/consumers#single-active-consumer).

Are these supported in RoadRunner? As I understand at least with single active consumer, RabbitMQ will take care that there's only one consumer, but was wondering if there's something I need to be aware of on RoadRunner's side - traps and pitfalls type of thing.

Thank you in advance for any insight you may offer into these questions

[rustatian]

Hey @shieldz80 👋
If you specify a pipeline with a queue, RR will allocate only one consumer. However, you should specify a prefetch of 1 to synchronously consume messages. So, only after the message was acknowledged, RR will consume the next one.
And yes, RR supports exclusive consumers: https://docs.roadrunner.dev/docs/queues-and-jobs/amqp

[shieldz80]

Hi @rustatian,

Created #1941

Thank you

[rustatian]

Thanks, @shieldz80 👍
I'll expose the multiply option. I hope this help you.
I can also build a binary for you to test with multiply set to true, to see, if that would help.

[rustatian]

Wait, I don't see an requeue_on_fail option set to true in your configuration. Could you please retest with that option?

[shieldz80]

Hi @rustatian,

I can also build a binary for you to test with multiply set to true, to see, if that would help.

Yes sure, why not.

[shieldz80]

Wait, I don't see an requeue_on_fail option set to true in your configuration. Could you please retest with that option?

Unfortunately, the results are the same as observed before with requeue_on_fail: true (I've also mentioned this in the Feature Request)

Processing log for one of the queues

[2024-06-12T03:46:49.356968+00:00] q1.INFO: Processing message with id 1 from queue q1 {"id":1,"queue":"q1"} []
[2024-06-12T03:46:53.358140+00:00] q1.INFO: Message with id 1 from queue q1 was processed successfully {"id":1,"queue":"q1"} []
[2024-06-12T03:46:53.361934+00:00] q1.INFO: Processing message with id 2 from queue q1 {"id":2,"queue":"q1"} []
[2024-06-12T03:46:53.362141+00:00] q1.ERROR: RuntimeException: Processing failed in /home/consumer/app/src/Task/Handler/GenericHandler.php:45 Stack trace: #0 /home/consumer/app/src/Task/Handler/GenericHandler.php(32): Shieldz\ConsumerApp\Task\Handler\GenericHandler->doProcessing(3) #1 /home/consumer/app/consumer.php(39): Shieldz\ConsumerApp\Task\Handler\GenericHandler->process(Object(Spiral\RoadRunner\Jobs\Task\ReceivedTask)) #2 {main} [] []
[2024-06-12T03:46:53.363823+00:00] q1.INFO: Processing message with id 3 from queue q1 {"id":3,"queue":"q1"} []
[2024-06-12T03:46:57.364105+00:00] q1.INFO: Message with id 3 from queue q1 was processed successfully {"id":3,"queue":"q1"} []
[2024-06-12T03:46:57.364897+00:00] q1.INFO: Processing message with id 4 from queue q1 {"id":4,"queue":"q1"} []
[2024-06-12T03:46:57.364948+00:00] q1.ERROR: RuntimeException: Processing failed in /home/consumer/app/src/Task/Handler/GenericHandler.php:45 Stack trace: #0 /home/consumer/app/src/Task/Handler/GenericHandler.php(32): Shieldz\ConsumerApp\Task\Handler\GenericHandler->doProcessing(5) #1 /home/consumer/app/consumer.php(39): Shieldz\ConsumerApp\Task\Handler\GenericHandler->process(Object(Spiral\RoadRunner\Jobs\Task\ReceivedTask)) #2 {main} [] []
[2024-06-12T03:46:57.365597+00:00] q1.INFO: Processing message with id 5 from queue q1 {"id":5,"queue":"q1"} []
[2024-06-12T03:46:59.365825+00:00] q1.INFO: Message with id 5 from queue q1 was processed successfully {"id":5,"queue":"q1"} []
[2024-06-12T03:46:59.367722+00:00] q1.INFO: Processing message with id 6 from queue q1 {"id":6,"queue":"q1"} []
[2024-06-12T03:47:02.368020+00:00] q1.INFO: Message with id 6 from queue q1 was processed successfully {"id":6,"queue":"q1"} []
[2024-06-12T03:47:02.368880+00:00] q1.INFO: Processing message with id 7 from queue q1 {"id":7,"queue":"q1"} []
[2024-06-12T03:47:06.369010+00:00] q1.INFO: Message with id 7 from queue q1 was processed successfully {"id":7,"queue":"q1"} []
[2024-06-12T03:47:06.369764+00:00] q1.INFO: Processing message with id 8 from queue q1 {"id":8,"queue":"q1"} []
[2024-06-12T03:47:09.369919+00:00] q1.INFO: Message with id 8 from queue q1 was processed successfully {"id":8,"queue":"q1"} []
[2024-06-12T03:47:09.371341+00:00] q1.INFO: Processing message with id 9 from queue q1 {"id":9,"queue":"q1"} []
[2024-06-12T03:47:14.371596+00:00] q1.INFO: Message with id 9 from queue q1 was processed successfully {"id":9,"queue":"q1"} []
[2024-06-12T03:47:14.373173+00:00] q1.INFO: Processing message with id 10 from queue q1 {"id":10,"queue":"q1"} []
[2024-06-12T03:47:15.373583+00:00] q1.INFO: Message with id 10 from queue q1 was processed successfully {"id":10,"queue":"q1"} []
[2024-06-12T03:47:15.375500+00:00] q1.INFO: Processing message with id 11 from queue q1 {"id":11,"queue":"q1"} []
[2024-06-12T03:47:19.375756+00:00] q1.INFO: Message with id 11 from queue q1 was processed successfully {"id":11,"queue":"q1"} []
[2024-06-12T03:47:19.390277+00:00] q1.INFO: Processing message with id 12 from queue q1 {"id":12,"queue":"q1"} []
[2024-06-12T03:47:21.391060+00:00] q1.INFO: Message with id 12 from queue q1 was processed successfully {"id":12,"queue":"q1"} []
[2024-06-12T03:47:21.393342+00:00] q1.INFO: Processing message with id 13 from queue q1 {"id":13,"queue":"q1"} []
[2024-06-12T03:47:23.393757+00:00] q1.INFO: Message with id 13 from queue q1 was processed successfully {"id":13,"queue":"q1"} []
[2024-06-12T03:47:23.395396+00:00] q1.INFO: Processing message with id 14 from queue q1 {"id":14,"queue":"q1"} []
[2024-06-12T03:47:27.395655+00:00] q1.INFO: Message with id 14 from queue q1 was processed successfully {"id":14,"queue":"q1"} []
[2024-06-12T03:47:27.397606+00:00] q1.INFO: Processing message with id 15 from queue q1 {"id":15,"queue":"q1"} []
[2024-06-12T03:47:29.397886+00:00] q1.INFO: Message with id 15 from queue q1 was processed successfully {"id":15,"queue":"q1"} []
[2024-06-12T03:47:29.399989+00:00] q1.INFO: Processing message with id 16 from queue q1 {"id":16,"queue":"q1"} []
[2024-06-12T03:47:32.400372+00:00] q1.INFO: Message with id 16 from queue q1 was processed successfully {"id":16,"queue":"q1"} []
[2024-06-12T03:47:32.402220+00:00] q1.INFO: Processing message with id 17 from queue q1 {"id":17,"queue":"q1"} []
[2024-06-12T03:47:35.402455+00:00] q1.INFO: Message with id 17 from queue q1 was processed successfully {"id":17,"queue":"q1"} []
[2024-06-12T03:47:35.403190+00:00] q1.INFO: Processing message with id 18 from queue q1 {"id":18,"queue":"q1"} []
[2024-06-12T03:47:37.403457+00:00] q1.INFO: Message with id 18 from queue q1 was processed successfully {"id":18,"queue":"q1"} []
[2024-06-12T03:47:37.405668+00:00] q1.INFO: Processing message with id 19 from queue q1 {"id":19,"queue":"q1"} []
[2024-06-12T03:47:41.405954+00:00] q1.INFO: Message with id 19 from queue q1 was processed successfully {"id":19,"queue":"q1"} []
[2024-06-12T03:47:41.407684+00:00] q1.INFO: Processing message with id 20 from queue q1 {"id":20,"queue":"q1"} []
[2024-06-12T03:47:42.407894+00:00] q1.INFO: Message with id 20 from queue q1 was processed successfully {"id":20,"queue":"q1"} []
[2024-06-12T03:47:42.409138+00:00] q1.INFO: Processing message with id 21 from queue q1 {"id":21,"queue":"q1"} []
[2024-06-12T03:47:43.409408+00:00] q1.INFO: Message with id 21 from queue q1 was processed successfully {"id":21,"queue":"q1"} []
[2024-06-12T03:47:43.428206+00:00] q1.INFO: Processing message with id 22 from queue q1 {"id":22,"queue":"q1"} []
[2024-06-12T03:47:47.428758+00:00] q1.INFO: Message with id 22 from queue q1 was processed successfully {"id":22,"queue":"q1"} []
[2024-06-12T03:47:47.440209+00:00] q1.INFO: Processing message with id 23 from queue q1 {"id":23,"queue":"q1"} []
[2024-06-12T03:47:49.440940+00:00] q1.INFO: Message with id 23 from queue q1 was processed successfully {"id":23,"queue":"q1"} []
[2024-06-12T03:47:49.442779+00:00] q1.INFO: Processing message with id 24 from queue q1 {"id":24,"queue":"q1"} []
[2024-06-12T03:47:50.443154+00:00] q1.INFO: Message with id 24 from queue q1 was processed successfully {"id":24,"queue":"q1"} []
[2024-06-12T03:47:50.446913+00:00] q1.INFO: Processing message with id 25 from queue q1 {"id":25,"queue":"q1"} []
[2024-06-12T03:47:51.447201+00:00] q1.INFO: Message with id 25 from queue q1 was processed successfully {"id":25,"queue":"q1"} []
[2024-06-12T03:47:51.448520+00:00] q1.INFO: Processing message with id 2 from queue q1 {"id":2,"queue":"q1"} []
[2024-06-12T03:47:54.448823+00:00] q1.INFO: Message with id 2 from queue q1 was processed successfully {"id":2,"queue":"q1"} []
[2024-06-12T03:47:54.449815+00:00] q1.INFO: Processing message with id 4 from queue q1 {"id":4,"queue":"q1"} []
[2024-06-12T03:47:59.449968+00:00] q1.INFO: Message with id 4 from queue q1 was processed successfully {"id":4,"queue":"q1"} []