JetStream. "Queue group" feature with pull subscribers

[ruslan528]

Can we distribute mesages from one subject to dynamic pool of applications with pull readers? Now we do that using one push consumer with "Deliver Queue Group" specified and one subscriber with such group in each app. So message processed only in one available application. Can we do that using pull subscribers? It's important that:

1. Every message must be processed as soon as possible
2. No matter what app do that processing
3. Only one app must process one concrete message
4. We can have one, two or more apps in a moment.
   In Kafka it realized with "consumer group" feature.

[derekcollison]

Yes pull consumers will allow that behavior.

[ruslan528]

Please, describe in short hand how to do that with pull subscribers.

Or you meaning that in future? You wrote this comment nats-io/jetstream#103 (comment) 4 years ago. Is it stil not implemented? :(

[derekcollison]

You can add pull consumers for all messages or any subset that can be expressed by a subject (wildcard). You then run as many application binding to the same consumer.

When scaling back down we recommend draining the subscriptions.

https://natsbyexample.com/examples/jetstream/pull-consumer/go

[EvilaMany]

+1
How to implement it?

[EvilaMany]

In documentation you suggest to use pull over push. I just want to create a working queue with many dynamic workers. Currently, with pull subscription, I can see that single message may be consumped by multiple workers at the same time.
How to fix it?

[jnmoyne]

You need to create only one consumer, with a 'durable' name and then have all the instances of your client consuming application consume (using Consume() or Fetch() depending on your preference) from that same consumer and messages will automatically be distributed between those instances, nothing more to do or implement.

[EvilaMany]

You need to create only one consumer, with a 'durable' name and then have all the instances of your client consuming application consume (using Consume() or Fetch() depending on your preference) from that same consumer and messages will automatically be distributed between those instances, nothing more to do or implement.

I did exactly you discribed, but when I ran few workers at the same time, I can see that few of them consume SAME messages at the same time. It may probably be related to that I set timeout for pull - 5 sec, but not sure

[jnmoyne]

It's not possible for 2 clients attached to the same durable consumer to get the same message, unless the second one is a re-delivery (you can check the delivery count in the JS metadata of the message), you need to figure out why you are having messages re-delivered (e.g. ack wait is too low, or you are sending a negative Ack rather than a positive one, ...)

[EvilaMany]

yea, thank you. Finally got the reason, and it's redelivery and the fact, that in Python version of client lib ConsumerConfig -> backoff overrides ack_wait parameter. Because I had ack_wait=30, max_deliver=2, and backoff=[1, 60], final ack_wait became = 1 sec, and messages got redelivered almost instantly (because handing of tasks takes > 1sec).

Don`t know if another client libraries has the same defect, but such parameter overriding looks very strange and non intuitive

[jnmoyne]

Edit: it is actually the server's behavior that if you specify backoff then the first backoff value is used and overrides ack wait.

nats-server/server/consumer.go

Line 558 in 237b8a7

config.AckWait = config.BackOff[0]

.

So this is behaving as intended, maybe make it clearer in the docs (e.g. nats-io/nats.docs#828).