How to troubleshoot messages not picked up by consumer

[yonigibbs]

Hi All,

I am doing some chaos engineering and can pretty reliably replicate a situation where a message in a work queue seems to be stuck, i.e. it's in the queue but not picked up by the queue consumer. Messages before it were all picked up correctly, and newly published messages after it are also being picked up correctly.

To get into this situation I load, say, 100 messages into the queue, then start a JVM process which is processing the items (using Kotlin, in a codebase heavily using coroutines, in case it ends up being relevant). This process picks up multiple messages and processes them concurrently. In that process I cause a java.lang.OutOfMemoryError to occur in some thread. In response to this we shut down all running coroutines, NAKing the messages that are in progress, and unsubscribe the JetStream subscription. The JVM process runs for a few seconds, processes some of the items, then dies as expected. If I run it a few times eventually it manages to process all the items in the queue, but one or two are always left in the queue. If I then remove the OutOfMemoryError and leave the process up and running, it never picks up the message, no matter how long it's left running.

Because this is simulating a crash of the system, it's possible that there are some race conditions between threads NAKing the messages they're processing, and the subscription being unsubscribed. But I'd expect the system to eventually recover and for the message to eventually be made available to some new consumer that starts up. However this doesn't seem to happen. It seems to remain stuck in the queue.

Does anyone have any ideas how to start troubleshooting this?

Note that if I stop the Java process, remove the consumer, then start up the Java process (which programmatically re-creates the consumer), then this time it does pick up the "stuck" message.

I'm using NATS 2.10-alpine, just running in Docker Compose locally at the moment. Below are some commands I ran that show information that might be pertinent...

nats stream info

$ nats stream info MY-QUEUE
Information for Stream MY-QUEUE created 2024-01-08 16:33:07

           Description: foo bar
              Replicas: 1
               Storage: File

Options:

             Retention: WorkQueue
       Acknowledgments: true
        Discard Policy: Old
      Duplicate Window: 0s
            Direct Get: true
     Mirror Direct Get: true
     Allows Msg Delete: true
          Allows Purge: true
        Allows Rollups: false

Limits:

      Maximum Messages: 100,000,000
   Maximum Per Subject: unlimited
         Maximum Bytes: unlimited
           Maximum Age: unlimited
  Maximum Message Size: unlimited
     Maximum Consumers: unlimited

Replication:

                Mirror: MY-QUEUE-ARCHIVE

Mirror Information:

           Stream Name: MY-QUEUE-ARCHIVE
                   Lag: 0
             Last Seen: 757ms

State:

              Messages: 1
                 Bytes: 189 B
        First Sequence: 71 @ 2024-01-08 16:41:45 UTC
         Last Sequence: 127 @ 2024-01-08 16:41:45 UTC
      Deleted Messages: 56
      Active Consumers: 1
    Number of Subjects: 1

nats consumer info

$ nats consumer info MY-QUEUE MY-QUEUE-C-UNKNOWN
Information for Consumer MY-QUEUE > MY-QUEUE-C-UNKNOWN created 2024-01-08T16:33:07Z

Configuration:

                    Name: MY-QUEUE-C-UNKNOWN
             Description: foo bar
               Pull Mode: true
          Filter Subject: my.queue
          Deliver Policy: All
              Ack Policy: Explicit
                Ack Wait: 2m0s
           Replay Policy: Instant
         Max Ack Pending: 1,000
       Max Waiting Pulls: 512
      Inactive Threshold: 5d0h0m0s

State:

  Last Delivered Message: Consumer sequence: 147 Stream sequence: 127 Last delivery: 5m32s ago
    Acknowledgment Floor: Consumer sequence: 147 Stream sequence: 127 Last Ack: 5m31s ago
        Outstanding Acks: 0 out of maximum 1,000
    Redelivered Messages: 0
    Unprocessed Messages: 0
           Waiting Pulls: 1 of maximum 512

nats consumer report

$ nats consumer report MY-QUEUE
╭─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
│                                   Consumer report for MY-QUEUE with 1 consumers                                   │
├──────────────────────────────┬──────┬────────────┬──────────┬─────────────┬─────────────┬─────────────┬───────────┬─────────┤
│ Consumer                     │ Mode │ Ack Policy │ Ack Wait │ Ack Pending │ Redelivered │ Unprocessed │ Ack Floor │ Cluster │
├──────────────────────────────┼──────┼────────────┼──────────┼─────────────┼─────────────┼─────────────┼───────────┼─────────┤
│ MY-QUEUE-C-UNKNOWN           │ Pull │ Explicit   │ 2m0s     │ 0           │ 0           │ 0           │ 127       │         │
╰──────────────────────────────┴──────┴────────────┴──────────┴─────────────┴─────────────┴─────────────┴───────────┴─────────╯

nats stream view

$ nats stream view MY-QUEUE
[71] Subject: my.queue Received: 2024-01-08T16:41:45Z

{"my":"json"}


16:52:10 Reached apparent end of data

(For the latter I had to stop the service and delete the consumer otherwise I hit the error described in this issue: nats-io/natscli#920)

Thanks!

[yonigibbs]

Some extra info:

• I notice the Ack Floor on the consumer is 127, but the stream contains a message with sequence number 71. Should that be possible in a work queue? If the floor is 127 shouldn't message 71 be gone, like all the rest are?
• From our logs it looks like our consumer never received 71. Maybe it received it but the app crashed before it managed to deliver it to our code which logs the sequence number.
• I tried to use the NATS CLI to get at that message. See below...

$ nats consumer sub --no-ack MY-QUEUE MY-QUEUE-C-UNKNOWN
nats: error: no message received: nats: timeout

[yonigibbs]

@pdebakker-auguria: myself and @gregorycollins had a long conversation on Slack with Scott, we supplied them with a couple of git repos that they could use to reproduce this issue, and last we heard (a couple of months back), the team had recognised this as a bug in the NATS server (I think) and it was on their backlog. I haven't heard since. Unfortunately for our project we had to back out of using NATS because of this issue, which is a shame, coz it has a lot of really great features we were using. Hopefully someone from the NATS team will be able to give you more info. Good luck!

[scottf]

It's a verified issue and it's being worked on. I don't know the current status, I've asked someone on the server team to reply.

[wallyqs]

This issue is something that we have improved in the v2.10.14 release (like in #5227). Thanks again for putting together the @yonigibbs with the reproduction, I ran it with the latest version and this does not show up anymore using your test.
@pdebakker-auguria which version are you running on at the moment? please let us know if you still see this issue with the v2.10.14 version in your setup.

[pdebakker-auguria]

@wallyqs If I run into the issue again I will save the nats server version and let you know. Locally we just use the :latest version so I'm not quite sure on the version number.
Last week we also finally updated the version of our nats.go package.
Truth be told, I haven't seen the issue this week anymore, but I am hesitant to say it is gone completely because I've cried wolf about this before to my team (saying it was fixed because I hadn't seen the issue in a while and it then popping up again a day later).

But thank you guys for the hard work and the update here. It is much appreciated!

[wmxhw]

@wallyqs I had a similar problem using the InterestPolicy，when send kill -9 to running nats.
nats log:
[WRN] Detected consumer '$G > {my stream name} > {my consumer name}' ack floor xxxx is ahead of stream's last sequence xxxx

nats-server version v2.10.14
golang client github.com/nats-io/nats.go v1.34.1

I tried modifying the configuration but it still didn’t work.

nats.conf:
...
jetstream {
...
sync_interval: 5s
}
...

Thanks!