Slow consumer error when fetching from work-queue stream

[heikkilamarko]

Our environment:

NATS.Client  : 1.0.4
NATS Server  : 2.9.17
.NET version : .NET 7.0
OS           : Windows Server 2019 Standard

We have configured a JetStream pull subscriber for a work-queue stream as follows:

var js = SubConnection.CreateJetStreamContext();
var options = PullSubscribeOptions.Builder().WithDurable("durable_name").Build();
using var sub = js.PullSubscribe("subject_name", options);

...

var msgs = sub.Fetch(1, 1000);

...

We have set the AsyncErrorEventHandler as follows:

options.AsyncErrorEventHandler = (s, e) => _logger.LogError($"nats: async error: error='{e.Error}' sid='{e.Subscription.Sid}' subject='{e.Subscription.Subject}' conn='{e.Conn.Opts.Name}'");

The setup ran without problems for almost two weeks and then suddenly stopped working. In our logs we see the slow consumer error logged by our AsyncErrorEventHandler:

[ERR] nats: async error: error='Slow Consumer' sid='2' subject='_INBOX.8iAznkRtJqb8oomkc2MUut' conn='connection_name'

The connection is used just for fetching messages from the work-queue one by one:

var msgs = sub.Fetch(1, 1000);

The message rate is about 10 messages in a minute or less, so we shouldn't have the slow consumer problem. The are no slow consumers detected by the NATS Server, only the client SDK.

Any idea what migth cause the slow consumer error?

[scottf]

Moved to discussion

[scottf]

So I'll need a lot more context. But on the client side, it means that a messages has come in that crosses the connection pending limits of max messages or max bytes, meaning there are unprocessed messages sitting in the queue.

What are your connection options. How often are you doing the fetch?

[heikkilamarko]

Here is the method we use to create the NATS connection:

private IConnection CreateConnection(string name)
{
  var options = ConnectionFactory.GetDefaultOptions();
  options.Url = _options.Url;
  options.User = _options.User;
  options.Password = _options.Password;
  options.Name = name;
  options.Timeout = 10_000;
  options.MaxReconnect = NATS.Client.Options.ReconnectForever;
  options.ClosedEventHandler = (s, e) => _logger.LogInformation(e.Error, $"nats: closed: conn='{e.Conn.Opts.Name}'");
  options.DisconnectedEventHandler = (s, e) => _logger.LogInformation(e.Error, $"nats: disconnected: conn='{e.Conn.Opts.Name}'");
  options.ReconnectedEventHandler = (s, e) => _logger.LogInformation(e.Error, $"nats: reconnected: conn='{e.Conn.Opts.Name}'");
  options.AsyncErrorEventHandler = (s, e) => _logger.LogError($"nats: async error: error='{e.Error}' sid='{e.Subscription.Sid}' subject='{e.Subscription.Subject}' conn='{e.Conn.Opts.Name}'");

  return new ConnectionFactory().CreateConnection(options);
}

Here is the message handling logic:

public async Task ListenAsync(CancellationToken stoppingToken)
{
  try
  {
    _logger.LogInformation("starting nats subscription");

    var js = SubConnection.CreateJetStreamContext();

    var options = PullSubscribeOptions.Builder().WithDurable(_options.SubDurable).Build();

    using var sub = js.PullSubscribe(_options.SubSubject, options);

    while (true)
    {
      if (stoppingToken.IsCancellationRequested)
      {
        _logger.LogInformation("cancellation requested for nats subscription");
        break;
      }

      try
      {
        var msgs = sub.Fetch(1, 1000);
        if (!msgs.Any()) continue;
        var msg = msgs[0];

        _logger.LogInformation("message received ({subject})", msg.Subject);

        // Handle the message...       

        _logger.LogInformation("message handled ({subject})", msg.Subject);

        msg.Ack();
      }
      catch (Exception err)
      {
        _logger.LogError(err, "message handling failed");
      }
    }

    _logger.LogInformation("stopping nats subscription");
  }
  catch (Exception err)
  {
    _logger.LogError(err, "nats subscription failed");
    throw;
  }
}

[heikkilamarko]

Stream config:

{
  "name": "myapp",
  "subjects": ["myapp.>"],
  "retention": "workqueue",
  "max_consumers": -1,
  "max_msgs_per_subject": -1,
  "max_msgs": -1,
  "max_bytes": -1,
  "max_age": 0,
  "max_msg_size": -1,
  "storage": "file",
  "discard": "old",
  "num_replicas": 1,
  "duplicate_window": 120000000000,
  "sealed": false,
  "deny_delete": false,
  "deny_purge": false,
  "allow_rollup_hdrs": false,
  "allow_direct": false,
  "mirror_direct": false
}

Consumer config:

{
  "ack_policy": "explicit",
  "ack_wait": 30000000000,
  "deliver_policy": "all",
  "durable_name": "myapp",
  "max_ack_pending": 1000,
  "max_deliver": -1,
  "max_waiting": 512,
  "replay_policy": "instant",
  "num_replicas": 0
}

[scottf]

So first, I'm not seeing anything obvious in the code. Is this happening repeatedly? let's keep an eye on this and see if we can figure out how to repeat it.

Second, I would offer a note.

1. Since you are getting 1 message at a time, just use PullExpiresIn instead of fetch. Less [internal] overhead. This means you have to call NextMessage yourself.
2. I would also change your timeout larger, closer to what your time expectation is. Less wait looping, less traffic to the server, plus like fetch, NextMessage returns as soon as it gets that 1 message.
3. If you change to that, you want to differentiate NATSTimeoutException from other exceptions.

try
{
    sub.PullExpiresIn(1, 5000); // the server gives 5 seconds to allow for a message to show up
    
    // Add a little extra to the timeout to allow for traffic back and forth to the server, to make sure
    // we don't timeout while the message is coming over the wire.
    Msg msg = sub.NextMessage(5050); 

    _logger.LogInformation("message received ({subject})", msg.Subject);

    // Handle the message...       

    _logger.LogInformation("message handled ({subject})", msg.Subject);

    msg.Ack();
}
catch (NATSTimeoutException)
{
    // normal behavior when NextMessage times out, probably don't even need to log
    _logger.LogInformation("No messages available({subject})", msg.Subject);
}
catch (Exception err)
{
    _logger.LogError(err, "message handling failed");
}

[heikkilamarko]

We are now using the new simplified API. We catch the NATSSlowConsumerException and exit the process with exit code 1, allowing the orchestrator (HashiCorp Nomad in our case) to restart the process. We haven't encountered the exception thus far.

public async Task ListenAsync(CancellationToken stoppingToken)
{
  try
  {
    _logger.LogInformation("starting nats subscription");

    var consumer = SubConnection.CreateConsumerContext(_options.SubStream, _options.SubConsumer);

    while (true)
    {
      if (stoppingToken.IsCancellationRequested)
      {
        _logger.LogInformation("cancellation requested for nats subscription");
        break;
      }

      try
      {
        var msg = consumer.Next(2000);
        if (msg is null) continue;

        _logger.LogInformation("message received ({subject})", msg.Subject);

        // Handle the message...       

        _logger.LogInformation("message handled ({subject})", msg.Subject);

        msg.Ack();
      }
      catch (NATSSlowConsumerException)
      {
        throw;
      }
      catch (Exception err)
      {
        _logger.LogError(err, "message handling failed");
      }
    }

    _logger.LogInformation("stopping nats subscription");
  }
  catch (Exception err)
  {
    _logger.LogError(err, "nats subscription failed");
    Environment.ExitCode = 1;
    throw;
  }
}

[scottf]

@heikkilamarko Thank you for the update. For simplification, you will never get slow consumer as it uses pull consumer underneath instead of push.
The simplification beta is closing soon, any feedback would be appreciated. You can find me directly on slack if you like.