How to troubleshoot NatsJSApiNoResponseException when calling CreateOrUpdateConsumerAsync?

[robertmircea]

How can I troubleshoot the following exception which occurs when trying to create a consumer (CreateOrUpdateConsumerAsync)? Unfortunately, I see this exception only on production. On my local environment it just works:

NATS.Client.JetStream.NatsJSApiNoResponseException: No API response received from the server
   at NATS.Client.JetStream.NatsJSContext.JSRequestAsync[TRequest,TResponse](String subject, TRequest request, CancellationToken cancellationToken)
   at NATS.Client.JetStream.NatsJSContext.JSRequestAsync[TRequest,TResponse](String subject, TRequest request, CancellationToken cancellationToken)
   at NATS.Client.JetStream.NatsJSContext.JSRequestResponseAsync[TRequest,TResponse](String subject, TRequest request, CancellationToken cancellationToken)
   at NATS.Client.JetStream.NatsJSContext.CreateOrUpdateConsumerAsync(String stream, ConsumerConfig config, CancellationToken cancellationToken)

Code to create consumer:

        var consumerConfig = new ConsumerConfig(durableName)
        {
            DurableName = durableName,
            MaxDeliver = natsSettings.Value.MaxDeliveryAttempts,
            AckWait = natsSettings.Value.AckWait,
            MaxAckPending = natsSettings.Value.MaxInFlight,
            MaxBatch = natsSettings.Value.BatchSize,
            FilterSubject = subjectName
        };
        
       consumer = await js.CreateOrUpdateConsumerAsync(streamName, consumerConfig, cancellationToken);

Values used for stream, subject and durable name:

            {
                "StreamName": "x-y-auditevents",
                "Subject": "x-y.audit.gateway.events",
                "DurableName": "av-x-y-auditevents"
            }

I've captured with tcpdump the communication between server and client and this is what is captured:

CONNECT {"echo":true,"verbose":false,"pedantic":false,"tls_required":false,"user":"x","pass":"yy","name":"NATS .Net Client","lang":"C#","version":"1.0.0","protocol":1,"headers":true,"no_responders":true}
PING
PING
SUB _INBOX.JjOSXCKhF9eQpFhSt8zONt.* 1
PUB $JS.API.CONSUMER.CREATE.x-y-auditevents.av-x-y-auditevents.x-y.audit.gateway.events _INBOX.JjOSXCKhF9eQpFhSt8zONt.JjOSXCKhF9eQqHhSt8zONt 294
{"stream_name":"x-y-auditevents","config":{"deliver_policy":"all","durable_name":"av-x-y-auditevents","ack_policy":"explicit","ack_wait":60000000000,"max_deliver":5,"filter_subject":"x-y.audit.gateway.events","replay_policy":"instant","max_ack_pending":1000,"max_batch":1000}}
PONG
PING

[mtmk]

did you capture INFO server sends first? U;n guessing JetStream isn't enabled.

[robertmircea]

That was it! Thanks! Now I am getting a different error:

[2024-03-04 02:07:51.900 WRN] Can't find subscription for x-y.audit.gateway.events/2 [NATS.Client.Core.Internal.SubscriptionManager]
[2024-03-04 02:07:51.900 WRN] Can't find subscription for x-y.audit.gateway.events/2 [NATS.Client.Core.Internal.SubscriptionManager]
[2024-03-04 02:07:51.900 WRN] Can't find subscription for x-y.audit.gateway.events/2 [NATS.Client.Core.Internal.SubscriptionManager]
[2024-03-04 02:07:51.900 WRN] Can't find subscription for x-y.audit.gateway.events/2 [NATS.Client.Core.Internal.SubscriptionManager]

[mtmk]

maybe I'm missing something but I meant using it with the consumer. ACK/NAK should still work as expected e.g.:

foreach (var msg in consumer.ConsumeAsync<NatsMemoryOwner<byte>>())
{
  // same as above
  using var memoryOwner = msg.Data;
  try
  {
    // custom deserialization
    await msg.AckAsync();
  }
  catch (Exception e)
  {
    // handle maybe NAK
    await msg.NakAsync();
  }
}

but as I said maybe I'm missing something, feel free to correct me.

[robertmircea]

You are right—I was using the NATS part, not Jetstream with NatsMemoryOwner. Just for my clarification: Are there any future plans to handle deserialization errors more elegantly? Because working with the memory owner and manually deserializing just to catch and ignore the deserialization error unnecessarily complicates the code.

[mtmk]

yes, kind of, have a look at #407 and feel free to chip in.

[robertmircea]

I am still not sure how to prevent these log messages:
[2024-03-04 02:07:51.900 WRN] Can't find subscription for x-y.audit.gateway.events/2 [NATS.Client.Core.Internal.SubscriptionManager]

You suggested to keep a reference to the subscription/consumer. Can you suggest please an improvement to the code I posted above in order to avoid NATS client to log this warning? Thanks a lot for your support.

[mtmk]

the code above should be good. sorry to be pain but could you post a complete Program.cs for me to debug?

[robertmircea]

What happens if one the messages fail to deserialize using the deserializer assigned at startup to SerializerRegistry? Are these messages going to be skipped? Do I need to handle these somehow? Is it possible to log the event?

[mtmk]

Serialization errors would only be thrown when the subscription is disposed which might not be ideal. Unfortunately they're not logged at the moment.

• we have a PR related to that Message error processing #407

If you're expecting serialization issues I recommend using NatsMemoryOwner as your serialization type then deserialize using NatsSerializer yourself, without incurring extra allocations.

[mtmk]

@robertmircea, we just release a fix for handling serialization exceptions. It might break some code if it was relaying on serialization exceptions (to be effectively swallowed in some cases). Have a look at the release notes for an example: https://github.com/nats-io/nats.net.v2/releases/tag/v2.1.4 as always feel free to drop a line and let us know if you see any isses with it or have comments.

[robertmircea]

thanks! Did not have the chance to try it yet but as a preview: In case of Error, I still need to ack/nack the message as expected, right?

Do I get anywhere the deserialization error even if not throwing? Just to be aware that it was an issue with the payload, perhaps to log it...

[mtmk]

yes, you can still ack/nack/terminate messages as usual, and yes the exception would be NatsDeserializeException and you can access the actual serializer exception (it'd be the InnerException) and even the payload as byte array:

await foreach (var msg in consumer.ConsumeAsync<Order>())
{
    Console.WriteLine($"Received: {msg.Subject}");

    if (msg.Error is { } error)
    {
        // Handle serialization errors

        if (error is NatsDeserializeException
            {
                RawData: { } rawData,
                InnerException.Message: { } errorMessage,
            })
        {
            Console.WriteLine($"Error message: {errorMessage}");

            var payload = Encoding.ASCII.GetString(rawData); // or hex dump, etc.
            Console.WriteLine($"Payload: {payload}");
        }
        else
        {
            Console.WriteLine($"Unknown Error: {error}");
        }
        
        // We can't recover from this.
        // Terminate, so no consumer will receive it again.
        await msg.AckTerminateAsync();

        continue;
    }

    Console.WriteLine($"Processing order: {msg.Data}");
    await msg.AckAsync();
}