Description
Symfony version(s) affected: 4.3.0+
Description
As far as I've seen it isn't possible to have two consumers on the same transport. We have the possibility to define the stream, group and consumer per transport. But none of this config options allow to have two or more consumer commands messenger:consume
running on the same transport.
How to reproduce
- Use a config similar to this:
framework:
messenger:
transports:
async_priority_high:
dsn: 'redis://redis-server/priority_high'
- Run two or more consume commands in parallel for the same transport
bin/console messenger:consume async_priority_high
- Send a message through the
async_priority_high
transport.
Now there is the chance that one of the commands exits with an exception like:
Symfony\Component\Messenger\Exception\TransportException: Could not acknowledge redis message "1579079882286-4".
This also means that the message was consumed twice.
Unfortunately this happens because the message is only being acknowledged (Redis XACK
) after the message is handled and leaves it for the duration of the handling in the stream to be consumed by the other consumer command too.
EDIT: This is not true. The issue does only happen once if the two consumers are started around the same time and at least one message is already in the queue.
Possible Solution
Using redis XACK
right after XREADGROUP
in Symfony\Component\Messenger\Transport\RedisExt\Connection:get()
and using the Symfony\Component\Messenger\Transport\RedisExt\Connection:reject()
method to re-add the message to the redis stream.
EDIT: With the edit above, this isn't a solution.
Additional context
I am sorry if I did get the usage of messenger wrong, but IMO having multiple consume commands running at the same time should be supported.