How can I create a durable consumer?

[robertmircea]

I am trying to ConsumeAsync as a durable consumer by specifying DurableName in consumer options and I get the following exception when trying to create the consumer:

NATS.Client.JetStream.NatsJSApiException: consumer expected to be ephemeral but a durable name was set in request
   at NATS.Client.JetStream.Internal.NatsJSResponse`1.EnsureSuccess()
   at NATS.Client.JetStream.NatsJSContext.JSRequestResponseAsync[TRequest,TResponse](String subject, TRequest request, CancellationToken cancellationToken)
   at NATS.Client.JetStream.NatsJSContext.CreateConsumerAsync(String stream, ConsumerConfig config, CancellationToken cancellationToken)
   at Nats.NatsV2ConsumerProcessor`1.CreateProcessor(NatsConsumerMessageProcessorOptions`1 options, CancellationToken cancellationToken) in Nats/NatsV2ConsumerProcessor.cs:line 51

The code looks like this:

    private async Task CreateProcessor(NatsConsumerMessageProcessorOptions<T> options, CancellationToken cancellationToken)
    {
        var msgSerializer = new NatsMessagePackContextSerializer<T>();
        var consumerConfig = new ConsumerConfig
        {
            MaxDeliver = options.MaxDeliver,
            AckWait = options.AckWaitMs,
            AckPolicy = ConsumerConfigAckPolicy.Explicit,
            DurableName = options.NatsDurableName,
            DeliverSubject = options.NatsSubjectName,
            MaxAckPending = options.MaxAckPending,
            MaxBatch = options.BatchSize,
        };

        var consumer = await js.CreateConsumerAsync(options.JetStreamName, consumerConfig, cancellationToken);
        while (!cancellationToken.IsCancellationRequested)
        {
            try
            {
                await consumer.RefreshAsync(cancellationToken); // or try to recreate consumer

                await foreach (var msg in consumer.ConsumeAsync(serializer: msgSerializer).WithCancellation(cancellationToken))
                {
                    // Process message, let processor to ack/nack messages
                    await options.ProcessMsgFunc(msg, cancellationToken).ConfigureAwait(false);
                    await msg.AckAsync(cancellationToken: cancellationToken);
                }
            }
            catch (NatsJSProtocolException e)
            {
                //no need to Nack any messages here because exception are already handled at this point
                logger.Error(e, "Exception while processing NATS v2 message");
            }
            catch (NatsJSException e)
            {
                // log exception
                logger.Error(e, "Exception while processing NATS NatsJSException v2 message");
                await Task.Delay(1000, cancellationToken); // backoff
            }
        }
    }

What is the proper way to create a durable consumer?

[mtmk]

Off the top of my head I think server is expecting a name. You can supply it as an init property:

var consumerConfig = new ConsumerConfig
{
    MaxDeliver = options.MaxDeliver,
    AckWait = options.AckWaitMs,
    AckPolicy = ConsumerConfigAckPolicy.Explicit,
    Name = options.NatsDurableName, // <--- here
    DurableName = options.NatsDurableName,
    // DeliverSubject = options.NatsSubjectName, <-- remove
    MaxAckPending = options.MaxAckPending,
    MaxBatch = options.BatchSize,
};

Or you can use the constructor which takes care of the minimum defaults: name, durable_name and ack_policy:

var consumerConfig = new ConsumerConfig(options.NatsDurableName)
{
    MaxDeliver = options.MaxDeliver,
    AckWait = options.AckWaitMs,

    // Set by the .ctor
    // AckPolicy = ConsumerConfigAckPolicy.Explicit,
    // Name = options.NatsDurableName,
    // DurableName = options.NatsDurableName,

    // DeliverSubject = options.NatsSubjectName, <-- remove
    MaxAckPending = options.MaxAckPending,
    MaxBatch = options.BatchSize,
};

edit: DeliverSubject implies push consumer which is deprecated. consumer.ConsumeAsync() needs a pull consumer.

[robertmircea]

I've set the name in consumer config, but when trying to create the consumer, the following exception is thrown:

NATS.Client.JetStream.NatsJSApiException: consumer deliver subject forms a cycle
   at NATS.Client.JetStream.Internal.NatsJSResponse`1.EnsureSuccess()
   at NATS.Client.JetStream.NatsJSContext.JSRequestResponseAsync[TRequest,TResponse](String subject, TRequest request, CancellationToken cancellationToken)
   at NATS.Client.JetStream.NatsJSContext.CreateConsumerAsync(String stream, ConsumerConfig config, CancellationToken cancellationToken)
   at Nats.NatsV2ConsumerProcessor`1.CreateProcessor(NatsConsumerMessageProcessorOptions`1 options, CancellationToken cancellationToken) in Nats/NatsV2ConsumerProcessor.cs:line 52

[mtmk]

Ah didn't notice that actually. I think that's for push consumers (which are kind of depticated). Just remove DeliverSubject from the config.

edit: btw there is another property FilterSubjects if you want to only consume certain subjects from the stream if that's the intention here.

[robertmircea]

No more exception when creating the consumer, but I am not able to receive any message. The v1 configuration of the consumer which worked was:

                // Important:The consumer must have a deliver subject when made this way or it will be
                // understood to be a pull consumer by the server.
                var cc = ConsumerConfiguration.Builder()
                    .WithDurable(durableName) //The name of the Consumer, which the server will track, allowing resuming consumption where left off. By default, a consumer is ephemeral. To make the consumer durable, set the name.
                    .WithMaxAckPending(maxAckPending)
                    .WithAckWait(ackWaitMs)
                    .WithMaxDeliver(maxDeliver)
                    .WithReplayPolicy(ReplayPolicy.Instant)
                    .WithFilterSubject(subjectName)
                    .Build();
                jetStreamManagementContext.AddOrUpdateConsumer(streamName, cc);
                var psob = new PullSubscribeOptions.PullSubscribeOptionsSubscribeOptionsBuilder()
                    .WithConfiguration(cc);
                
                using var sub = jetStreamContext.PullSubscribe(subjectName, psob.Build());
[...]

This configuration was actually using WithFilterSubject.

Currently, the v2 consumer configuration is:

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

[mtmk]

I think you might be hitting a bug. When you turn on logging do you see warnings of 409 Exceeded MaxRequestBatch?

Consume batch size is 1000 by default. Assuming your options.BatchSize is less, can you try this see if it helps:

var opts = new NatsJSConsumeOpts { MaxMsgs = options.BatchSize };
await foreach (var msg in consumer.ConsumeAsync(opts: opts, serializer: msgSerializer).WithCancellation(cancellationToken))
{
    // ...
}

[robertmircea]

Great! It worked! Batch size is less than 1000.

[robertmircea]

Thank you! So, is it a bug? It requires a fix in the library?

[robertmircea]

It will also be nice to enhance documentation for the consumer to specify/exemplify how to create a durable one.

[mtmk]

Thank you! So, is it a bug? It requires a fix in the library?

I think it is. I think it should throw an exception at that point because it is not recoverable. I created an issue #236 feel free to chip in.

[robertmircea]

Something strange happens with the consumer code above: although I publish a single message in the stream, the consumer code is called multiple times immediately. The same message is delivered multiple times (NumDelivered increments) as if the previous delivery was not acked (I am acking manually in ProcessMsgFunc later on). The culprit might be the value of AckWait which is not set right. As it seems, the AckWait must be provided as nanoseconds, therefore the conversion in my code from millis.

        var consumerConfig = new ConsumerConfig(options.NatsDurableName)
        {
            DurableName = options.NatsDurableName,
            MaxDeliver = options.MaxDeliver,
            AckWait = options.AckWaitMs * 1_000_000,
            //DeliverSubject = options.NatsSubjectName,
            MaxAckPending = options.MaxAckPending,
            MaxBatch = options.BatchSize,
            FilterSubject = options.NatsSubjectName
        };

        var consumer = await js.CreateConsumerAsync(options.JetStreamName, consumerConfig, cancellationToken);
        while (!cancellationToken.IsCancellationRequested)
        {
            try
            {
                await consumer.RefreshAsync(cancellationToken); // or try to recreate consumer

                var opts = new NatsJSConsumeOpts { MaxMsgs = options.BatchSize };
                await foreach (var msg in consumer.ConsumeAsync(opts: opts, serializer: msgSerializer).WithCancellation(cancellationToken))
                {
                    // Process message, let processor to ack/nack messages
                    await options.ProcessMsgFunc(msg, cancellationToken).ConfigureAwait(false);
                    //await msg.AckAsync(cancellationToken: cancellationToken);
                }
            }

[mtmk]

there is an int overflow. Try 1_000_000L instead. I'm hoping we'd use TimeSpan for all time values #235 it's breaking change but it'd eliminate errors like these.

edit: don't know what happens on the server side with negative values of AckWait, so that might not be the issue!

[robertmircea]

Thanks for advice, that was it! Using Timespan everywhere for sure it would be much less error prone.