Jetstream messages being handled after redelivery

[Guillembonet]

Observed behavior

Given the following scenario:

1. One consumer which runs in multiple replicas (i.e. Kubernetes deployment with replicas) (for example 2)
2. Multiple existing messages in the waiting to be delivered (for example 10)
3. Consumer runs with Consume() or Messages() (same outcome), and uses a dummy handler which waits 28 seconds, then acks.
4. The Jetstream server uses the default AckWait of 30 seconds.

I observe the following:

1. Replica 1 runs Consume() (or .Next())
2. Replica 1 starts processing message 1.
3. Replica 2 runs Consume() (or .Next()), does nothing because Replica 1 pre-buffered all 10 messages.
4. Replica 1 acks message 1 and starts processing message 2.
5. Due to AckWait being reached, the unacknowledged messages are redelivered, so Replica 2 starts processing message 2.
6. Replica 1 and 2 process messages 3 to 8 because they both pre-buffered all of them.

This effectively leads to 2 undesirable outcomes:

1. Both replicas consume almost all messages (which should be fine given consumers are idempotent), and this leads to more replicas not increasing the consumption throughput.
2. Consume calls the provided handler function with messages which are already expired due to AckWait.

This is an edge case which only occurs when there are multiple replicas of the same consumer running, messages are slow to process and there are multiple messages waiting to be delivered.

Expected behavior

I would say there is 2 options:

1. This is the expected behaviour and it would be useful to add it to the readme.
2. Users of Consume() or Messages() should receive messages which have the full AckWait timeout to process instead of the effective AckWait being related to how long it took to process messages within the current batch.

If this is expected behaviour I would like to know if there is a simple way to overcome this. For now, the most reasonable to me seems to use Fetch() instead and create a context with AckWait as timeout right after.

Server and client version

server: v2.11.8
client: v1.45.0

Host environment

No response

Steps to reproduce

To replicate the observed behaviour you can use the following 2 snippets:
server: to create streams and consumers and send some messages:

func main() {
	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
	defer cancel()

	nc, err := nats.Connect("nats://localhost:4222")
	if err != nil {
		log.Fatal(err)
	}

	js, err := jetstream.New(nc)
	if err != nil {
		log.Fatal(err)
	}
	defer js.CleanupPublisher()

	stream, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
		Name:      "events",
		Retention: jetstream.InterestPolicy,
		Subjects:  []string{"events.>"},
	})
	if err != nil {
		log.Fatal(err)
	}

	stream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
		Durable: "processor-1",
	})

	for i := range 10 {
		_, err := js.Publish(ctx, "events.example", []byte(fmt.Sprint(i)))
		if err != nil {
			log.Fatal(err)
		}
	}

	slog.Info("Published events")
}

client: to use the consumer and ack messages after 28 seconds:

func main() {
	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
	defer cancel()

	nc, _ := nats.Connect("nats://localhost:4222")

	js, err := jetstream.New(nc)
	if err != nil {
		log.Fatal(err)
	}

	cons, err := js.CreateOrUpdateConsumer(ctx, "events", jetstream.ConsumerConfig{
		Durable: "processor-1",
	})
	if err != nil {
		log.Fatal(err)
	}

	cc, _ := cons.Consume(func(msg jetstream.Msg) {
		fmt.Println("got msg", string(msg.Data()))
		time.Sleep(28 * time.Second)
		msg.Ack()
		fmt.Println("acked msg", string(msg.Data()))
	})
	defer cc.Stop()

	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, os.Interrupt)

	<-sigChan
}

to replicate run a container with jetstream: docker run -d --rm --name nats -p 4222:4222 nats:2.11 -js, run the first snippet once, then run the second snippet 2 times in parallel.

[ripienaar]

imo whenever you have slow message processing and such a very short ack wait, the only thing that make sense is to fetch 1 message at a time.

[Guillembonet]

imo whenever you have slow message processing and such a very short ack wait, the only thing that make sense is to fetch 1 message at a time.

@ripienaar sure, this is a valid outcome, but what I presented was a simplified case. By default the MaxMessages is set to 500, which means that if there is 500 messages waiting for delivery and you take 0.5 seconds to consume each, this will still cause the scenario I presented for 440 messages (30 seconds / 0.5msg/second = 60 messages consumed before ackwait causes redelivery).

Of course this is an edge case which only happens when messages are not consumed fast enough and they accumulate, but the issue becomes worse at that point because the consumers are effectively halving the consumption throughput due to double consumption.

I would say this is still relevant given that with the existing defaults, this is not such an unusual scenario, no?

I'm not saying this is a bug too, but I couldn't find info on how other people is handling this situation. Maybe they are just ensuring consumers have a higher consumption rate than publish rate of events and then ignoring this.

[ripienaar]

I am saying two things:

• Do not pre-fetch messages so any consumer is only handling 1 message at a time and if it takes a hour, its not going to move onto expired messages
• Don't rely on defaults for ack wait, this is a property that should be tuned for your needs there is no one-size-fits-all value for ack wait

[Guillembonet]

Okay, thanks for your time and input!

I guess this is just something to keep in mind when choosing stream/consumer settings, but not something the library should solve.

In my use-case I think we will either rely on setting tuning and monitoring (and maybe disable pre-fetch with the PullMaxMessages option), and if we would really want to implement "safer" batch consumers we can do that using fetch and injecting a context which times out after ackwait on our own.

Hope this helps others :)

Feel free to close the issue if you believe this is fine.

[Guillembonet]

btw, this is might be an issue in client wrappers such as watermill-nats, which seem to assume ackwait starts when the message handler is called: https://github.com/ThreeDotsLabs/watermill-nats/blob/6e1bc30cf486162ba23630d078b7d3cd5562baaa/pkg/jetstream/message_handler.go#L33

cc: @AlexCuse @m110

[CosmicPredator]

@Guillembonet I think this is an expected outcome. Consume() and Messages() are NATS Jetstream's push model where the first consumer that registers will pre-fetch all the messages from the queue.

As @ripienaar suggested, you can fetch N message at a time using Fetch() which is a pull model. Another thing to note is that while doing Fetch(), the AckWait starts only after the message gets delivered to your consumer, which I think, is ideal for your case.

[Guillembonet]

@Guillembonet I think this is an expected outcome. Consume() and Messages() are NATS Jetstream's push model where the first consumer that registers will pre-fetch all the messages from the queue.

As @ripienaar suggested, you can fetch N message at a time using Fetch() which is a pull model. Another thing to note is that while doing Fetch(), the AckWait starts only after the message gets delivered to your consumer, which I think, is ideal for your case.

@CosmicPredator, Consume() and Messages() are implementing pull consumers: https://github.com/nats-io/nats.go/blob/main/jetstream/README.md#continuous-polling.

My point is that while this might be an expected behaviour, it might be valuable to add a note in the readme which explains this, as I only stumbled upon this behavior after digging deep in the library.

Also, using Fetch() is valid, but I would like to know if this will somehow be worse in performance.

[CosmicPredator]

@Guillembonet I think this is an expected outcome. Consume() and Messages() are NATS Jetstream's push model where the first consumer that registers will pre-fetch all the messages from the queue.
As @ripienaar suggested, you can fetch N message at a time using Fetch() which is a pull model. Another thing to note is that while doing Fetch(), the AckWait starts only after the message gets delivered to your consumer, which I think, is ideal for your case.

@CosmicPredator, Consume() and Messages() are implementing pull consumers: https://github.com/nats-io/nats.go/blob/main/jetstream/README.md#continuous-polling.

My point is that while this might be an expected behaviour, it might be valuable to add a note in the readme which explains this, as I only stumbled upon this behavior after digging deep in the library.

Also, using Fetch() is valid, but I would like to know if this will somehow be worse in performance.

1. Yeah it'd be nice if this info is documented.
2. Guess you need to check that for your use case and scale 🙂.

[Jarema]

TL;DR: before switching to fetch, tweak PullMaxMessages (lower it to single digits) and AckWait.

Few clarifications:

Messages() and Consumer() is pull and it doesnt mean it will consume all messages from the stream.

Both of those methods allow quite wide fine tuning, and by default, they're optimzed for performance and low IO footprint.
That means that it will do interleaved-fetching of 500 messages per batch by default.

That is optimal scenario if your client business logic can keep up with how fast JetStream can deliver messages.
However - if you're slower at processing, it will lead to redeliveries and handicap the scaling.

Definately check out https://pkg.go.dev/github.com/nats-io/nats.go/jetstream#section-readme

I suspect that in your case its not about AcksWait values, but rather, reducing - PullMaxMessages.
This will define how many messages each Fetch under the hood will get. You can of course also fine tune ack wait but in my experience, if you're reaching ack wait 30s while you normally process each message in few seconds, it simply means buffer is too big. However, if you process them on average in 28seconds, 30s is definately too small, as @ripienaar said.

With proper PullMaxMessasges it will be also better scaling across many clients, as the limiting factor seems to be a redelivery.

Consume with PullMaxMessages of, lets say, 10 or even 2, is still faster and has more optimized latency than manually calling Fetch(). That is because when you call Fetch() server is idling when the client is processing current batch, while in Consume it will get the next batch when current is partially consumed (you can further tweak that with PullThresholdMessages).

[Guillembonet]

@Jarema yes, seems reasonable! thanks for your time!

Do you think it would be useful to add a note in the docs (maybe here) to note that due to pre-buffering, handler functions are called with messages that might have been pulled before, and that, therefore, the effective ackwait is unpredictable?

Maybe not this specific, but I see this lead to a misconception in my case and I would say the same for watermill-nats.

[CosmicPredator]

TL;DR: before switching to fetch, tweak PullMaxMessages (lower it to single digits) and AckWait.

Few clarifications:

Messages() and Consumer() is pull and it doesnt mean it will consume all messages from the stream.

Both of those methods allow quite wide fine tuning, and by default, they're optimzed for performance and low IO footprint. That means that it will do interleaved-fetching of 500 messages per batch by default.

That is optimal scenario if your client business logic can keep up with how fast JetStream can deliver messages. However - if you're slower at processing, it will lead to redeliveries and handicap the scaling.

Definately check out https://pkg.go.dev/github.com/nats-io/nats.go/jetstream#section-readme

I suspect that in your case its not about AcksWait values, but rather, reducing - PullMaxMessages. This will define how many messages each Fetch under the hood will get. You can of course also fine tune ack wait but in my experience, if you're reaching ack wait 30s while you normally process each message in few seconds, it simply means buffer is too big. However, if you process them on average in 28seconds, 30s is definately too small, as @ripienaar said.

With proper PullMaxMessasges it will be also better scaling across many clients, as the limiting factor seems to be a redelivery.

Consume with PullMaxMessages of, lets say, 10 or even 2, is still faster and has more optimized latency than manually calling Fetch(). That is because when you call Fetch() server is idling when the client is processing current batch, while in Consume it will get the next batch when current is partially consumed (you can further tweak that with PullThresholdMessages).

I also agree that in most of the cases, Consume() with PullMaxMessages = N produces same behaviour as as Fetch(N). But I got a question. In case of using Consume(), at what phase, the AckWait starts? Right after the messages gets queued or when the messages is received by the consumer?