Consuming partitioned stream

[ealeykin]

Hi,

I see a lot of questions related to kafka-like partitioning and every answer mentions subject mapping to calculate/add partition number to a subject, however I don't see anything related to how to deal with consumption afterwards.

Let's say I partitioned subjects. Created multiple stream per partition (also why separate streams ?).

I have a service consuming messages for a specific subject containing some events from customers - customer id and also partition number - a part of a subject.

Now I want to increase replicas count - how a consumer/subscription/application should be configured to allow reading sequentially (by customer id order) from partitioned streams.

The question is only related to consumption and setup on client side. I have found no examples, only some assumptions related to KV.

Please advise.

Thanks

[jnmoyne]

For the partitioned consumption of messages you do not need to use more than one stream. You can just use one stream, use subject transformation in that stream to insert a partition number automatically as part of the subject (e.g. the first token of the subject) and then you can create a consumer per partition (using a subject filter that only matches the messages with that particular partition number). You could even have one consumer per instance of your consuming application and have multiple partitions per instance.

There is an implementation of all this (Golang) that is all ready to go (initial version of course) and that I am hoping to be able to release soon (it will be in the Orbit repo), but unfortunately while 2.11 has now finally been released I'm still waiting for some of it's new features to be exposed in the nats.go client library before I can release that consumer group library.

But essentially this is all described in this PR nats-io/nats-architecture-and-design#263

[ealeykin]

@jnmoyne Thanks for the info, very interesting.

In your overview you mention it uses 2.11 pinned consumer? Let me try to recap your static approach:

Having pinned consumers available in 2.11 it's noW possible to:

1. Create N consumers per stream (N - number of partitions) - e.g. my-consumer-group-1 ... my-consumer-group-N
2. Each consumer has a filter to pull a single partitioned subject - e.g. my.subject.0 .. my.subject.N
3. A client have to subscribe to all consumer and pin somehow, so that only one consumer will pull messages for each partition, others will be in stand by mode for this partition, but at the same time will be consuming other partitions.

Elastic approach - more or less the same but has an intermediate stream.

Is that correct? If yes - that means consuming application have to create N threads to pull from N consumers. This last bit is not completely clear.

[jnmoyne]

Yes, it uses the new 'pinned consumer' functionality of 2.11 in order to provide fault-tolerance for the client instances of a consumer group member.

For the static approach, the stream contains a partition number in the first token of the subject already, using the library you can create one (or more) 'static partitioned consumer group'(s) and define a 'membership' (list of member names (e.g. "m1 m2 m3")) for the consumer group, as well as a 'max number of members' (which is actually the number of partitions) then the library takes care of distributing the partitions between the members of the group such that each member (this is done automatically by the library) has a number of partitions assigned to it and it creates a consumer on the stream with the list of filters corresponding to the list of partitions that are assigned to it.

The pinned consumer functionality means you ran run more than one instance of a member (e.g. run 2 instances of your client consuming application for member 'm1') but only one of those instances can be the pinned one at a time, until you administratively force the pinned consumer to step down, or it crashes, or gets suspended, etc... in which case one of the other instances of the member (i.e. the second instance of the consuming application for member 'm1') becomes the pinned one and starts receiving messages.

The consuming application just calls 'consume' from the library and that creates a single go routine to consume messages from all the partitions that were assigned to that client member.

Elastic: the stream doesn't already have a partition number as part of the subject of the messages in the stream so it creates an intermediary stream (work queue) that sources from the stream, adds a partition number subject token, and instances of consuming client applications simply join the consumer group and the library automatically creates the JetStream consumers for the members of the consumer group and automatically adjust those consumer's list of subject filters as the membership of the elastic consumer group changes.

[jnmoyne]

A client side library implementing static and elastic 'partitioned consumer groups' functionality has now just be released in Orbit.go: https://github.com/synadia-io/orbit.go/releases/tag/pcgroups%2Fv0.1.0 (and https://github.com/synadia-io/orbit.go/blob/pcgroups/v0.1.0/pcgroups/README.md)