Streams feature request: consumer groups

[WeirdBob]

Hi team,

First of all, congrats for this new version, streams looks very promising and we look forward to tear down our kafka cluster and use only rabbitmq for all our messaging needs.

One key feature which will prevent us from going to production with this version is the lack of a minimal implementation of consumer groups, ie being able to declare multiple consumers as a group and having rabbitmq decide which one should consume (active/passive election) for high availability purposes. I would guess your raft implemetation would be a great way to solve this!

The sister feature would then be partitioning to allow multiple consumers to share the work on a stream while keeping the natural ordering of messages. This can be worked around by creating multiple streams and their associated consumers, even if it's not as transparent as having it natively supported by the cluster.

Again, thanks for everything, Rabbitmq has been my go-to message broker for the last decade, and streams are a very nice addition to its already rich list of features.

[kjnilsson]

Would something akin to the single active consumer feature found in other RabbitMQ queue types work for you if it was available for streams?

On Fri, 13 Aug 2021 at 09:50, WeirdBob ***@***.***> wrote: Hi team, First of all, congrats for this new version, streams looks very promising and we look forward to tear down our kafka cluster and use only rabbitmq for all our messaging needs. One key feature which will prevent us from going to production with this version is the lack of a minimal implementation of consumer groups, ie being able to declare multiple consumers as a group and having rabbitmq decide which one should consume (active/passive election) for high availability purposes. I would guess your raft implemetation would be a great way to solve this! The sister feature would then be partitioning to allow multiple consumers to share the work on a stream while keeping the natural ordering of messages. This can be worked around by creating multiple streams and their associated consumers, even if it's not as transparent as having it natively supported by the cluster. Again, thanks for everything, Rabbitmq has been my go-to message broker for the last decade, and streams are a very nice addition to its already rich list of features. — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#3306>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAJAHFDYOT23SAHJWWGOWSLT4TME7ANCNFSM5CDE7CLQ> . Triage notifications on the go with GitHub Mobile for iOS <https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675> or Android <https://play.google.com/store/apps/details?id=com.github.android&utm_campaign=notification-email> .

-- *Karl Nilsson*

[WeirdBob]

Hi Karl,
If by "single active consumer" you mean one consumer is active and if it dies another one takes its place, yes, it would be sufficient for me. I don't recall having this feature in the classic queues, it's not the same as priority consumers where it's hard to ensure the messages are processed in the right order.

I assume some other would appreciate the higher throughput of partitioning but I don't think I'll need it soon (we only process a few million messages a day, so we don't stress the clusters)

[kjnilsson]

Both classic and quorum queues support single active consumers and we are considering the same for streams. https://www.rabbitmq.com/consumers.html#single-active-consumer

On Fri, 13 Aug 2021 at 10:16, WeirdBob ***@***.***> wrote: Hi Karl, If by "single active consumer" you mean one consumer is active and if it dies another one takes its place, yes, it would be sufficient for me. I don't recall having this feature in the classic queues, it's not the same as priority consumers where it's hard to ensure the messages are processed in the right order. I assume some other would appreciate the higher throughput of partitioning but I don't think I'll need it soon (we only process a few million messages a day, so we don't stress the clusters) — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#3306 (reply in thread)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAJAHFHRZNWIUKKE4X4F6QTT4TPFPANCNFSM5CDE7CLQ> . Triage notifications on the go with GitHub Mobile for iOS <https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675> or Android <https://play.google.com/store/apps/details?id=com.github.android&utm_campaign=notification-email> .

-- *Karl Nilsson*

[WeirdBob]

For streams, the single active consumer would still be "single in a given group" but multiple active consumers from different groups?
The use case is a classic event sourcing one, where we have multiple type of consumers doing different things at the same time on a given stream, but for a specific type of consumer only one instance would be active.

[kjnilsson]

We could consider multiple single active consumers over a single stream yes. We don’t have a fixed design yet.

On Fri, 13 Aug 2021 at 15:18, WeirdBob ***@***.***> wrote: For streams, the single active consumer would still be "single in a given group" but multiple active consumers from different groups? The use case is a classic event sourcing one, where we have multiple type of consumers doing different things at the same time on a given stream, but for a specific type of consumer only one instance would be active. — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#3306 (reply in thread)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAJAHFG5EE537PTWCM5VNYTT4USUHANCNFSM5CDE7CLQ> . Triage notifications on the go with GitHub Mobile for iOS <https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675> or Android <https://play.google.com/store/apps/details?id=com.github.android&utm_campaign=notification-email> .

-- *Karl Nilsson*

[WeirdBob]

That would be a very useful feature indeed!

[adminnz]

We are excited about Streams as well, but currently the lack of partitions and consumer groups is a blocker for us.
Being able to define a stream with N partitions (based on a routing key would be nice) and then either:

• Supporting up to N consumers in the same "consumer-group", where partitions are automatically assigned to consumers to handle.
  So if 3 partitions and only 1 consumer in the group, all partitions are assigned to it. When a new consumer joins the consumer-group reallocated the partitions from the consumer that has the most partitions allocated. Repeat until N consumers joined.

• Support any number of consumers, where they behave in a competing consumer model per partition they are allocated.

[WeirdBob]

Hi @acogoluegnes
Maybe our way of thinking is biased by the way kafka works and there may very well be other solutions to this problem than partitions.

A typical use case is event sourcing:

• We have to ensure messages for a given client are processed in order (we must avoid a consumer processing a "modified" event while another one is still processing the "created" event)
• Consumption has to be highly available (if a consumer dies, another one should be able to resume processing the messages at the last offset committed by the dead consumer)
• We may need to scale out to process different clients in parallel (because processing a message can be slow, not necessarily because of the throughput of messages).

In kafka, the 2 first points are covered by consumer groups and leader election, for the third one we create mutiple partitions and use the client id as a partitioning key.

How would you propose to enforce message ordering in the case of parallel/competing consumers?

[acogoluegnes]

With competing consumers you would lose the message ordering.

Thanks for the clarification of your use case. We are thinking about 2 other features that combined could meet your expectations: single active consumer and stream partitioning. The semantics of single active consumer in streams would be roughly the same as with queues. Partitioning would be a client feature as it mostly implies routing.

These are in early design stages, so no promises and ETA yet.

[aivans]

@acogoluegnes the points @WeirdBob mentioned in the previous posts are still very relevant and the design reference is Kafka.

I looked at the SuperStream feature which provides Stream partitioning. This is a step forward and improves performance.

Are there any plans to develop a Consumer Group like feature? The SuperStream feature allows for a single process to consume all partitions. It would be very useful to spread the consumption across multiple processes for even better performance, scalability and HA.

Thanks

[michaelklishin]

There are plans to finish several stream features in flight and then see. We do not make promises on our roadmap.

[aivans]

Thanks @michaelklishin Looking forward :)

[aivans]

One more note.
It is hard to adopt Streams in a multi-instance service setup. At any given point in time, only one instance must be enabled to consume from the stream. Considering that usually the baseline configuration is identical for all instances, an additional mechanism needs to be in place such that only one instance starts consuming (a shared distributed lock or another alternative)

Practically, I find Stream useful:

• in the "Large fanout" use cases where different Consumers/Services subscribe to the same data
• as a replacement for dedicated exclusive queues in multi-instance service setup where every instance must update an internal cache

I am not expecting a commitment on the roadmap, but are there any current concrete plans to improve this scenario?

[michaelklishin]

#3754 is that mechanism

[chicco785]

In the use case where messaging ordering does not matter, the single active + superstream features pose some limitations to the ability to scale up processing. What are the options here? any guidance would be of help :)

Suppose your consumers are k8s containers and they all do the same annoying processing and that ordering of messages is not important. Ideally, by auto-scaling consumers with the same id, you would increase the parallel throughput in case of processing that requires long times. Achieving this with single active + superstream features somehow imposes to predefine the maximum number of parallel processors as the number of partitions of the super stream.
Is there a way to have an arbitrary number of parallel consumers receiving different messages?

Thx!

[chicco785]

@kjnilsson thanks a lot!

few questions came to my mind following your reply:

1. what happens if your ingress rate increase, can you dynamically increase the number of partitions? (without deleting and re-creating the stream)
2. is there a tradeoff when having an high number of partitions in case your ingress rate is not yet up to the maximum level for which you computed the number of partitions?

using normal queues probably is not an option in our case (if i am not wrong there is quite a difference compared with streams on the throughput capacity, and we need 250.000++ message/sec to start from)

[kjnilsson]

1. No at least not currently that is why I suggest you create more partitions than you need to cover potential growth.
2. Everything has a resource impact and RabbitMQ "partitions" are streams. You probably don't want 1000 partitions but 10 would probably be fine maybe even 100. How many do you think you might need to cover the throughput you expect?

[chicco785]

1. That maybe a bit complicated, we have projects that start with a number of sensors, that depending on the success of the PoC, lead to deployment of additional sensors. This additional sensors are unknown at PoC time since, it is not only about the size of the grid to monitor but the evaluation by the customer of the trade-off between deployment of additional sensors and the acceptable precision of estimated network state, which is also evolving with legal regulations.
2. I am not much concerned about the resource cost, but about the fact that having at a given time more partitions than the one needed may cause some performance degradation to the streams. I don’t have yet precise measurement of the processor, but it should in the order of 100.000 messages per second (may have some ideas to get up to 500.000 but there will be trades off).

[kjnilsson]

Partitions are just streams and super streams are just streams bound to a particular exchange. You can do this manually if you create a stream using the convention used by super-streams then bind it to the exchange like the others then restart your apps they may just start using the new stream as a super stream partition. We haven't tested this but don't see why it won't work.

You will give up on ordering in this case however (which is why this isn't a feature right now).

[chicco785]

thanks a lot for the insight :)

[huangjiatian]

"it is important to understand that, in AMQP 0-9-1, messages are load balanced between consumers."
@kjnilsson it seem stream break this behavior?

[kjnilsson]

where are you getting this quote from?

yes streams don't do consumer load balancing, this is well documented

[huangjiatian]

thanks, i get this quote from Basic Concepts About Rabbitmq

[ioware37]

https://pulsar.apache.org/docs/3.2.x/concepts-messaging/#subscriptions,
the idea may matched the shared subscription.