jetstream: slow consumer

[oliverpool]

Hi,

I have an application which has to do some heavy processing on every message received.

I created a Jetstream subscription, to be sure not to miss any message and send a Msg.InProgress() regularly during processing.
Due to some external constraints, my program can only handle one message at a time and I was hoping that I could rely on NATS: Jetstream to handle this for me.

However I still get nats: slow consumer, messages dropped on connection [16] for subscription on ....

What should I do, to prevent those errors?

Here is the logic of my code:

func main() {
	nc, err := nats.Connect(url,
		nats.DisconnectErrHandler(func(nc *nats.Conn, err error) {
			logger.Log("nats", "disconnected", "reason", err)
		}),
		nats.ReconnectHandler(func(nc *nats.Conn) {
			logger.Log("nats", "reconnected")
		}),
	)

	js, err := nc.JetStream(nats.PublishAsyncMaxPending(256)) //nolint:gomnd

	_, err = js.AddStream(&nats.StreamConfig{
		Name:     "BARN",
		Subjects: []string{"BARN.>"},
	})

	msgs := make(chan *nats.Msg)
	js.ChanQueueSubscribe(subj, queue, msgs,
		nats.Durable(queue),
		nats.MaxAckPending(2048),
		nats.AckWait(3*keepaliveDelay/2), // add some margin
	)

	for m := range msgs {
		// keepalive to inform nats of "InProgress"
		sendNak := make(chan bool)
		go keepalive(m, sendNak)

		// long processing
		time.Sleep(5 * keepaliveDelay)

		// everything was fine
		close(sendNak)
		err = m.Ack()
	}
}

func keepalive(m *nats.Msg, sendNak <-chan bool) {
	for {
		select {
		case <-time.After(keepaliveDelay):
			_ = m.InProgress()
		case nak := <-sendNak:
			// when something is wrong, but might work later
			if nak {
				time.Sleep(nakDelay)
				_ = m.Nak()
			}
			return
		}
	}
}

[wallyqs]

The channel is unbuffered so it is prone to dropping messages. I suggest using a larger buffer size like 65536 or more for jetstream, depending on the type of traffic.

[oliverpool]

Thank you for your reply.

If I create a large buffer, but slowly consume them, then I will still be considered a "slow consumer" since no message will be Acked in time, right?

I think that I should switch to a pull approach, so that I only fetch a message when I am ready, however I have a bug there when the nats-server goes down: #890

[derekcollison]

Slow consumer is a core NATS condition where a NATS system will protect itself from apps that consume all messages on a connection at a rate that cause buildup in the NATS system itself.

For MaxAckPending, that will not disconnect any NATS clients but will stop messages from flowing if that limit is exceeded.

[oliverpool]

@derekcollison thank you so much, I mis-understood the MaxAckPending setting.
Setting it to 1 seems to fix everything!

I can now reliably use a ChanQueueSubscribe with my slow subscriber.

[oliverpool]

To not be treated as a slow consumer, I adjusted the nats.MaxAckPending(1), setting, as hinted by @derekcollison 👍