How to create queue group consumer with new jetstream api.

[gudron]

Hello...i am looking for new jetstream API and i cant find replacement flow for old API.

How to create queue-group consumer?
How to create channel based consumer?
How to create push-type consumers?
How to create consumer for one-to-many communication model?

How long NATS community will support for old API?

old API just bttr...sorry

[Jarema]

The new JetStream API leverages fully the Pull Consumer.

1. You do not need to create queue-groups with Pull Consumer. It will load balance the load across all subscriptions to that Consumer.

2. The new API provides Messages, that delivers messages when calling Next, which can deliver the same uses cases that channel based did.

3. No push consumer support as of now, as with improvement of Pull Consumers over the time, the use cases between Push and Pull overlapped.

4. One to many:
   You mean one publisher and many clients receiving all messages?
   First of all, I would consider if that requires JetStream at all. If yes, and you need persistence, the solution is to create many consumers on the stream. Each will get its copy of the messages.

5. There are no concrete plans to move old API out of support. Howewer, some new featues, like multiple consumer fitlers, are not very ergonomic to use with old API. Also the old API does not allow for immediate consumption of messages for Pull Consumers (user need to aggregate all messages received in Fetch, while new API delivers messages as they arrive), nor it has a way to continously get the messages (user have to loop Fetch). Some new features might land only in new API.

[jnmoyne]

Functionally the new JS API has almost all of the functionalities of the old API (the only missing being 'push consumer without a queue group', which in my experience is never (or almost never) needed) and some new ones (e.g. support for acks=none or acks=all). And the performance is in most cases equivalent or better with the new API than with the old one.

[gudron]

(the only missing being 'push consumer without a queue group', which in my experience is never (or almost never) needed)

push consumer without a queue group is the easiest way for Choreography SAGA pattern ... sagas on pull-type consumers...srsly?

[Jarema]

@gudron both Push and Pull consumers serve as the API for the same server construct - Consumers.

The main difference is how the handle backpressure, by triggering messages requests versus flow control.
Pull Consumer is much more efficient at that.
They also have some difference how you load traffic (many clients vs many clients with queue-groups), but the result is functionally, again, the same.

From user perspective, there is no use-case difference between Consume/Messages on Pull Consumer, or Subscribe on Push.
Both provide you messages at quick rate, with the same latency, but Pull at the same time is much better at handling backpressure.

Please elaborate how those internal differences affect SAGA pattern.

[eau-u4f]

related to that conversation I think, is it possible to have one publisher message, consumed only by ONE of the available consumers and NOT all of them in a reliable fashion?

1. producer send one message to stream.
2. n workers are consuming the stream.
3. the first worker to consume the message is the one processing it, no other worker would receive that message.

[jnmoyne]

That is exactly how a 'durable' pull consumer works, you can have n workers pulling from the same durable (using it's durable name) consumer and the messages will be on-demand distributed between the workers pulling from the durable consumer, that message only gets sent to the worker that pulled it (unless the consumer requires acks in which case the message could be sent more than once if it's not acknowledged by the worker in time (or if it is negatively acknowledged by the worker)). Alternatively you get the same behavior using a durable push consumer with DQ (but that's only an option if you are using the old Jetstream API, at this point you should just be using the simplified JetStream API which only uses pull consumers under the covers).

[eau-u4f]

the consumer ack policy then should NOT be AckExplicitPolicy, correct?

I am confused, as WorkQueuePolicy requires AckExplicitPolicy..

I probably have misunderstood the acknowledgement, I was under the impressionion that Ack() mean the message IS completed, but InProgress() would update it expiry and hence prevent delivery to another worker (if the current worker fails) until the worker completes its job and Ack() / DoubleAck() the message to say ok, this one is DONE.

[jnmoyne]

A durable consumer can have any ack policy. The difference with WorkQueuePolicy is that with work queue you can only have one consumer (on which many workers can consume messages and those messages are distributed between them) and once the messages are acknowledged by the worker they are deleted from the stream - so the stream operates like a proper queue. While with the regular limits policy you could have more than one durable consumer defined on the stream (and for each durable have many workers getting messages from the consumer) - a single message will be sent to only one worker per consumer but a copy of this message will be seen by all the consumers on the stream.

This may help: https://docs.nats.io/nats-concepts/jetstream/consumers#persistence-durable-ephemeral

[jnmoyne]

With explicit ack policy the message is delivered once to a worker, at that time the 'ack wait' timer starts, if the worker doesn't positively acknowledge the message within the 'ack wait' period then that message will be available for re-delivery to another worker (or the same worker) (you can check the delivery count in the JetStream metadata for the message when you receive it). If the worker knows that the processing is taking a long time and could potentially exceed the 'ack wait' time and it still needs more time to process the message, then it can send an 'in progress' which will reset that ack wait timer (avoiding a potential re-delivery of the message once the ack wait times out).

[eau-u4f]

@jnmoyne : no matter the policy I'm trying (when in doubt use brute force) as consumer durable or not, when I send message to my task stream, all my "workers" (worker being my small client test program waiting for messages to simulate processing), like:
$ ./worker & ; ./worker & which now both set the same Durable Name to try....

Here is my stream configuration:

        stream, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
                Name:              "TASKS",
                Description:       "Task queue stream",
                Subjects:          []string{"tasks.*"},
                Retention:         jetstream.WorkQueuePolicy, // Important: Work queue ensures exclusive consumption
                MaxAge:            24 * time.Hour,            // Tasks expire after 24 hours
                Storage:           jetstream.FileStorage,
                Replicas:          1,
                MaxMsgsPerSubject: 1,
        })

here is the consumer:

          // Create main consumer with work queue semantics
          consumer, err := js.CreateOrUpdateConsumer(ctx, "TASKS", jetstream.ConsumerConfig{
                  Name:          "task-processor",
                  Durable:       "task-processor",
                  Description:   "process tasks exclusively",
                  FilterSubject: "tasks.*",
                  AckPolicy:     jetstream.AckExplicitPolicy,
                  AckWait:       30 * time.Second, // How long to wait for ack before redelivery
                  MaxAckPending: 1,                // Key: Only 1 unacked message per consumer
                  MaxDeliver:    3,                // Maximum delivery attempts before DLQ
                  DeliverPolicy: jetstream.DeliverAllPolicy,
                  ReplayPolicy:  jetstream.ReplayInstantPolicy,
                  BackOff: []time.Duration{ // Exponential backoff for retries
                          1 * time.Second,
                          5 * time.Second,
                          10 * time.Second,
                  },
          })

when I send one message to the stream, both workers receive it... so it's not exactly ONCE.. there is something in the semantic that I don't understand.

I have the following in my message processing handler:
fmt.Printf("\n=== Processing Task: %s (Delivery #%d) ===\n", task.ID, metadata.NumDelivered)

and while I run both concurrently, they BOTH receive the message:

Worker 1: === Processing Task: 8eb14624-4bb2-455e-9817-64108f1ea3a6 (Delivery #1) ===
Worker 2: === Processing Task: 8eb14624-4bb2-455e-9817-64108f1ea3a6 (Delivery #2) ===

I created a subject if we don't want to pollute this one:

#1901

I'm just trying to have a behavior where exactly ONE receive the message and NOT the other unless a redelivery is necessary (AckWait expiry), what I witness is 2 deliveries... whil

[eau-u4f]

ok got it, BackOff overrides AckWait consumer settings, my bad, works as intended now. :)

[jnmoyne]

Indeed, and it's mentioned in the docs (https://docs.nats.io/using-nats/developer/develop_jetstream/consumers#delivery-reliability)

You can control the timing of re-deliveries using either the single AckWait duration attribute of the consumer, or as a sequence of durations in the BackOff attribute (which overrides AckWait).
You can also control the timing of re-deliveries when messages are negatively acknowledged with Nak(), by passing a nakDelay() option (or using NakWithDelay()), otherwise the re-delivery attempt will happen right after the reception of the Nak by the server.

[eau-u4f]

thank you, a lot, for clarifying my mind.
I am now trying to see whether I should use WorkQueuePolicy but you cannot have a single consumer with multiple subjects where each consumer instances would only receive its subject (to avoid processing the subject in the worker and using redelivery to reach a worker interested into that specific subject).
All while keeping the "exactly once" delivery semantic, I guess LimitsPolicy might be the way to go an defining multiple durable consumers with different subject.

[jnmoyne]

You can create as many durable consumers as you want on a work queue stream (with as many subject filters as you want for each of those consumers), as long as 2 consumers do not have an overlapping subject filter.