AMQP supports setting quality of service:
https://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.qos
This limits the number (or size) of unacknowledged messages sent to a client.
By default, not setting qos implies that the messages sent are unlimited, meaning that the messages will be sent to the client(s) as quickly as possible. This can cause a large number of messages to queue up on the client(s), which is undesirable because those messages cannot be processed by another server.
Suppose you have a topic with one consumer, and you send hundreds or thousands of messages to the topic. The broker will send the messages to the consumer as fast as the transport will allow. Now suppose the messages take a long time to process on the consumer... maybe each one takes 30 seconds. Queueing up 100 messages means that the consumer has committed to 3000 seconds of work. It would be nice to recognize that such a backlog of work is present, and to bring up additional consumers on the topic to process the work. But because the messages are already consumed by the original consumer, the new consumers cannot help with the original load of messages.
For more details, check out https://www.rabbitmq.com/blog/2012/05/11/some-queuing-theory-throughput-latency-and-bandwidth/
I propose a two step approach to fixing this in oslo.messaging:
(1) Add configuration options to manually define qos count/size, and simply call the basic.qos method when new channels are created.
(2) Implement a "controlled delay" algorithm in the client in order to adjust the buffer size dynamically, similar to the java client (https://gist.github.com/msackman/2658712) as described in the above article. This belongs lower in the stack (py-amqp/kombu/etc) but would ultimately need to be enabled/utilized within oslo.messaging.
Fix proposed to branch: master
Review: https:/