How global and consumer-specific QoS updates correspond

[mvrsss]

Describe the bug

Hi,

While updating the QoS of a consumer after channel creation, I’ve noticed that the prefetch count of a consumer is not correctly set.

Initially, the channel was created with a QoS of a value 1.

At this point, new QoS with a prefetch of 5 is sent to a broker.

Reproduction steps

1. Create a consumer with a prefetch count of 1.
2. Update prefetch count of a channel to 5.
3. Observe that prefetch of the channel is set to 5, but the prefetch of the consumer is still 1. Also, the number of unacked messages in the channel is 1.

Expected behavior

1. The prefetch count of 5 should have been updated to a value of 5.
2. The number of unacked messages may have been increased to a value up to 5, as QoS has increased, more messages could have been delivered.

Additional context

I also have tested the behaviour of a channel with a BasicQoS having a global=true value, which resulted in the expected behaviour. However, I acknowledge that this feature will be deprecated. https://blog.rabbitmq.com/posts/2021/08/4.0-deprecation-announcements/#removal-of-global-qos

I would like to update the prefetch count of an active consumer. What steps I should follow to reflect the changes on a consumer?

As a note, I found this in the AMQP protocol documentation https://www.rabbitmq.com/resources/specs/amqp0-9-1.extended.xml,

<method name="qos-ok" synchronous="1" index="11" label="confirm the requested qos">
<doc> This method tells the client that the requested QoS levels could be handled by the server. The requested QoS applies to all active consumers until a new QoS is defined. </doc>
<chassis name="client" implement="MUST"/>
</method>

[michaelklishin]

RabbitMQ deviates from the AMQP 0-9-1 spec, this has been documented for years.

Global updates to QoS cause a lot more trouble than they are worth, hence the deprecation.
Dynamically controlled prefetch in general creates more confusion and unexpected behavior than it helps, well, anything.

[adamncasey]

Linking this to a discussion on-going in the celery community: celery/celery#6067 (comment)

Celery uses dynamically updating prefetch values to implement it's delayed message / ETA task types - my understanding is that if a message is delivered to a consumer which shouldn't be executed for a while it is held and the prefetch count is bumped by one to accommodate it being held.

The original use case for this issue being created was to support developers of consumers not knowing/caring up front about exact details of expected message throughput and allowing some kind of 'sensible default' which could change as throughput changes. Statically configured prefetch values do require some quite careful up front consideration which isn't trivial in some cases.

With these two use cases it does feel to me like updating existing consumer prefetch values would be a useful RabbitMQ feature. Any thoughts @michaelklishin ?

[kjnilsson]

The credit mechanism in AMQP (1.0) is effectively this. It allows clients to dynamically adjust (up and down) the amount of credit provided and thus can get fairly fine grained control over the flow of messages between the broker and the client.

It may be possible to allow the legacy protocol to update the prefetch dynamically but for now cancelling and reconsuming is already supported and the next consumer should pick up the updated QoS.

[michaelklishin]

There are well known ways of doing delayed delivery without dynamically updated prefetch:

• NServiceBus docs
• Jack Vanlightly's explanation of what exactly NServiceBus does

[ansd]

This feature is implemented in RabbitMQ 4.0 by AMQP 1.0 link flow control.

[michaelklishin]

A related discussion #11955.