NATS Jetstream: Freeze consumer

[truongnhatanh7]

Context: I got a running http server. Before the http server is ran, I run RegisterListener on main goroutine to register the listener.

func (b *biz) RegisterListener(ctx context.Context) error {
	stream, _ := b.initSyncStream(ctx) // Use create or update stream, then return the stream
	
	consumer, _ := stream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
		Durable:       "sync-consumer",
		FilterSubject: "debug",
		DeliverPolicy: jetstream.DeliverAllPolicy,
		AckPolicy:     jetstream.AckExplicitPolicy,
	})

	go func() {
		consumeCtx, err := consumer.Consume(func(msg jetstream.Msg) {
			_ = msg.InProgress()

			// Decode msg
			var event XEvent
			err = json.Unmarshal(msg.Data(), &event)
			if err != nil {
				return
			}

			// Main sync logic
			// Do something
			
			// Ack
			_ = msg.Ack()
		}, jetstream.ConsumeErrHandler(func(consumeCtx jetstream.ConsumeContext, err error) {
			logger.Logger.Errorf("failed to consume msg: %v", err)
		}))
		if err != nil {
			return
		}
		defer consumeCtx.Stop()

		<-ctx.Done()
		logger.Logger.Println("pi-listener: stopping signal handler")
	}()

	return nil
}

Next: I deployed nats via helm chart on a VPS on K8S (https://nats-io.github.io/k8s/helm/charts/), I could use the CLI from local to list all streams in the remote NATS.

My problem: Using nats docker image on local, my code could dispatch and consume event normally (default setting + jetstream), but when using NATS on k8s, it hangs indefinitely.
This is my custom config:

project: default
source:
  repoURL: https://nats-io.github.io/k8s/helm/charts/
  targetRevision: 1.3.2
  helm:
    parameters:
      - name: config.cluster.replicas
        value: '3'
      - name: config.jetstream.enabled
        value: 'true'
      - name: config.cluster.enabled
        value: 'true'
  chart: nats
destination:
  namespace: staging
  name: in-cluster

The rest is default.
I've tried serveral consume pattern with the local nats docker and it works fine so I don't think the problem is coming from the consume logic. The problem only occurs when I use local + nats on k8s (vps), or when I deploy my code to ECS + nats on k8s (vps). I did check all the subject name + network ping + ..., it all works fine.

[truongnhatanh7]

Update, use the demo nats server to test and the issue still persists -> The problem maybe originated from client side

[piotrpio]

Hello @truongnhatanh7. So when it freezes, you don't see any errors being reported in ConsumeErrHandler?

And second question, are you sure there is no error when running consumer.Consume()? I see that you're silently returning from the goroutine in case of an error, if that's the case it's worth checking if Consume() is really running.

consumeCtx, err := consumer.Consume(func(msg jetstream.Msg) {
	// handler
}, jetstream.ConsumeErrHandler(func(consumeCtx jetstream.ConsumeContext, err error) {
	logger.Logger.Errorf("failed to consume msg: %v", err)
}))
if err != nil {
        // log error here
	return
}

[truongnhatanh7]

Hi, currently, I've just fixed it. The reason might be the way that I dispatch the message (I create a record in DB to save the status of the background job, then dispatch the event). I'll monitor it and get back to you if the problem is still there