Facing issue with c# client in 3 node partitioned k8s cluster

[romitkmehta]

We have a 3 node nats io jetstream set up in k8s with the below configuration

we have 3 partitoned stream and this is one of the streams below

Nats Node config: 2vCPU 4GB RAM

Note we are not using helm for deployment but rather using helm template files & deploying by ourseleves with GOMEMLIMIT of 3GB (FIle is added at end of question)

Also pls note: I have verified logs of all 3 broker nodes and it does not have any errors or exceptions from nats broker side.

nats stream info :

Information for Stream LRPS created 2025-03-06 18:27:03

              Subjects: LRPS.Test.TestTopic1.*, LRPS.Test.TestTopic2.*, LRPS.Test.TestTopic3.*, LRPS.Test.TestTopic4.*, LRPS.Test.TestTopic5.*, LRPS.Test.TestTopic6.*, LRPS.Test.TestTopic6.*, LRPS.Test.TestTopic7.*, LRPS.Test.TestTopic8.*, LRPS.Test.TestTopic9.*, LRPS.Test.TestTopic10.*, LRPS.Test.TestTopic11.*, LRPS.Test.TestTopic12.*, LRPS.Test.TestTopic13.*, LRPS.Test.TestTopic14.*, LRPS.Test.TestTopic15.*, LRPS.Test.TestTopic16.*, LRPS.Test.TestTopic17.*, LRPS.Test.TestTopic18.*, LRPS.Test.TestTopic19.*, LRPS.Test.TestTopic20.*, LRPS.Test.TestTopic21.*, LRPS.Test.TestTopic22.*, LRPS.Test.TestTopic23.*, LRPS.Test.TestTopic24.*, LRPS.Test.TestTopic25.*, LRPS.Test.TestTopic26.*
              Replicas: 3
               Storage: File

Options:

     Subject Transform: LRPS.*.*.* to LRPS.{{wildcard(1)}}.{{wildcard(2)}}.{{wildcard(3)}}.{{partition(2,1,2,3)}}
             Retention: Limits
       Acknowledgments: true
        Discard Policy: Old
      Duplicate Window: 2m0s
     Allows Msg Delete: true
          Allows Purge: true
        Allows Rollups: false

Limits:

      Maximum Messages: unlimited
   Maximum Per Subject: unlimited
         Maximum Bytes: 4.0 GiB
           Maximum Age: 7d0h0m0s
  Maximum Message Size: unlimited
     Maximum Consumers: unlimited

Cluster Information:

                  Name: nats
                Leader: nats-1
               Replica: nats-0, current, seen 2ms ago
               Replica: nats-2, current, seen 2ms ago

State:

              Messages: 230,682
                 Bytes: 323 MiB
        First Sequence: 1 @ 2025-03-06 18:27:05
         Last Sequence: 230,682 @ 2025-03-07 16:01:57
      Active Consumers: 52
    Number of Subjects: 230,482

This is one of the consumer configurations for LRPS stream:

Information for Consumer LRPS > TestTopic1-0 created 2025-03-06T18:27:05+05:30

Configuration:

                    Name: TestTopic1-0
               Pull Mode: true
          Filter Subject: LRPS.Test.TestTopic1.*.0
          Deliver Policy: New
              Ack Policy: Explicit
                Ack Wait: 1h0m0s
           Replay Policy: Instant
         Max Ack Pending: 1,000
       Max Waiting Pulls: 512

Metadata:

              ConsumerID: f5b2c005-b5da-446c-919e-3009fecb182a
    LastMsgDeliveredTime: 2025-03-07 10:24:30

Cluster Information:

                    Name: nats
                  Leader: nats-1
                 Replica: nats-0, current, seen 92ms ago
                 Replica: nats-2, current, seen 90ms ago

State:

  Last Delivered Message: Consumer sequence: 2,197 Stream sequence: 243,120 Last delivery: 1m48s ago
    Acknowledgment Floor: Consumer sequence: 2,197 Stream sequence: 240,926 Last Ack: 1m48s ago
        Outstanding Acks: 0 out of maximum 1,000
    Redelivered Messages: 0
    Unprocessed Messages: 0
           Waiting Pulls: 1 of maximum 512
		   
		   
		   

Information for Consumer LRPS > TestTopic1-1 created 2025-03-06T18:27:05+05:30

Configuration:

                    Name: TestTopic1-1
               Pull Mode: true
          Filter Subject: LRPS.Test.TestTopic1.*.1
          Deliver Policy: New
              Ack Policy: Explicit
                Ack Wait: 1h0m0s
           Replay Policy: Instant
         Max Ack Pending: 1,000
       Max Waiting Pulls: 512

Metadata:

              ConsumerID: 4571bea9-52ac-4586-92d7-93e036e5d663
    LastMsgDeliveredTime: 2025-03-07 10:24:31

Cluster Information:

                    Name: nats
                  Leader: nats-1
                 Replica: nats-0, current, seen 92ms ago
                 Replica: nats-2, current, seen 90ms ago

State:

  Last Delivered Message: Consumer sequence: 2,193 Stream sequence: 245,428 Last delivery: 6m47s ago
    Acknowledgment Floor: Consumer sequence: 2,193 Stream sequence: 240,839 Last Ack: 6m47s ago
        Outstanding Acks: 0 out of maximum 1,000
    Redelivered Messages: 0
    Unprocessed Messages: 0
           Waiting Pulls: 1 of maximum 512

As per #654 I have created single instance of connection per microservice & multiple application threads produce on same connection. This is the logic of producer

public async Task ProduceAsync(string topic, Message<string, string> message)
    {
        try
        {
            if(IsNatsConnectionAlive())
            {
                List<Task> produceTasks = [];
                NatsMessageModel natsMessage = new()
                {
                    Key = message.Key,
                    Value= message.Value
                };
                if(!topicModeDictionary.ContainsKey(topic) || topicModeDictionary[topic]==0) 
                {
                    Logger($"Producd message on core nats : {topic}", Levels.Info);
                    produceTasks.Add(_natsConnection!.PublishAsync(topic,natsMessage,serializer:natsMesgSerializer).AsTask());
                }
                else
                {
                    var js_topic = $"{topic}.{message.Key}";
                    Logger($"Producd message on jetstream : {js_topic}", Levels.Info);
                    produceTasks.Add(_natsJSContext.PublishAsync(js_topic,natsMessage,serializer:natsMesgSerializer).AsTask());
                }
                if(Settings.NatsSettings.IsBackupStreamRequired)
                {
                    produceTasks.Add(pubSubBackupService.ProduceAsync(topic,message,"1"));
                }
                await Task.WhenAll(produceTasks).ConfigureAwait(false);
                IncrementMessagesProduced(topic);
                producedMessageInfo[topic] = DateTime.UtcNow.ToString(CultureInfo.InvariantCulture);
            }
            else
            {
                Logger($"Nats Connection was null or not open in  ProduceAsync", Levels.Critical);
            }
        }
        catch(Exception ex)
        {
            Logger($"Exception while producing to topic {topic} message: {message.Value} Exception : {ex}", Levels.Error);
        }
    }

Consistently we are getting this exceptions in microservice application logs & things are not working smoothly in our application

7:34:31 ~ Error ~ Exception while producing to topic LRPS.Test.TestTopic1 Exception : NATS.Client.JetStream.NatsJSPublishNoResponseException: No response received from the server at NATS.Client.JetStream.NatsJSContext.PublishAsync[T](String subject, T data, INatsSerialize1 serializer, NatsJSPubOpts opts, NatsHeaders headers, CancellationToken cancellationToken) at NatsUtility.Core.Services.NatsService.ProduceAsync(String topic, Message2 message) in /app/NatsUtility.Core/Services/NatsService.cs:line 247

we have over 1000+ occurances of this exception but we can see in our test cases around 70% things work but 30% dont work fine.

Sometimes when starting consumers we get this exception too

NATS.Client.JetStream.NatsJSApiNoResponseException: No API response received from the server: Timeout 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.CreateOrUpdateConsumerInternalAsync(String stream, ConsumerConfig config, ConsumerCreateAction action, CancellationToken cancellationToken) at NATS.Client.JetStream.NatsJSContext.CreateOrUpdateConsumerAsync(String stream, ConsumerConfig config, CancellationToken cancellationToken) at NatsUtility.Core.Services.NatsConsumer.StartJetstreamConsumerAsync(Boolean isPartitioned, Int32 partitionId) in /app/NatsUtility.Core/Services/NatsConsumer.cs:line 169

As stated in #349, We assumed our nodes were corrupted but we confirmed by running the same test on only 1 broker with 1 broker connection as shown below with replication of 1

Original Connection String: "nats://nats-0.nats-headless.nats.svc.cluster.local:4222,nats://nats-1.nats-headless.nats.svc.cluster.local:4222,nats://nats-2.nats-headless.nats.svc.cluster.local:4222"

New Connection String: "nats://nats-0.nats-headless.nats.svc.cluster.local:4222" (Tested 1 by 1 from applications)

We are able to see same exception even during 1 broker connection tests. Can anyone pls provide any feedback on what we are doing wrong?

Nats.yaml

---
# Source: nats/templates/stateful-set.yaml
apiVersion: apps/v1
kind: StatefulSet
metadata:
  labels:
    app.kubernetes.io/component: nats
    app.kubernetes.io/instance: nats
    app.kubernetes.io/managed-by: Helm
    app.kubernetes.io/name: nats
    app.kubernetes.io/version: 2.10.25
    helm.sh/chart: nats-1.2.10
  name: nats
spec:
  podManagementPolicy: Parallel
  replicas: 3
  selector:
    matchLabels:
      app.kubernetes.io/component: nats
      app.kubernetes.io/instance: nats
      app.kubernetes.io/name: nats
  serviceName: nats-headless
  template:
    metadata:
      annotations:
        checksum/config: 62e0788bac90b80358eaaa8a0ca51168da7d7cf3fa4720b93e13b161e36e692f
      labels:
        app.kubernetes.io/component: nats
        app.kubernetes.io/instance: nats
        app.kubernetes.io/managed-by: Helm
        app.kubernetes.io/name: nats
        app.kubernetes.io/version: 2.10.25
        helm.sh/chart: nats-1.2.10
    spec:
      affinity:
        nodeAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            nodeSelectorTerms:
            - matchExpressions:
              - key: service
                operator: In
                values:
                - nats
      containers:
      - args:
        - --config
        - /etc/nats-config/nats.conf
        env:
        - name: POD_NAME
          valueFrom:
            fieldRef:
              fieldPath: metadata.name
        - name: SERVER_NAME
          value: $(POD_NAME)
        - name: GOMEMLIMIT
          value: "3GiB"
        image: nats:2.10.26-alpine
        lifecycle:
          preStop:
            exec:
              command:R
              - nats-server
              - -sl=ldm=/var/run/nats/nats.pid
        livenessProbe:
          failureThreshold: 3
          httpGet:
            path: /healthz?js-enabled-only=true
            port: monitor
          initialDelaySeconds: 10
          periodSeconds: 30
          successThreshold: 1
          timeoutSeconds: 5
        name: nats
        ports:
        - containerPort: 4222
          name: nats
        - containerPort: 6222
          name: cluster
        - containerPort: 8222
          name: monitor
        readinessProbe:
          failureThreshold: 3
          httpGet:
            path: /healthz?js-server-only=true
            port: monitor
          initialDelaySeconds: 10
          periodSeconds: 10
          successThreshold: 1
          timeoutSeconds: 5
        startupProbe:
          failureThreshold: 90
          httpGet:
            path: /healthz
            port: monitor
          initialDelaySeconds: 10
          periodSeconds: 10
          successThreshold: 1
          timeoutSeconds: 5
        resources:
          requests:
            cpu: "1500m"
            memory: "3500Mi"
          limits:
            cpu: "1500m"
            memory: "3500Mi"
        volumeMounts:
        - mountPath: /etc/nats-config
          name: config
        - mountPath: /var/run/nats
          name: pid
        - mountPath: /data
          name: nats-js
      - args:
        - -pid
        - /var/run/nats/nats.pid
        - -config
        - /etc/nats-config/nats.conf
        image: natsio/nats-server-config-reloader:0.16.1
        name: reloader
        volumeMounts:
        - mountPath: /var/run/nats
          name: pid
        - mountPath: /etc/nats-config
          name: config
      - args:
        - -port=7777
        - -connz
        - -routez
        - -subz
        - -varz
        - -prefix=nats
        - -use_internal_server_id
        - -jsz=all
        - http://localhost:8222/
        image: natsio/prometheus-nats-exporter:0.15.0
        name: prom-exporter
        ports:
        - containerPort: 7777
          name: prom-metrics
      enableServiceLinks: false
      shareProcessNamespace: true
      volumes:
      - configMap:
          name: nats-config
        name: config
      - emptyDir: {}
        name: pid
  volumeClaimTemplates:
  - metadata:
      name: nats-js
    spec:
      accessModes:
      - ReadWriteOnce
      resources:
        requests:
          storage: 8Gi
      storageClassName: manual

[mtmk]

on your services what is the number of threads and how many CPUs? I suspect threadpool maybe starving if you're seeing a lot of timeouts on jetstream publish. I recommend to await on the publishes rather than collecting them as tasks or use Concurrent publishing if you need to publish fast. there is an example here

[romitkmehta]

Hello @mtmk,
Thank you for your reply. I did give it a try by removing Task.WhenAll and I awaited each publish operation instead of collecting them in the List of task and running them at once but no luck. I'd love to hear the clarification that why it could be due to thread pool starvation?
I could provide you metrics/logs or sample code if you would like to have a look.
Thanks!

[mtmk]

you're welcome. yes it would be good to see the code (especially if something I can reproduce on my end be very helpful). do you see the issue on a local server as well for example or is it only when running against the cluster? threadpool thing is just a guess when I see code not await publish that usually is problematic.

[romitkmehta]

Hey @mtmk
I hope you’re all doing well and thank your for your engagement. As suggested by you,
I’ve updated my ProduceAsync method to await each publish operation sequentially instead of collecting them into tasks and using Task.WhenAll. Here’s the revised code:

public async Task ProduceAsync(string topic, Message<string, string> message)
{
    try
    {
        if (IsNatsConnectionAlive())
        {
            NatsMessageModel natsMessage = new() { Key = message.Key, Value = message.Value };
            if (!topicModeDictionary.ContainsKey(topic) || topicModeDictionary[topic] == 0)
            {
                Logger($"Produced message on core NATS: {topic}", Levels.Info);
                await _natsConnection!.PublishAsync(topic, natsMessage, serializer: natsMesgSerializer).AsTask();
            }
            else
            {
                var js_topic = $"{topic}.{message.Key}";
                Logger($"Produced message on JetStream: {js_topic}", Levels.Info);
                await _natsJSContext.PublishAsync(js_topic, natsMessage, serializer: natsMesgSerializer).AsTask();
            }
            if (Settings.NatsSettings.IsBackupStreamRequired)
            {
                await pubSubBackupService.ProduceAsync(topic, message, "1");
            }
            IncrementMessagesProduced(topic);
            producedMessageInfo[topic] = DateTime.UtcNow.ToString(CultureInfo.InvariantCulture);
        }
        else
        {
            Logger("NATS connection was null or not open in ProduceAsync", Levels.Critical);
        }
    }
    catch (Exception ex)
    {
        Logger($"Exception while producing to topic {topic} message: {message.Value} Exception: {ex}", Levels.Error);
    }
}

This doesn't work as well and I observe the same exceptions as described above. Apparently, the code which I've had shared before (collecting all the produce tasks into list and executing it together) works very well in my local. Since the full application code is proprietary, I can’t share it directly, but I’m working on a sample app to reproduce the issue and will share it once it’s ready.

In the meantime, I’d really appreciate any insights or suggestions on what might be causing these timeouts. This has become a significant roadblock for our NATS implementation, and I’d be grateful for any guidance to help us resolve it. If you need more specifics about our setup (e.g., metrics, logs), just let me know—I’m happy to provide what I can.
Or if you want me to try a debug build with verbose logging I can try that too!

[mtmk]

thanks you. Really appreciate it if you can give me a sample app to reproduce on my end. that would be very helpful. in the mean time could please you give me a few metrics like message rates being published, other operations going on especially on the same connection and the system the application is running on and it's proximity to the cluster if not on the same datacenter. cheers.

[mtmk]

hi @romitkmehta, also been experimenting with a new JetStream publish API here if you have some to have a look would appreciate the feedback. thank you