Stream consumer hangs when running out of messages

[Raz Hemo]

Hello,

TL;DR - When starting an AMQP consumer for a stream queue in a rabbit cluster, it hangs indefinitely when reaching the end of the stream.

We made a simple application using python kombu that consumes from a RabbitMQ stream that has a lot of messages (for example 10k) in it. However, we encountered an issue where the consumer hangs when reaching the last message, and upon resuming the production of messages, the consumer will not continue receiving them until it is restarted. 

We've managed to figure out that:

* this only happens when consuming from a cluster. When there is only a single rabbit node there is no hang.

* this only happens when using the AMQP wrapper. Using the stream plugin, there is no hang.

* This problem also occurs when using the rust client (lapin), so it is not client-specific.

To simulate the problem, simply create a stream and:

- start publisher, stop it after a few seconds

- start consumer and wait for it to reach the end of the stream

- start publisher again

- the consumer should not receive those new messages as they arrive.

Here is our consumer:

```

from kombu import Connection, Exchange, Consumer, Queue  # noqa

def callback(body, message):

    print(message)

    message.ack()

with Connection('amqp://rabbitmq:rabbitmq@rabbit:5672//') as conn:

    with Consumer(conn, queues=[Queue('predictions', exchange=Exchange('predictions', type='fanout', durable=False), durable=True, queue_arguments={'x-queue-type':'stream'}, consumer_arguments={'x-stream-offset': 0})], prefetch_count=1000, callbacks=[callback]):

        while True:

            conn.drain_events()

```

[Gabriele Santomaggio]

We can't reproduce the issue.

here is the code, publisher:

```

from kombu import Connection, Exchange, Consumer, Queue  # noqa

with Connection('amqp://test:test@localhost//') as conn:

    # produce

    producer = conn.Producer(serializer='json')

    for i in range (1,10000):

      producer.publish( {'hello': 'world'},

                        exchange="predictions")

```

consumer:

```

from kombu import Connection, Exchange, Consumer, Queue, eventloop  # noqa

def callback(body, message):

    print(message)

    message.ack()

with Connection('amqp://test:test@localhost:5672//') as conn:

    with Consumer(conn, queues=[Queue('my-test', 

        exchange=Exchange('predictions', type='fanout', durable=False), durable=True, 

        queue_arguments={'x-queue-type':'stream'}, 

        consumer_arguments={'x-stream-offset': 0})], 

        prefetch_count=1000, callbacks=[callback]):

        for _ in eventloop(conn):

            pass

```

Can you post the stream status like:

```

rabbitmq-streams stream_status predictions

```

and also the logs to see if there are errors.

Cheers

-

Gabriele

[Raz Hemo]

Just to make sure, are you trying to reproduce the issue on a cluster? As I said, there is no problem with only one node. The problem occurs only on a cluster.

Anyway, I didn't see any logs on the server, and this is the output of stream_status:

Status of stream predictions on node rabbitmq@labbeast1 ...

┌─────────┬─────────────────────┬────────┬──────────────────┬──────────────┬─────────┬──────────┐

│ role    │ node                │ offset │ committed_offset │ first_offset │ readers │ segments │

├─────────┼─────────────────────┼────────┼──────────────────┼──────────────┼─────────┼──────────┤

│ writer  │ rabbitmq@beast29    │ 60541  │ 60541            │ 0            │ 2       │ 1        │

├─────────┼─────────────────────┼────────┼──────────────────┼──────────────┼─────────┼──────────┤

│ replica │ rabbitmq@labbeast1  │ 60541  │ 60541            │ 0            │ 0       │ 1        │

├─────────┼─────────────────────┼────────┼──────────────────┼──────────────┼─────────┼──────────┤

│ replica │ rabbitmq@labthanos1 │ 60541  │ 60541            │ 0            │ 1       │ 1        │

└─────────┴─────────────────────┴────────┴──────────────────┴──────────────┴─────────┴──────────┘

The reader I started this time got stuck after ~150k messages were in the stream. At this point the producer was paused. I restarted the producer and the consumer did not advance. Only after restarting the consumer it could read further along.

I will note that this doesnt reproduce 100% of the time, it sometimes takes some finicking and several restarts of both the producer and consumer.

[Raz Hemo]

Sorry, i copied the wrong output, this is the output of stream_status at the correct moment in time where the hang occurred:

Status of stream predictions on node rabbitmq@labbeast1 ...

┌─────────┬─────────────────────┬────────┬──────────────────┬──────────────┬─────────┬──────────┐

│ role    │ node                │ offset │ committed_offset │ first_offset │ readers │ segments │

├─────────┼─────────────────────┼────────┼──────────────────┼──────────────┼─────────┼──────────┤

│ writer  │ rabbitmq@beast29    │ 277050 │ 277050           │ 0            │ 2       │ 1        │

├─────────┼─────────────────────┼────────┼──────────────────┼──────────────┼─────────┼──────────┤

│ replica │ rabbitmq@labbeast1  │ 277050 │ 277050           │ 0            │ 1       │ 1        │

├─────────┼─────────────────────┼────────┼──────────────────┼──────────────┼─────────┼──────────┤

│ replica │ rabbitmq@labthanos1 │ 277050 │ 277050           │ 0            │ 0       │ 1        │

└─────────┴─────────────────────┴────────┴──────────────────┴──────────────┴─────────┴──────────┘

[Karl Nilsson]

This looks like a bug. Looking at the code I can see something that doesn't look quite right. If I create a branch with a few changes would you be able run your tests against the docker image artifact that gets created?

Cheers

Karl

--
You received this message because you are subscribed to the Google Groups "rabbitmq-users" group.
To unsubscribe from this group and stop receiving emails from it, send an email to rabbitmq-user...@googlegroups.com.
To view this discussion on the web, visit https://groups.google.com/d/msgid/rabbitmq-users/fb053907-77e5-4e16-b029-2e3143716ecfn%40googlegroups.com.

--

Karl Nilsson

[Raz Hemo]

yes, i'd be able to do that. Thanks :)

out of curiosity, i'd be happy to know what happened

[Karl Nilsson]

https://hub.docker.com/layers/pivotalrabbitmq/rabbitmq/stream-queue-consumer-fix-otp-max/images/sha256-59b33eb5ca476e57bea72962c4b4138b79499c2b7fc20a0b934a92371cf9010d?context=explore

I don't know for sure this will fix it but it was doing something internally that crossed rabbit nodes when it didn't need to. Theoretically that could possibly trigger some unexpected race conditions. Either way it is a good change as it avoid cross node communication for common usage patterns. 

To view this discussion on the web, visit https://groups.google.com/d/msgid/rabbitmq-users/759e1412-1b37-49e7-bf5b-61124fc5b7cen%40googlegroups.com.

--

Karl Nilsson

[Raz Hemo]

Hi,

So far I'm trying to reproduce with your image and it seems like it no longer hangs. It does sometimes give me a connection refused when initializing, though, but atleast that's a more easily manageable problem. :)

[Raz Hemo]

Im not sure if the connection refused is a problem with rabbitmq as it could be something on our end. Either way, i'm happy about the fix :)

[Karl Nilsson]

Great thanks for testing!

To view this discussion on the web, visit https://groups.google.com/d/msgid/rabbitmq-users/605a0927-d674-44cb-bbd3-48255eeee00an%40googlegroups.com.

--

Karl Nilsson

[Gabriele Santomaggio]

Raz,

Fixed here: https://github.com/rabbitmq/rabbitmq-server/pull/3550 
It will be in the next RabbitMQ release.

Cheers

-

Gabriele