What is the exact meaning of "Unprocessed Messages" in the "nats consumer info" output?

[fmontorsi-equinix]

Context for this question: I wrote a small golang example to understand the impact of settings like the "MaxWaiting" parameter of jetstream.ConsumerConfig.
I'm using Jetstream and nats:2.10.16.

My example app is composed by:

• a single publisher goroutine publishing every 1 sec a spike of 1000 messages to a Jetstream stream
• 10 consumer using jetstream.Consumer.Consume() method to run async reception
  The app runs for 10 sec and then stops and prints how many messages were published with NO error and how many messages were processed by the 10 consumers.

What I noticed is that if MaxWaiting is large enough, then at the end of my test, the "nats consumer info" output looks good:

Configuration:

                    Name: processor
               Pull Mode: true
          Filter Subject: events.*
          Deliver Policy: All
              Ack Policy: Explicit
                Ack Wait: 10.00s
           Replay Policy: Instant
         Max Ack Pending: 1,000
       Max Waiting Pulls: 100,000

State:

  Last Delivered Message: Consumer sequence: 10,000 Stream sequence: 10,000 Last delivery: 4.08s ago
    Acknowledgment Floor: Consumer sequence: 10,000 Stream sequence: 10,000 Last Ack: 4.08s ago
        Outstanding Acks: 0 out of maximum 1,000
    Redelivered Messages: 0
    Unprocessed Messages: 0
           Waiting Pulls: 0 of maximum 100,000

but if I start to decrease the MaxWaiting config for the stream, at the end of my app I see 'Unprocessed Messages' being potentially very large.

Configuration:

                    Name: processor
               Pull Mode: true
          Filter Subject: events.*
          Deliver Policy: All
              Ack Policy: Explicit
                Ack Wait: 10.00s
           Replay Policy: Instant
         Max Ack Pending: 1,000
       Max Waiting Pulls: 10

State:

  Last Delivered Message: Consumer sequence: 5,000 Stream sequence: 5,000 Last delivery: 1m8s ago
    Acknowledgment Floor: Consumer sequence: 5,000 Stream sequence: 5,000 Last Ack: 1m8s ago
        Outstanding Acks: 0 out of maximum 1,000
    Redelivered Messages: 0
    Unprocessed Messages: 5,000
           Waiting Pulls: 0 of maximum 10

So my question is what is the exact meaning of Unprocessed Messages stats? Does that indicates an issue on the Publisher-side (which is strange since my app reports 0 publishing errors) or rather an issue on the Consumer-side?

thanks!

[ripienaar]

Unprocessed is a message that has never had any delivery attempts - in other words its messages not yet received by the clients at all.

[fmontorsi-equinix]

Thanks @ripienaar ,
So if I get "unprocessed messages" greater than zero, is that a problem on the PUB or the SUB side?

thanks

[Jarema]

It means that Consumer didn't yet asked for given number of messages.

That is not an issue and can have different reasons to happen:

• publishing faster than consuming
• not consuming at all
• in the process of going through the messages.

If you reduce Max Waiting, you are limiting the clients how quickly they can consume messages and that is most probably the cause.

10 is very little. Each client using Consume tries to optimize the fetching process and will have pending 1 or 2 Pull requets.

[fmontorsi-equinix]

thanks @Jarema ,
According to your explanation, in my example it looks like I'm limiting the clients by reducing Max Waiting a lot.
However I would expect that, even if limited, my 10 consumers should, at some point recover and be able to process all msgs in the stream, if I stop publishing. I think this is not happening. Instead what happens is that the Consumers seem to get stuck: I don't see any more any logged message from my 10 goroutines saying "processing message"... what am I missing?

thanks

[fmontorsi-equinix]

Indeed I can confirm that using the ConsumeErrHandler (https://pkg.go.dev/github.com/nats-io/nats.go/jetstream#ConsumeErrHandler ) in my consumer, when MaxWaiting is low, I do get my error handler invoked with an error "nats: Exceeded MaxWaiting".
This error happens for all 10 consumers simultaneously. After that they process a few more messages and finally they get completely stuck.

The result is something like in my original post: a big gap between the number of messages published (10000) and those actually consumed.
This sounds like a bug to me, as I would expect the consumer to never stop (perhaps process messages at a lower pace) just by reducing the value of MaxWaiting... what do you think?

My main concern is that I don't have any good criteria to size MaxWaiting and I'm concerned that in production, even if I set MaxWaiting to something really large, my consumers might get stuck at some point...

thanks

[fmontorsi-equinix]

Btw I wonder if this nats-io/nats.go#1709 could be the fix I'm looking for...