Publisher confirmation tracking and exceptions

[lukebakken]

Please see the start of the discussion here:

#1687 (comment)

[lukebakken]

@bpolitaiev

My point is that Nack is not unexpected scenario, as overflow | reject-publish is the only way to wait when queue is full.

Sure, but the vast majority of RabbitMQ users do not use queues that way ("wait until queue is full").

I really do not want to force users to have to check if BasicPublishAsync succeeded on every call.

@bording @danielmarbach any thoughts?

[michaelklishin]

Publisher-side rate limiting is a common approach in distributed systems. All kinds of client-side circuit breakers are. That's what TCP does, using a very basic set of metrics sent by the peer.

And it will side step basic.nacks entirely.

[mllab-nl]

@michaelklishin Here the tricky part is - you do not want the rate limiter to limit to much and having idle consumers.
In essence the publisher becomes an artificially created limit on the throughput of the system.

I would argue that if your consumers are scaled dynamically based on the number of elements in the Q and Message rates and you are not getting nacks because of Publisher rate limiting - you system is not running in its max throughput

[michaelklishin]

@mllab-nl welcome to distributed systems. Something somewhere has to give. Why is it always RabbitMQ and its client libraries and not apps that have adopted suboptimal strategies, is a mystery I have been trying to understand since 2010 or so.

[michaelklishin]

you are not getting nacks because of Publisher rate limiting

99.9% of RabbitMQ users do not get basic.nacks because they do not use queue length limit as a throttling mechanism. It was never designed to be a throttling mechanism, it's nothing more than a "pressure relief valve."

Most clients will never outpace a stream, or at least a superstream with 5 or 7 nodes, which was recommended below. Most systems naturally will not exceed sustained or even peak publishing rate of 5 million messages per second a superstream can deliver in a 5 or 7 node cluster.

The entire problem here is self-inflicted and specific to one particular case where a safety feature is used as a rate limiting mechanism, and then it is RabbitMQ (client libraries) that have to change. No, it does not. Your team has made a decision to use basic.nack and queue length limit as a flow control mechanism, it's on you to face the downsides of such a design.

[michaelklishin]

One obvious piece of evidence of this is the fact that people ask "how do I trigger a basic.nack relatively often. They need it to test publisher confirm handlers, not because they use overflow set to reject-publish.

After suggesting several alternative designs, I'm going to lock this discussion. It no longer serves a purpose and it is tiring to hear that RabbitMQ must bend over backwards to accommodate app design choices that may be reasonable but clearly have downsides.

1. reject-publish-dlx with a dead lettering "loop" can be used to accumulate the "overflow" and have "additional capacity" consumers
2. Using more queues, more cores for those queues, faster disks and "reasonably evenly distributing" exchange types such as random, consistent hashing and x-module exchanges (a part of rabbitmq_sharding) will help increase the degree of parallelism and avoid large accumulated "overflows" and thus basic.nacks
3. Streams and superstreams can be an option at least for some of these cases, the biggest problem there is partitioning the workload across consumers

[michaelklishin]

Those who "wait till the queue is full" should probably use RabbitMQ Streams with a respective stream client. Offset management will be much easier to pull off there.

[bpolitaiev]

Just like with queues, there's nothing wrong with declaring N streams and using stream filtering. Consuming a batch from a stream, however, is trivial and may require a minimal amount of polling but that's because your consumers were designed this way.

I have a feeling that we do not understand each other. Each message is a unit of work, it can be of any size (can take 100ms or 10 minutes to handle) and I have 1000s of workers crunching huge backlog of work. This work is naturally distributed and no worker is idle. I can't see how can this be achieved with N streams and filtering.

[michaelklishin]

@bpolitaiev if every message is a "unit of work" then you don't need polling to determine that N messages are available. The standard channel QoS prefetch and manual acknowledgements will be sufficient.

Error communication using channel errors has been a part of AMQP since 2006 and is not going to change.

With streams and superstreams, retries, repeatable non-destructive reads is a core protocol and data structure feature. Partitioning the work between consumers is the only part that's up to you.

[bpolitaiev]

@michaelklishin I do not need to determine message availability, I simply need to distribute work across 1000s of workers. As simple as that. In order to make sure workers never idle - I have to maintain full queue (limit is a reasonable 2000 number).
So 1000s of consumers are crunching the work 1 by 1, and single Producer is non-stop keeping queue full. To keep the queue full I get thousands of Nack exceptions. I would love to avoid having them as it does seem excessive.

[michaelklishin]

You only get basic.nacks because you use overflow-reject. Some combinations of features work together better than others. We will not delay 7.0 release any further over this, and I do not recall others bringing up such a workload.

Perhaps drop overflow-reject and overprovision consumers instead, which will make basic.nacks as a "flow control" message a non-issue.

[michaelklishin]

In other words, there are those who use overflow-reject with relatively low rates, and those who need to move data to N consumers as quickly as possible who do not use overflow-reject, and therefore do not experience exceptions-as-flow-control.

An absolute majority of queues do not use overflow-reject, so one way or another, that setting is not essential and arguably should not be used if overflowing is very common, or at least the limit should be much higher.

[michaelklishin]

I am not interested in continuing this part of the discussion. You have chosen to use a feature that generates basic.nack by design, with a queue (or a set of queues) with higher rates and using a limit that is likely too low (thus a lot of messages are rejected).

And now you are asking us to change a client library to soothe this self-inflicted pain. I'm afraid the answer is "your should reconsider using overflow-reject for such queues.

[bpolitaiev]

@michaelklishin Thank you for your time! I agree with your argument about usage of 'overflow-reject'. We have a case of hyper fast work distribution which may really be a rare case.

[michaelklishin]

Even then, with a certain number of 4.x quorum queues tuned a bit I don't see why anyone would need overflow-publish, which was introduced around 2017-2018 because one large customer really wanted to reproduce something they were doing with another messaging tool, whether that made much sense or not.

[mllab-nl]

@michaelklishin @bpolitaiev @lukebakken we have the same setup:
Having HUGE amount of units of work (millions) to do.
Each unit might represent considerable size (up to 10 mb)
Each unit should be processed reliably (thus confirming after processing, re-processed on infra or code failure)
Processing time of each unit can vary significantly.
To be processed by a dynamic set of consumers (from 0 to thousands)

Based on our understanding and documentation we have considered NOT to publish the entire backlog to RabbiMQ Q
Large backlogs
Most RabbitMQ queues are designed to converge towards the empty state and are optimised as such and can perform worse when there are millions of messages on a given queue.

Thus only a limited portions of the backlog is published.
To archive that we also utilize overflow-reject and also observe a lot of exceptions in metrics.

Guys If you could spare a moment and consider a few questions below, that will be hugely beneficial to all the readers!

1. Is docs/our understanding correct that publishing a huge backlog to a Q is NOT a good Idea?
2. If 1 is true how would you suggest to do that ? We stopped at using overflow-reject. You also mentioned overprovision consumers instead. Could you please explain what you mean ? Is this applicable to the usecase discussed here

Thank you all for this nice discussion !

For the overflow-reject case my understanding is an informed flow control would be the best solution.
Now the flow control opens and closes the floodgate. When it is open much more messages flood in than allowed and they are being nacked.
IMO the proper would be : Is the floodgate is closed and after the is some available capacity in the Q the broker should inform the publishers that it will accept X without nacking - thus slightly opened floodgate. This will reduce the total overhead of Opening/closing the floodgate and nacking a lot of messages. The floodgate should be only fully opened when considerable capacity is available ( > 20-30 % .. ). Would be nice to know if there are some developments in this area. I have not found any.

@danielmarbach are you referring to a client side implementation to dynamically adjust the flow and strive for maintaining some Q limits (count messages/ total size)
#1713 (reply in thread)
Having such an option will be definitely beneficial and could avoid getting to a higher set overflow-reject frequently

[michaelklishin]

A queue.declare with passive set to true is the AMQP 0-9-1 mechanism of obtaining the number of messages in Ready state.

[michaelklishin]

I don't see how

To be processed by a dynamic set of consumers (from 0 to thousands)

Processing time of each unit can vary significantly

Would work any differently for RabbitMQ Stream protocol clients. Streams do not expect that a delivery can take a certain amount of time to process. How you manage your consumers is up to you, a stream can have thousands of consumers although most users have way fewer.

A superstream specifically sounds even better positioned for a large number of monomorphic (effectively identical) consumers. Partitioning the workload between them will be the key part.

This discussion has gone completely off topic.

[michaelklishin]

After chatting with the author of the above blog post @acogoluegnes, we have concluded that the complexity of work distribution and consumer management can be comparable to handling basic.nacks, so a classic queue v2 or a QQ in 4.x will be acceptable options (possibly with turning for ingress), and that getting rid of overflow-publish and testing of publishing surges may be enough.

[mllab-nl]

Thank you for your suggestions guys !

In our case a QQ in 4.x is used only to distribute work(publish with confirms) to a lot of consumers with Consumer Acknowledgements and tiny prefetch of 1.
The thing is - there is a huge amount of work and, for resource efficiency, you only need SOME of it in the Q at any moment in time.
What we are looking is to understand how to have only SOME in the Q (Preferably implemented by the broker)
This leads you to maxlength
This leads to reject-publish and this leads to initial question about nacks and exceptions

Do I understand correctly that you suggest to remove the Q limits (reject-publish) and implement Q limits client side in the producer ?

[michaelklishin]

reject-publish-dlx can be used to collect "the excess" into one or more queues (e.g. using the consistent hashing exchange or the random exchange) that are then processed by a separate group of identical consumers.