[Other] Allow for competing consumers to the same stream? How do we implement this?

[ThomasShih]

Community Support Policy

• I have read RabbitMQ's Community Support Policy

RabbitMQ version used

4.0.3

How is RabbitMQ deployed?

Community Docker image

Steps to reproduce the behavior in question

We are currently using rabbitmq streams for event messaging, messages are published onto the stream with various headers (such as topic type and action type). We have multiple consumers that consumer from the stream and process messages that they care about with stream filtering. Each consumer only care about a subset of the stream messages.

Thinking about the future, we may need to scale up some consumers. It makes sense that we can vertically scale the consumers, but how would we go about horizontally scaling consumers in such a way that they are not repeat processing messages?

I would assume that if we choose to horizontally scale consumers, we will lose the ordering guarantee that streams provide, but for certain consumers, we know that this is a tradeoff that we are able to make.

---

The requirements/assumptions are:

• The messages I care about are a subset of messages within the stream that has a certain combination of headers for which I can filter
• I don't care about ordering
• Each message should be processed only once, while this isn't a hard rule, it should behave the same way as how multiple consumers compete for messages in queues

A few ideas that I have:

• A proxy consumer that is hooked up directly to the stream, that 'pull' the messages to a separate queue, and then we use the classic AMQP way of processing messages with multiple consumers
• A consumer that has knowledge of how many other consumers exist for the same task, and then hashes the stream pointer position to only care about a subset of the messages we care about. Ie if there is two consumers, one would only care about even position messages and the other only odd position messages

---

Not sure if I'm describing the problem well, but I'm wondering if anyone has better ideas on the idea of horziontally scaling consumers?

[michaelklishin]

Consumers on a stream are competing (with repeatable non-destructive reads) by definition, unless the Single Active Consumer is enabled.

You already know about stream filtering (plus a blog post on the topic). There's nothing wrong with your idea of a "dispatching consumer" that would republish some messages but I'm not sure I understand why the combination of regular (non-SAC) consumers would not be sufficient if you use a stream protocol client.

[ThomasShih]

Thank you for the reply!

I think the key thing is we want distructive reads (if by that we mean the 'ack' success option from consumers for regular queues) for a specific consumer type

We are essentially using the stream as a message bus, where a message might be consumed by 0,1, or multiple consumers, and consumers can filter off any header combinations they have. A specific filter may also apply to multiple consumer types.

---

A good example I can think of is say a txt/email notification, we have a consumer for txt notification, and a different consumer for email notification. When a message with say a header of type:notification goes through, the side effect we expect is for me to receive both a text message and a email.

And lets say the email notification consumer takes 2x as long, the hope is for us to have 1x txt consumer and 2x email consumer per say.

But instead of the email consumers sharing the same messages, to effectively partition the messages in such a way that each email consumer is responsible for half of the messages.

In this particular case, we don't care about ordering for email consumers, we just care that the emails are sent.

---

I'm not sure I understand why the combination of regular (non-SAC) consumers would not be sufficient if you use a stream protocol client.

Say we also want to add an notification log, that essentially records which type:notification messages have been published to the stream, then this notification consumer will use the same filters as txt and email, but in this case, we may want to perserve ordering.

---

Am I overthinking this? Is the best use case to do as you say with using a dispatching consumer?

[michaelklishin]

If you want destructive consumption then streams, by definition, are not what you are looking for.

Although you can make publishers publish to a stream, then have a "dispatching app" relay some messages to N queues each with C competing consumers.

I don't see anything wrong with this idea except that a single stream will outpace quite a few quorum queues, and such a setup will require a reasonable number of cores on each node (4 seems to be a realistic minimum to me given the parallelism expectations).

[michaelklishin]

Am I overthinking this?

I don't have enough context to offer an informed opinion but…

Is the best use case to do as you say with using a dispatching consumer?

Well, streams alone won't be enough and such "multi-tier" topologies is one of the reasons why AMQP 0-9-1 and RabbitMQ were originally developed :)

So I think this idea has legs.

[ThomasShih]

Thank you!

[kjnilsson]

Not sure if I'm describing the problem well, but I'm wondering if anyone has better ideas on the idea of horziontally scaling consumers?

RabbitMQ Super streams may be an option although you have to pre-partition you workload so wont be able to scale consumption dynamically but you can retain order based on the key you use for partition selection.

https://www.rabbitmq.com/docs/streams#super-streams

[ThomasShih]

Thank you! I guess we need to make the design choice between horizontally partitioning the stream itself or use a dispatching app as a proxy as mentioned in #13794 (reply in thread). We will give it a thought :)