JetStream server error: 409 Exceeded MaxRequestBatch of 60

[robertmircea]

What setting can cause this exception for a NATS consumer?

NATS.Client.JetStream.NatsJSProtocolException: JetStream server error: 409 Exceeded MaxRequestBatch of 60
   at System.Threading.Channels.AsyncOperation`1.GetResult(Int16 token)
   at NATS.Client.JetStream.NatsJSConsumer.ConsumeAsync[T](INatsDeserialize`1 serializer, NatsJSConsumeOpts opts, CancellationToken cancellationToken)+MoveNext()
   at NATS.Client.JetStream.NatsJSConsumer.ConsumeAsync[T](INatsDeserialize`1 serializer, NatsJSConsumeOpts opts, CancellationToken cancellationToken)+MoveNext()
   at NATS.Client.JetStream.NatsJSConsumer.ConsumeAsync[T](INatsDeserialize`1 serializer, NatsJSConsumeOpts opts, CancellationToken cancellationToken)+System.Threading.Tasks.Sources.IValueTaskSource<System.Boolean>.GetResult()

Consumer code:

        var consumerConfig = new ConsumerConfig(opt.NatsDurableName)
        {
            DurableName = opt.NatsDurableName,
            MaxDeliver = opt.MaxDeliver,
            AckWait = opt.AckWait,
            MaxAckPending = opt.MaxAckPending,
            MaxBatch = opt.BatchSize,
            FilterSubject = opt.NatsSubjectName
        };

        var consumer = await js.CreateOrUpdateConsumerAsync(opt.JetStreamName, consumerConfig, cancellationToken);

        while (!cancellationToken.IsCancellationRequested)
        {
            try
            {
                await consumer.RefreshAsync(cancellationToken); // or try to recreate consumer

                var opts = new NatsJSConsumeOpts { MaxMsgs = opt.BatchSize };
                await foreach (var msg in consumer.ConsumeAsync<T>(opts: opts, cancellationToken: cancellationToken))
                {
                    if (await IsNatsJetstreamMessageError(cancellationToken, msg)) continue;

                    await opt.ProcessMsgFunc(msg, cancellationToken).ConfigureAwait(false);
                }
            }
            catch (NatsJSProtocolException e)
            {
                logger.LogError(e, "Exception while processing NATS v2 message");
            }
            catch (NatsJSException e)
            {
                // log exception
                logger.LogError(e, "Exception while processing NATS v2 message");
                await Task.Delay(1000, cancellationToken); // backoff
            }
            catch (Exception e)
            {
                logger.LogError(e, "Exception while processing NATS v2 message");
            }
        }

[mtmk]

there maybe a stream setting for max-batch? (is there?🤔) or it might be a bug, not sure. Could you give me a Program.cs, with stream setup included so I can repro on my end?

[robertmircea]

I think I found the cause but it would be useful if you could take a look. It seems that the exception occurs when NatsJSConsumeOpts.MaxMsg is greater than consumer's config MaxBatch value. If I don't specify NatsJSConsumeOpts.MaxMsg at all, then it is initialized with default value 1000 which in my case is higher than MaxBatch=120. Is this an issue or intended behaviour?

Short app to reproduce attached:
Nats409Case.zip

[mtmk]

If I don't specify NatsJSConsumeOpts.MaxMsg at all, then it is initialized with default value 1000 which in my case is higher than MaxBatch=120. Is this an issue or intended behaviour?

I think you have a good point. We should either not provide a default (which would be a breaking change) or adjust the max to max-batch.

[mtmk]

I think your example demonstrates the issue perfectly. Looking at the exception handling I would do something like this:

async Task ConsumerLoop(string name, CancellationToken cancellationToken)
{
    var backOffTimeouts = new[]
    {
        250, 500, 1_000, 3_000, 10_000, 30_000, 60_000, 120_000, 300_000,
    };
    var attempt = 0;
    var jitter = Random.Shared;

    while (!cancellationToken.IsCancellationRequested)
    {
        try
        {
            await consumer.RefreshAsync(cancellationToken); // or try to recreate consumer

            await foreach (var msg in consumer.ConsumeAsync<string>(opts: consOpts, cancellationToken: cancellationToken))
            {
                attempt = 0;
                await msg.AckAsync(cancellationToken: cancellationToken);
                Console.WriteLine($"{name}: Msg = {msg.Data}");
            }
        }
        catch (NatsJSException e)
        {
            // log exception
            Console.WriteLine("Exception while processing NATS v2 message (NatsJSException)" + e);

            // backoff
            var delay = backOffTimeouts[Math.Min(++attempt, backOffTimeouts.Length - 1)];

            // Introduce jitter by randomly adjusting the delay +/- 15%
            var jitterAmount = (int)(delay * 0.15);
            delay += jitter.Next(-jitterAmount, jitterAmount);

            Console.WriteLine($"Waiting {delay}ms before next attempt...");
            await Task.Delay(delay, cancellationToken);
        }
    }
}

Backoff timer series, and retrying for all exceptions are my personal views and might be a little contentious and might also be different for environment and setup. Also this doesn't apply to this particular case since it's obviously a configuration issue in the application and can never recover. But the idea is that Consume() method internally already recovers from all the errors we know to be transient, and any other exceptions are going to be consumer issues which are potentially human errors e.g. consumer is deleted and someone needs to run a nats consumer create manually, hence we backoff up to five minutes and keep retrying.

I hope it makes sense. Let me know your thoughts.

[robertmircea]

This type of example should definitely be included in the official NATS client documentation. It would greatly benefit future users. When I first started with the NATS client, I would have appreciated seeing examples that were as close to production-ready as possible, demonstrating how to implement common use cases like this one on message consumption. Based on my experience, understanding the various consumer parameters and how to develop a resilient system requires considerable time and adjustment. For instance, my scenario involves multiple parallel pull consumers in durable mode, distributing tasks from a work-queue stream. Each consumer interacts with a rate-limited REST API, which has a much lower throughput than the NATS consumer. Some of the challenges I faced included:

• Determining the appropriate type of consumer for this use case (e.g., push vs. pull, durable).
• Finding the optimal MaxBatch size and MaxAckPending to achieve the right balance of message throughput for this throttled API send, ensuring efficient workload distribution across multiple application instances.
• Avoiding the slow consumer problem on NATS server.
• In another scenario, retrieving messages in batches from NATS Jetstream, transforming them into a different format, and then performing batch saves to a database or Elasticsearch. Performing batch inserts into a database or Elasticsearch is significantly faster when sending inserts in batches. It's also important to handle errors ack/nak during batch saves effectively.