Jetstream Consumers - Creation of Consumers on Stream with replication takes more time

[arunsmenon88]

Hello NATS team,
For a prototype, we have deployed a 3 instance NATS cluster ( all 3 instances running on same machine)
We are trying to implement a use case which would need consumers to be dynamically created and deleted on a NATS stream (Stream replication is set to 3).
We observe considerable delay in the Consumer create operation when the stream is replicated. Could you please review the data and suggest how this operation can be optimized?

As a test case, we tried creating 1000 streams with 1 consumer each (ie, 1000 consumers).
A percentile histogram of 1000 Consumer create operation using Golang client API "CreateOrUpdateConsumer" looks as :

Summary of time taken to create consumers
Percentile 10: 172671 microsseconds 
Percentile 25: 289535 microsseconds 
Percentile 50: 452863 microsseconds 
Percentile 95: 767487 microsseconds 
Percentile 99: 797183 microsseconds 
Percentile 99.9: 803839 microsseconds 
Percentile 99.99: 803839 microsseconds 
Percentile 100: 803839 microsseconds 
Number of entries :  1000

If we run the same test without replicating the stream, we see the results as

Summary of time taken to create consumers
Percentile 10: 556 microsseconds 
Percentile 25: 603 microsseconds 
Percentile 50: 675 microsseconds 
Percentile 95: 934 microsseconds 
Percentile 99: 1059 microsseconds 
Percentile 99.9: 1824 microsseconds 
Percentile 99.99: 4859 microsseconds 
Percentile 100: 4859 microsseconds 
Number of entries :  1000

As seen in the values, we see a huge latency in the create operation when we have replication enabled ( and is set to 3)

the code snippet used for this test is :

         for consumerId := 0; consumerId < noOfConsumers; consumerId++ {
		streamName := fmt.Sprintf("CONSUMER%d", consumerId)

		subject := fmt.Sprintf("consumer.%d", consumerId)
		cfg.Name = streamName
		cfg.Subjects = []string{subject}

		streamCreateStart := time.Now().UnixMicro()
		stream, err := js.CreateOrUpdateStream(ctx, cfg)
		if err != nil {
			log.Println("Unable to create stream : ", err)
			os.Exit(1)
		}
		streamCreateEnd := time.Now().UnixMicro()
		streamCreateSamples[consumerId] = Sample{consumerId, (streamCreateEnd - streamCreateStart), true}

		time.Sleep(500 * time.Millisecond)

		consumerName := fmt.Sprintf("consumer%d", consumerId)
		consumerCreateTimeStart := time.Now().UnixMicro()
		_, err = stream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
			Durable:       consumerName,
			FilterSubject: subject,
			AckPolicy:     jetstream.AckExplicitPolicy,
		})
		if err != nil {
			log.Print("unaable to create consumer ->", err)
			os.Exit(1)
		}
		consumerCreateTimeEnd := time.Now().UnixMicro()
		consumerCreateSamples[consumerId] = Sample{consumerId, (consumerCreateTimeEnd - consumerCreateTimeStart), true}
	}

Our test bed details are :

• nats-server version v2.10.12
• linux distro Rocky Linux 8.9
• 32 GB RAM , 12 core , 13th Gen Intel(R) Core(TM) i7-1355U 1.70 GHz

[Jarema]

Hey.

The differences comes from the fact, that replicated stream needs to create a Raft Group and assign nodes for all replicas.
That process can take some time, while single replica stream just creates it on one node and returns immediately, as it does not require Raft group creation and communication for it's operations.

Keep in mind that creating 1k streams with 1 consumer on each looks like a pattern that could be improved.
Having less streams, each with filtered consumers might be better idea, depending on the needs.
It also gives you more flexibility which consumers are R1 or R3, for an R3 stream.

[arunsmenon88]

@Jarema Thanks for the suggestion, As you suggested , i may not need 1K streams (that was just a test i tried to understand the latency involved in create)
1 stream, with filtered consumers will work good for us. But we dont want these consumers to be statically created (ie, created once and used everywhere).
Basically, we are trying to implement a work queue model, where we can create a consumer dynamically on a subject (or subject filter), process the messages in the subject and then delete the consumer. The then application can wait for a trigger for next such work available in another subject (or subject filter) and repeat the same.
I was trying to explore the below 2 options:

1. On input side, i have a common workqueue subject and a list of individual work id subjects (like workqueue, work.1, work.2 .. etc) with work appearing randomly for any work id.
2. Have a common consumer accross my applications to fetch from a workqueue stream/subject 'workqueue' (which acts as aload balancer) which ensures that the message in 'workqueue' is delivered to only one application at a time

Now, i can think of either of 2 options below (to achieve flow mapping of the specific work.* subject to a particular consumer for some time):
3.a. Once the application recieves the workqueue message to indicate the presence of work in one of the worksubjects (like work.1, work2 .. etc) , then i shall create a consumer with subject filter as the specific work queue (eg: work.1).
But, it seems like the time taken to create/update a consumer at this point is ~500ms since its an R3 stream.

3.b Once the application recieves the workqueue message to indicate the presence of work in one of the worksubjects (like work.1, work2 .. etc) , then i can try to have a new stream with Source pointing to workqueue stream with subject transform added to filter only the specific work subject (eg : work.1). I have to still explore if this is feasible and the latencies involved here (still not sure how many transform rules can be configured and also the time taken to create a stream also should be accounted for)

To achieve flow mapping dynamically, i couldnt find any other methods (im already aware of the Subject Mapping and Partitioning, but we dont want static # based partitions and mapping here since the work.* can appear dynamically for any work id and we want to be flexible to map it to any consumers at runtime)
Kindly suggest

[Jarema]

I'll dig the details later, but it seems that "Consumer Groups", "pinning consumer" could be of some use here for the part when you want to map consumers at runtime?
nats-io/nats-architecture-and-design#279

It is something we work on right now.
We will have some client side utilities for that too.

[arunsmenon88]

Thanks @Jarema
From a quick glance of nats-io/nats-architecture-and-design#279, some of the properties of Consumer Groups is what we are looking for.
But, i would try to go through in detail and try to understand the proposal, especially with respect to a active Consumer failure, scale up, scale down

[ripienaar]

Such adhoc single subject stream walks are best done with the Dodext Get API. It can, without consumers, get all messages for a subject one by one (later in matches)

https://github.com/nats-io/nats-architecture-and-design/blob/main/adr/ADR-31.md#request

Depending on language the implementation may vary but this is the cheapest way to walk just one subject in a stream.

[arunsmenon88]

Thanks @ripienaar
If i understand right, this allows an application to read messages from a specific subject in a stream without actually creating a consumer on that stream with the requried subject filter?
If yes, on a stream with WorkQueuePolicy, would this also allow me to Ack the messages i have read , so that those messages are deleted from the stream (to allow exactly-once delivery semantics) ?

[ripienaar]

No for the WQ you wouldn’t be able to ack - but you could delete the message of course

[arunsmenon88]

This approach would work well for us when there are only very few subjects to poll messages from.
We could still manage the start sequence of messages per subject and deleting the messages from stream after processing them.

But , in a highly loaded system, we would potentially have subjects scaling up to > 10000 (eg : work.1, work.2 , ..., work.10000). And, consumers can dynamically handle a set of such work subjects (for eg : consumer-1 reading from 1000 work subjects), which we can make sure are mutually exclusive accross the available consumers (like assignign the work subjects to consumers using a separate work queue).
When using direct read , we have to invoke GetMsg for each of these work subjects which might add a lot of latency in processing the messages in the work subjects.

Since we want the consumers to be assigned to work subjects dynamically, we are not keen to use the static # method for subject partitioning (which might result in underterministic behaviours when consumers scale out and scale in)
Hence, something like a dynamic mapping of work subjects to consumers using either Subscription/Consumer or Configurable Subject transform might help us?
(This was the motivation to initiate nats-io/nats-server#5463)

[arunsmenon88]

We are trying to explore the possibilities to implement a work queue solution where we are able to map incoming work items to consumers (who will process the incoming work items and then wait for futher work items). We also came accross and old discussion thread #1635 and associated blog post https://docs.kraudcloud.com/blog/2023/11/22/exclusive-worker-tokens-with-nats/
We also want to implement something similar, but with mapping a work flow to a dedicated consumer for a deterministic amount of time instead of randomly distributing all messages in the work flow to random consumer within that time frame. Hence, im trying to explore how to map a particular work subject to a consumer till all the messages for that work item are processed by the same consumer (and no other consumer reads from the work subject till this consumer is done with processing the messages from this work subject)
Would this be possible?