consumer prefetch locked to 2000 by default even with prefetch_count=0

[pfribourg-ias]

Hi,

I have a simple issue:
with a single consumer on a queue, I want to have unlimited messages prefetch (unacked coming in).

To do so, in the documentation, it is said to call
channel.basic_qos(prefetch_size=0, prefetch_count=0, global_qos=True)
but it just don't work, and my consumer is locked to 2000 messages prefetched, no more.

I use pika v1.2.0 as shown here :

>>> import pika
>>> pika.__version__
'1.2.0'

Here is a sample producer that load 10 000 messages in a queue:

import pika

print("...")
queue_name = "my_queue"
credentials = pika.PlainCredentials('user', 'password')
connection_parameters = pika.ConnectionParameters(
    host='localhost', port='5673', credentials=credentials
)
connection = pika.BlockingConnection(connection_parameters)
channel = connection.channel()
channel.queue_declare(
    queue=queue_name,
    durable=True,
    arguments={"x-queue-type": "quorum"},
)
for _ in range(10000):
    channel.basic_publish(
        exchange='',
        routing_key=queue_name,
        body=b'Hello World !',
        properties=pika.BasicProperties(delivery_mode=2)
    )
connection.close()
print("done")

here is a sample consumer where I try to set unlimited prefetch count but it does not work and it lock at 2 000 messages:

import pika


def on_message(channel, method_frame, header_frame, body):
    print(method_frame.delivery_tag, body)


print("...")
queue_name = "my_queue"
credentials = pika.PlainCredentials('user', 'password')
connection_parameters = pika.ConnectionParameters(
    host='localhost', port='5673', credentials=credentials
)
connection = pika.BlockingConnection(connection_parameters)
channel = connection.channel()
channel.basic_consume(queue_name, on_message)
channel.basic_qos(prefetch_size=0, prefetch_count=0, global_qos=True)
try:
    channel.start_consuming()
except KeyboardInterrupt:
    channel.stop_consuming()
connection.close()
print("done")

here is a screen shot of RabbitMQ showing that only 2 000 message are prefetched:

I can raise prefetch_count up to 65 535 (the limit allowed) and I will be able to have 65 535 un-acked messages but I need unlimited un-acked messages.
I will take care of memory on the consumer side, I just want to have ALL message in, as a consumer.

The only way I found to have unlimited un-acked message is to manually call channel.basic_get but it is way slower than subscribe as a consumer.
How should I do to have a single consumer using basic_consume & have as much message as I want in prefetch ?

Thanks a lot.

[lukebakken]

You understand that you eventually MUST acknowledge message delivery, correct?

[pfribourg-ias]

Yes of course, but I have to wait for a very large number of message to aggregate them prior to acknowledge them.
Anyway, I still don't understand why setting prefetch_count to 0 don't do as it is said in the documentation, I may have setup something wrongly but I don't know where.

[lukebakken]

My guess is that there is a limit within RabbitMQ to prevent abuse from applications that set the limit to 0. I've been trying to find it in the code and will ask the team as well.

[lukebakken]

https://github.com/rabbitmq/rabbitmq-server/blob/master/deps/rabbit/src/rabbit_quorum_queue.erl#L743-L748

[kjnilsson]

Perhaps in this case using a stream would be better?

[pfribourg-ias]

thanks a lot @lukebakken for identifying the source of this !
So this mean it only apply to quorum queues I suppose ?
Well, I have to use quorum to be robust so I won't switch to classic anyway.
I suggest to point this out in the documentation if it's so specific to quorum.

@kjnilsson I am not familiar with stream but I will have a look thanks for pointing this out.