Distribute all partitions to a delivery group & trigger rebalance events on disconnected consumers.

[Tazer]

Hello,

We are looking into a PoC with NATS (currently using RabbitMQ and some PoC with Kafka).

The only problem I'm not being able to solve theoretically at least is:

Background
We have some services that are not really good at "horizontal scaling" cause they need to keep an in memory state, and load it on startup.
So they are stateful. This mean that it's fine IF only one consumer is working a single partition.

Kafka solution
The kafka solution is to use partition key and then each "connected" consumer in the group will gets 1..n partitions.
IF a consumer is disconnecting then there will be a rebalance event for all consumers so remaning consumers get 1..n partitions and each consumer can "reload states"

NATS Jetstream
Feels its close with Subject Mapping & Partitioning and using a DeliverGroup for the consumers.
The missing parts are:

• Can I distribute all existing partitions to connected consumers but only like consumer 1 get partition 0 and 1 and consumer 2 get partition 2 and 3.
• IF consumer 2 disconnects, is there a good way to listning for those events (consumer disconnect in the client) so actions can be taken for consumer 1 to stop, reload state , start processing all 4 partitions.

There are some talk about something similar here but not the exact use case. #2043

If it doesn't make sense I can try to clarify any parts.

[jnmoyne]

Hi @Tazer,

At the moment you would have to rely on your container orchestration (or something as simple as shell scripts) to start exactly n instances of your consuming process each one working against its own partition stream+consumer. If one of those consuming processes crashes or gets killed the container orchestration system should restart them and as long as you are using durable consumers when the consuming application gets restarted and reconnects it will continue from the last message it received. If you need to reload state each time when you restart one of those consuming applications then you would re-create the consumer for the partition each time. To detect if one of those consuming processes gets hung and stops pulling messages you would regularly get the consumer info for the n consumers and you can tell 'how far behind' the consumer is WRT the head of the stream, or you could also use the 'working queue' discard policy for the stream and look at the size of each partition's stream.

[Tazer]

Think what we want to-do is more dynamic. I would like to run 10 partitions, but I only want to run 1 consumer. until the load is big enough so we can then scale to more consumers running a few partitions each.

It's fine if it's not in core nats/jetstream , but would it be possible to use System Events https://docs.nats.io/running-a-nats-service/configuration/sys_accounts ?

Guess what I'm looking for is jetstream publishing events to internal streams that can be subscribed or key values, so we could build rebalance events and other logic in our consumer logic?

Cause I don't see how we can just relay on container orchestration like kubernetes. think we actually need smart "consumers" but that also means we need to get events from what's happening internally in jetstream?

[jnmoyne]

So the reason you want 'only one consumer per partition' is because the client application needs to receive all the messages in that 'partition' in order to maintain it's own state, right?

If that's the case you can maybe do what you want to do without having to worry about any kind of partitioning:

Have your client application code do the following: create 2 subscriptions to the stream.

• One subscription is to an ephemeral consumer on the stream with replay policy 'all', the messages received on this subscription are only for building up and maintaining state.
• One subscription is to a durable shared consumer (i.e. a pull consumer), the messages received on this subscription are used to trigger the processing, and the stream replay policy is 'new'.

This means that you can then start as many instances of your application as you want, and each instance uses it's own ephemeral consumer to get a copy of every message in the stream in order to maintain it's state, and all of the instances share the same durable consumer to distribute the processing of those messages.

So that when the application starts it first receives all the messages currently in the stream (thought the ephemeral consumer subscription) and uses that stream of messages only to build/maintain it's state. Then if it receives the same message also on the subscription to the durable consumer (you can use the JS sequence number in the message's metadata for that) then it means it should also process that message.

This way you can start-stop your client application process instances at any time, run as many instances as you want and not have to worry at all about partitioning, or re-partitioning.