Need some clarification on .net connection object in c#, slow consumers & Flow Control

[DarkPhantom-Dev]

Question1)
As per https://github.com/nats-io/nats.net, its using NatsClient instance to communicate with broker & also publish subscribe.

Ive been using NatsConnection object for opening connections & also for publish subscribe & even creating jetstream contexts.

Does using NatsClient have any advantages over NatsConnection ?

Question 2)
SubQn1) Are Slow Consumers a CoreNATS subscriber only concept or does it apply to jetstream consumers as well?

I tried this example https://docs.nats.io/using-nats/developer/connecting/events/slow#detect-a-slow-consumer-and-check-for-dropped-messages,

SubQn2) In my code for subscription, i do not provide any NatsSubChannelOpts, & i also use queue groups to make sure message is distributed only to one subscriber. till now ive not observed slow consumer message but do i need to take any precautions?

 await foreach (var msg in _natsClient.SubscribeAsync(topicSubscription,ConsumerGroup,serializer:natsMesgSerializer))
                        {
                           //ProcessingLogic
                        }

SubQn3) Lets assume slow consumer is detected, server canceling the subscription is quite scary, on client if we have the MessageDroppedCallback, what is recommended to ensure server does not cancel subscription? As per nats-io/nats.go#888 using MaxAckPending helps, wanted some more clarity on this.

SubQn4) If a subscription of CoreNATS is consuming but no messages are flowing yet, will server treat this as inactive & close it?

SubQn5 )Also As per https://stackoverflow.com/questions/70550060/performance-of-nats-jetstream to avoid overloading broker with publish calls, he states

"You have two options to flow-control your JetStream async publications:

specify a max number of in-flight asynchronous publication requests as an option when obtaining your JetStream context: i.e. js = nc.JetStream(nats.PublishAsyncMaxPending(100))"

How do i do this in c# client?

NatsConnection natsConnection = _natsBrokerConnection.GetConnection()!;
INatsJSContext _natsJSContext = _natsConnection.CreateJetStreamContext();
await _natsJSContext.PublishAsync(topicName,natsMessage,serializer:natsMesgSerializer);

Question 3) As per https://natsbyexample.com/examples/jetstream/pull-consumer/csharp/ & https://nats-io.github.io/nats.net/documentation/jetstream/consume.html

ConsumeAsync is the most performant of the consume methods due to real time streaming like behaviour. So does this mean FetchAsync can be ignored & we can assume to improve throughputs of our system its better to use ConsumeAsync?

Also the above documents show lot of flowcontrol techniques for FetchAsync, how to handle flow control in consumeAsync?

[DarkPhantom-Dev]

@mtmk can you pls help answer this

[mtmk]

no worries, see below. a lot of questions 😅

[mtmk]

Question1) As per https://github.com/nats-io/nats.net, its using NatsClient instance to communicate with broker & also publish subscribe.

Ive been using NatsConnection object for opening connections & also for publish subscribe & even creating jetstream contexts.

NatsClient is a simplified thin wrapper for ease of use. I suggest to keep using NatsConnection if you already are using it.

Does using NatsClient have any advantages over NatsConnection ?

Only advantage is it comes with better defaults for serialization and message channel for ease of use.

Question 2) SubQn1) Are Slow Consumers a CoreNATS subscriber only concept or does it apply to jetstream consumers as well?

It's a NATS Core concept.

I tried this example https://docs.nats.io/using-nats/developer/connecting/events/slow#detect-a-slow-consumer-and-check-for-dropped-messages,

SubQn2) In my code for subscription, i do not provide any NatsSubChannelOpts, & i also use queue groups to make sure message is distributed only to one subscriber. till now ive not observed slow consumer message but do i need to take any precautions?

 await foreach (var msg in _natsClient.SubscribeAsync(topicSubscription,ConsumerGroup,serializer:natsMesgSerializer))
                        {
                           //ProcessingLogic
                        }

If you are dropping messages when the subscription channel is full (NatsOpts setting) then you should also subscribe to the message dropped callback event.

SubQn3) Lets assume slow consumer is detected, server canceling the subscription is quite scary, on client if we have the MessageDroppedCallback, what is recommended to ensure server does not cancel subscription? As per nats-io/nats.go#888 using MaxAckPending helps, wanted some more clarity on this.

If you see messages being dropped you should slowdown or stop other consumers and subscribers for some time to give the system some space.

SubQn4) If a subscription of CoreNATS is consuming but no messages are flowing yet, will server treat this as inactive & close it?

No, connection would be kept alive with PINGs and the consumer also have heartbeats.

SubQn5 )Also As per https://stackoverflow.com/questions/70550060/performance-of-nats-jetstream to avoid overloading broker with publish calls, he states

"You have two options to flow-control your JetStream async publications:

specify a max number of in-flight asynchronous publication requests as an option when obtaining your JetStream context: i.e. js = nc.JetStream(nats.PublishAsyncMaxPending(100))"

How do i do this in c# client?

You can use the concurrent publish method on JetStream context.

NatsConnection natsConnection = _natsBrokerConnection.GetConnection()!;
INatsJSContext _natsJSContext = _natsConnection.CreateJetStreamContext();
await _natsJSContext.PublishAsync(topicName,natsMessage,serializer:natsMesgSerializer);

Question 3) As per https://natsbyexample.com/examples/jetstream/pull-consumer/csharp/ & https://nats-io.github.io/nats.net/documentation/jetstream/consume.html

ConsumeAsync is the most performant of the consume methods due to real time streaming like behaviour. So does this mean FetchAsync can be ignored & we can assume to improve throughputs of our system its better to use ConsumeAsync?

You should use ConsumeAsync in general which keeps the messages flowing with minimal overhead. FetchAsync is more for potentially for use cases like application wakes up and needs to consume some data and goes back to sleep for example.

Also the above documents show lot of flowcontrol techniques for FetchAsync, how to handle flow control in consumeAsync?

You can use the options to adjust thresholds and number of messages in batches for example.

[DarkPhantom-Dev]

@mtmk thank you very much for your response, much appreciated. I am unable to find any example of using PublishConcurrentAsync for flow control, can you pls help with this?

[mtmk]

no worries, I hope it helps. Let me dig it out. I think there is one in another discussion.

edit: #523 (comment)

var batch = 100;
var futures = new NatsJSPublishConcurrentFuture[batch];

for (var i = 0; i < batch; i++)
{
    futures[i] = await _js.PublishConcurrentAsync("foo", i);
}

for (var i = 0; i < batch; i++)
{
    await using var future = futures[i];
    var ack = await future.GetResponseAsync();
    ack.EnsureSuccess();
}

[DarkPhantom-Dev]

@mtmk this batching technique is quite difficult to implement & batch without slowing down operations, i want it to be like consumeAsync in await foreach.

I am doing same thing to produce messages like this. This is straightforward with minimum code. Also adding batching here might slow down operations in my usecase . I am trying to avoid batching.

NatsConnection natsConnection = _natsBrokerConnection.GetConnection()!;
INatsJSContext _natsJSContext = _natsConnection.CreateJetStreamContext();
await _natsJSContext.PublishAsync(topicName,natsMessage,serializer:natsMesgSerializer);

As i ask in this question:
SubQn5 )Also As per https://stackoverflow.com/questions/70550060/performance-of-nats-jetstream to avoid overloading broker with publish calls, he states

"You have two options to flow-control your JetStream async publications:

specify a max number of in-flight asynchronous publication requests as an option when obtaining your JetStream context: i.e. js = nc.JetStream(nats.PublishAsyncMaxPending(100))"

He is using PublishAsyncMaxPending(100) which is giving flow control to publishers, similar to MaxAckPending for jetstream consumer. I need some mechanism like this without batching. How do i do it?

[mtmk]

PublishAsyncMaxPending option is go specific. in c# we don't have that option since we can await in the language. if you want to await foreach like pattern you can use a bounded channel instead of an array and consume the future messages in a ChannelReader.ReadAllAsync() loop for example.

[DarkPhantom-Dev]

@mtmk i have added message drop handler on the connection to detect slow consumers, any action should we take after detecting slow consumer so that server does not close it?

[mtmk]

you need to somehow signal your application logic to slow down and take it easy whatever that might mean for your use case. e.g. you can create some kind of backpressure 'MODE' or 'STATE'

[DarkPhantom-Dev]

@mtmk @Jarema if slow consumers is a feature of core nats only, i am still observing slow consumers in my k8s server & server is disconnecting the consumer completely where all consumers are jetstream consumers only.

qn1) Slow Consumer Detected: WriteDeadline of 10s exceeded with 2 chunks of 939 total bytes.
[7] 2025/01/31 06:31:21.546834 [INF] 10.35.0.1:34586 - cid:454 - Slow Consumer Detected: WriteDeadline of 10s exceeded with 2 chunks of 941 total bytes.
[7] 2025/01/31 06:31:26.726137 [INF] 10.40.0.2:42418 - cid:455 - Slow Consumer Detected: WriteDeadline of 10s exceeded with 1522 chunks of 2142400 total bytes.
[7] 2025/01/31 06:31:53.523414 [INF] 10.42.0.2:47884 - cid:457 - Slow Consumer Detected: WriteDeadline of 10s exceeded with 800 chunks of 1180337 total bytes.
[7] 2025/01/31 06:33:43.014310 [INF] 10.42.0.2:37182 - cid:459 - Slow Consumer Detected: WriteDeadline of 10s exceeded with 1843 chunks of 2579094 total bytes.
[7] 2025/01/31 06:33:47.939854 [INF] 10.35.0.1:57148 - cid:458 - Slow Consumer Detected: WriteDeadline of 10s exceeded with 2784 chunks of 3955958 total bytes.
[7] 2025/01/31 06:34:08.324415 [INF] 10.40.0.2:38504 - cid:461 - Slow Consumer Detected: WriteDeadline of 10s exceeded with 2 chunks of 940 total bytes.
[7] 2025/01/31 06:34:13.929483 [INF] 10.35.0.1:57446 - cid:460 - Slow Consumer Detected: WriteDeadline of 10s exceeded with 1693 chunks of 2455704 total bytes.
[7] 2025/01/31 06:34:37.099977 [INF] 10.35.0.1:41854 - cid:462 - Slow Consumer Detected: WriteDeadline of 10s exceeded with 1038 chunks of 1551302 total bytes.
[7] 2025/01/31 06:34:43.071974 [INF] 10.42.0.2:50754 - cid:463 - Slow Consumer Detected: WriteDeadline of 10s exceeded with 1157 chunks of 1671439 total bytes

Now i understand that write deadline is configured to 10sec here, am i correct to assume because write deadline got exceeded the server closed this connection & thats why even though i am using only jetstream consumers i am facing this issue?

Question2) In my code i follow a single broker connection per application pattern, so lets assume one application has 20 consumers on same connection & out of this only 1 consumer is detected to be slow, it does not make any sense to close the entire connection itself right? Why is this happening? I can see multiple ConnectionDropped & Reconnection handlers being hit in client?

Requesting you to pls help me with this....

[mtmk]

JetStream is built on to of Core NATS so it would be affected. You can detect this in client application:

        await using var nats = new NatsConnection(new NatsOpts
        {
            // these are the defaults for NatsConnection
            SubPendingChannelCapacity = 1024,
            SubPendingChannelFullMode = BoundedChannelFullMode.DropNewest,
        });

        nats.MessageDropped += async (_, e) =>
        {
            // Log error, notify operations...
            // e.Pending is the number of messages in the channel.
        };

more info on slow consumers: https://docs.nats.io/running-a-nats-service/nats_admin/slow_consumers