Getting queues working in JetStream

[scottf]

Here is an example of a Push based Queue

https://github.com/nats-io/nats.net/blob/master/src/Tests/IntegrationTests/TestJetStreamPushSyncQueue.cs

With JetStream queues, all subscribers with the same deliver group (aka queue name) name will form the queue group and only one member of the group will be selected to receive any given message asynchronously.

[scottf]

From @alan994

I tired something like that in my example project and I'm getting error Consumer already exists . I assume this error is because my ConsumerSyncA and ConsumerSyncB have same durable name natsconsumersync1 , but in your test this is the same.

ConsumerSyncA:
https://github.com/alan994/NatsJetStreamExample/blob/main/Consumer1.Web/ConsumerSyncA%20.cs#L33

ConsumerSync B:
https://github.com/alan994/NatsJetStreamExample/blob/main/Consumer1.Web/ConsumerSyncB.cs

Publisher:
https://github.com/alan994/NatsJetStreamExample/blob/main/NatsJs.Publisher/Program.cs

[scottf]

So your consumers are already made before the example starts. Are they configured with Deliver Group?

[alan994]

I just run my sample again. Before starting I deleted all consumers and streams in Nats using the Nats-CLI tool. When I start my web app that has registered two consumers with the same durable name I get an error '[SUB-90012] Consumer is already bound to a subscription.'. That makes sense to me, I'm trying to register two consumers that have the same name and Nats is telling me ok I already have one consumer with this name, pick another name. The part that doesn't make sense to me is how can I register two consumers to listen on the same subject and ensure only one of them gets the message (some kind of load balancing).

[scottf]

@alan994 So there is a bug in the subscribe code, but I'm also pretty sure that your consumer may not have the deliver group. I put together an example that is working now, that creates the consumer ahead of time, then uses that durable consumer in a queue fashion.

https://github.com/nats-io/nats.net/blob/master/src/Samples/JetStreamPushSubscribeQueueDurable/JetStreamPushSubscribeQueueDurable.cs

[alan994]

Ok, I see now. I misunderstood some parts of the documentation. If I want to have multiple instances of application consumers I need to create only one consumer in Nats, but I can have multiple subscriptions on this consumer. I Modified my example and it is working ok. I just thought I understand and try to validate my understanding with the example below and got the error 'consumer deliver subject has wildcards [10079]'.

For given configuration
WithSubjects("send.>") on my stream
WithDeliverSubject("send.*") on my consumer configuration
and subject send.sms in my subscriptions

I was expecting
The message that has the subject set to send.email.priority will be delivered to stream, but won't be sent to consumer or subscription.
The message that has the subject set to send.email will be delivered to stream and then to consumer, but it will not be sent to the subscription
The message that has the subject set to send.sms will be delivered to stream, consumer, and one subscription.

I said ok, then I tried with this configuration
WithSubjects("send.>") on my stream
WithDeliverSubject("something-else") on my consumer configuration
and subject send.sms in my subscriptions

For this configuration, I was expecting a message with the subject send.mail will be only delivered to stream, but it was delivered to my consumer and one subscription.

What's the point of the delivery subject when I can have anything in there? Also, I was expecting that my subscription will receive only messages with the subject send.sms but I got message with subject send.mail.

I committed this last example to git repo so it should be reproducable. I also updated nuget package to latest version.

[scottf]

Delivery Subject is the "address" that the server and client agree that the server will send messages to. When creating a push consumer ahead of time, this field is necessary. If the client is creating a push consumer for you during subscribe, it will just generate a delivery subject if you don't set one in the ConsumerConfiguration.

Pull consumers use a temporary address, so that's why pull consumers can't have a deliver subject when creating, in fact that's how we tell the difference between pre-existing pull and push consumers.

[scottf]

The reason you got messages from both send.sms and send.mail is likely because you did not set a FilterSubject when you created your consumer. A blank filter subject is the same as > which means all.

So I hear you thinking, "but I subscribed to send.sms when I made the subscription (aka the SubscribeSubject), so that's what I should have gotten. And this is actually the bug I referenced. I should have rejected the subscribe because the SubscribeSubject did not match the FilterSubject, because no FilterSubject is wider than / not equal to the SubscribeSubject.

We are currently discussing strategies to fix this and should have a fix by end of week.

[alan994]

Thanks for the explanation! Unfortunately, this will not be the only dummy question I will ask.
I think I understand now. When a new version is published, I will test my understanding 😄.

[scottf]

No worries!

[ccapuano-ssc]

Hello, I have a question related to this. We are porting from the legacy STAN implementation/libraries. I currently have a set of agents that scale on demand subscribing to a given queue group. The question I have is related to the MaxAckPending residing on the consumer. Ideally what I am looking for is the MaxAckPending to be a product of the subscription (or some equivalent paradigm) so delivery stops to a given subscription in a queue group at that time. If I set the MaxAckPending to 5, lets say, and have 2 agents subscribed, I would like to have 10 messages in flight (5 to each agent). I am looking at FlowControl now as a possible solution, but I cannot glean from the documentation if this will work as I intend.

Please find my relevant configuration below
Thanks,
Chris

var queueName = $"jobs-queue";
            // create the consumer ahead of time
            var cc = ConsumerConfiguration.Builder()
                .WithReplayPolicy(ReplayPolicy.Instant)
                .WithDurable("jobs-in") //Names the consumer
                .WithDeliverSubject($"jobs-queue-deliver") //??
                .WithFlowControl(1000)
                .WithDeliverGroup(queueName) //Assigns the queue group
                .WithAckPolicy(AckPolicy.Explicit)
                .WithMaxAckPending(5) //Defines how many in flights
                .WithDeliverPolicy(DeliverPolicy.New) // Defines what messages to send
                .WithAckWait(ACK_TIMEOUT) // Defines how long to allow before queuing redlivery to queue group
                .Build();

            var consumer = jsm.AddOrUpdateConsumer(STREAM_NAME, cc);

            //Bind the subscription to the existing consumer details
            var pso = PushSubscribeOptions.BindTo(consumer.Stream, consumer.ConsumerConfiguration.Durable);
            
            jsSub = jsContext.PushSubscribeAsync("jobs.*", queueName, this.MessageHandler, false, pso);

[ccapuano-ssc]

Another note is we are looking to stick with push subscriptions. Pull is not an option we are looking to leverage at this time.

[scottf]

Have you tried a durable consumer using a queue, with the MaxAckPending set how you want? Can you look at this sample to see if it has what you need? https://github.com/nats-io/nats.net/blob/master/src/Samples/JetStreamPushSubscribeQueueDurable/JetStreamPushSubscribeQueueDurable.cs

[ccapuano-ssc]

That's the example I started with. The issue I am trying to overcome is that the MaxAckPending is tied to the consumer, not the subscription. When I set the maxAck to 5, and ran 3 agents on the same durable queue, across the 3 there were never more than 5 messages being handled at once, where in that scenario I would like to see 15. What I am looking for is to make subsequent subscriptions increase the MaxAckPending, or really, to tie that MaxAckPending to a subscription so any given subscriber is never passed more messages than that limit.

So far I have tried

• Changing MaxAckPending on each subscription to increment the total (using AddOrUpdateConsumer, which does work). This kind of works in that NATs starts delivering more messages, but I don't think it guarantees that any given subscriber never handles more than a configured limit of messages. Also to do this effectively, I would need to alter the MaxAckPending on subscriptions and unsubs, and I have not yet found a mechanism to pull the current number of subscriptions on a given Consumer.
• Holding an interlocked int tracking the currently processing messages in the subscribers and Nak'ing any messages received that the current subscriber cannot handle. This seems to work in that the message is relayed to another agent, but I have concerns here that in times of heavy load, Nak'ing may create a high amount of cycles between subscribers and the server. Is this a bad idea?

I also attemted to turn on FlowControl and Heartbeats, but I have not yet received any messages with a status header. Some information I came across seemed to indicate that FlowControl with Jetstream queue groups is incompatible. Is that true? Am I barking up the wrong tree there?

Thanks again.

[ccapuano-ssc]

Actually just put together a quick test on point 2 above and confirmed my worry. Nak'ing sends the server into a feedback loop where the messages are continually redelivered until a client can handle them. I don't think that approach works.

[scottf]

Ok, I think your analysis about MaxAckPending is correct and your workaround does seem viable. What do you do when a consumer exits?

Regarding flow control and heartbeats, these are all handled internally for you, you will never see them.

Also I don't think this is a client issue as all clients will have the same exact problem because the queuing just pushes everything. You might find some control with rate limit, but that's a byte limit, not a message limit and again, across queues not sure how it will work in reality.
You could also lengthen (or shorten) the ack wait which will change the time something is redelivered if not ack'd.

You might want to propose a server feature. That repo is https://github.com/nats-io/nats-server . Sorry it just doesn't seem like we support exactly what you are trying to accomplish.

[ccapuano-ssc]

Thanks for the reply. I assume when you say "consumer", you mean "subscription", right? Not the Jetstream "Consumer". With the workaround, I would update the "Consumer" on new subscription upping the MaxAckPending by a configurable amount of messages that the new subscription could handle. On unsubscribe/exit, I would update the Consumer again with a decrement of the MaxAckPending by that same amount. Essentially making it elastic. I will still have to deal with messages hitting subscriptions that shouldn't be handling them, but its much closer to the desired effect. What I cannot find, though, is a way to determine via the API how many subscriptions a given Jetstream consumer has. Is there somewhere I can pull that from?

In respect to FlowControl, can you provide more info on that? The documentation I saw seemed to indicate that when this is on, I should expect to handle messages of a flow control type to indicate that my subscriber is in fact responsive. This would be another avenue I could explore if that is the case as I could then control via flow control when a given subscription is able to accept more messages. Is that not a correct assumption?

Will definitely propose a feature on this too as I believe the server/queue group has all the necessary knowledge. I would expect its the same variables it uses to load balance across the group.

[perestoronin]

1 consumer == 1 subject ?
It's possible 1 consumer with many subjects ?

[scottf]

Yes but only with wildcards. For instance if your subjects are foo and bar, you won't be able to subscribe to both with one consumer. But if your subjects are blah.foo and blah.bar, you could subscribe to blah.* or blah.> and get both subjects.