RabbitMQ Quorum Queues Explained: The New Default Queue vs Classic Queue Type

This queue type is important when RabbitMQ is used in a clustered installation. Let’s explore…

Quorum Queues and Replication: Replacing the Classic Queue Type in RabbitMQ

In RabbitMQ 3.8.0, one of the most significant new features was the introduction of Quorum Queues. Quorum queue is a modern queue type that is expected to replace the default queue (now called Classic) in the future, for some use cases. The Quorum Queue is important when RabbitMQ is used in a clustered installation, where it provides less network-intensive message replication using the Raft consensus algorithm underneath. This approach significantly reduces the risk of split-brain and inconsistencies when compared to the older Guaranteed Multicast protocol.

Usage of Quorum Queues

A Classic Queue has a master running somewhere on a node in the RabbitMQ cluster, while the mirrors run on other nodes. Quorum Queues work in much the same way, with the leader—by default—running on the node that the client application was connected to when it created the queue, and followers running on the other nodes in the cluster.

In the past, replication of queues was specified by using policies in conjunction with Classic Queues. Quorum queues are created differently, but should be compatible with all client applications which allow you to provide arguments when declaring a queue. The x-queue-type argument needs to be provided with the value quorum when creating the queue.

For example, using the Elixir AMQP client1, declaring a Quorum Queue is as follows:
Queue.declare(publisher_chan, "my-quorum-queue", durable: true, arguments: [ "x-queue-type": "quorum" ])

An important difference between Classic and Quorum Queues is that Quorum Queues can only be declared durable; otherwise, the following error message is raised:
:server_initiated_close, 406, "PRECONDITION_FAILED - invalid property 'non-durable' for queue 'my-quorum-queue'

After declaring the queue, we can observe that it is indeed a quorum type on the Management Interface:

We can see that a Quorum queue has a Leader, which roughly serves the same purpose as it did for the Classic Queue Master. All communication is routed to the Queue Leader, which means the queue leader locality has an effect on the latency and bandwidth requirement of the messages; however, the effect should be lower than it was in Classic Queues.

Consuming from a Quorum Queue is done in the same fashion as other types of queues.

New for Quorum Queues

Quorum queues come with a few special features and restrictions. They cannot be non-durable, because the Raft log entries are always written to disk, so they can never be declared transient. 
As the use case for Quorum Queues is fault tolerance and data safety, they also cannot be declared as exclusive, which would mean they get deleted as soon as the consumer disconnects.
Since all messages in Quorum Queues are persistent, the AMQP “delivery-mode” option has no effect on their operation. Queues persist their state even after broker restarts, which aligns with their reliability goals.

Single Active Consumer

This is not unique to Quorum Queues, but it’s still important to mention that, even though the Exclusive Queue feature was lost, we gain a new feature that is even better in many ways and was a frequently requested enhancement.

Single Active Consumer enables you to attach multiple consumers to a queue, while only one of the consumers is active. This lets you create highly available consumers while ensuring, at any given moment, only one of them receives messages, which until now was not possible to attain with RabbitMQ.

An example of how to declare a queue with the Single Active Consumer feature in Elixir:

Queue.declare(publisher_chan,
             "single-active-queue",
                            durable: true,
                              arguments: [ "x-queue-type": "quorum", "x-single-active-consumer": true ])

The queue with the Single Active Consumer setting enabled is marked as SAC. In the image above, we can see that two consumers are attached to it (two channels executed Basic.consume on the queue). When publishing to the queue, only one of the consumers will receive the message. When that consumer disconnects, the other one should take exclusive ownership of the stream of messages.

Basic.get, or inspecting the message on the Management Interface, cannot be done with Single Active Consumer queues.

Tracking Retries and Poison Messages in Message Streams

Keeping a count of how many times a message was rejected is another much-requested feature for RabbitMQ, which has finally arrived with Quorum Queues. This lets you handle redelivery and poison messages more effectively than before, as previous implementations often suffered from the inability to give up retrying in the case of a stuck message, or had to keep track of how many times a message was delivered in an external database.

NOTE: For Quorum Queues, it is best practice to always have some limit on the number of times a message can be rejected. Letting this message reject count grow forever can lead to erroneous queue behaviour due to the Raft implementation.

Using Classic Queues, when a message was requeued for any reason, with the redelivered flag being set, what this flag essentially means is, “the message may have been processed already”. This helps you to check if the message is a duplicate or not. The same flag exists; however, it has been extended with the x-delivery-count header, which tracks the frequency of requeueing.

We can observe this header on the Management Interface:

As we can see, the redelivered flag is set, and the x-delivery-count header is 2.

Now your application is better equipped to decide when to give up retrying.

If that is not good enough, you can now define rules based on the delivery count to send the message to a different Exchange instead of requeuing. This can be done right from RabbitMQ; your application does not have to know about the retrying. Let me illustrate this with an example!

Example: Re-routing rejected messages! Our use case is that we receive messages which we need to process, from an application which, however, may send us messages which cannot be processed. The reason could either be because the messages are malformed, or because the application itself cannot process them for some reason or another, but we do not have a way to notify the Sending Application of these errors. These errors are common when RabbitMQ serves as a message bus in the system, and the Sending Application is not under the control of the Receiving Application Team.

We then declare a queue for the messages which we could not process:

And we also declare a fanout exchange, which we will use as a Dead Letter Exchange:

And bind the unprocessable-messages queue to it.

We create the application queue called my-app-queue and the corresponding policy:

We can use either Basic.reject or Basic.nack to reject the message. We must use the requeue property set to true.
Here’s a simplified example in Elixir:

def get_delivery_count(headers) do
   case headers do
       :undefined -> 0
       headers ->
           { _ , _, delivery_cnt } = List.keyfind(headers, "x-delivery-count", 0, {:_, :_, 0} )
           delivery_cnt
   end
end

receive do
   {:basic_deliver, msg, %{ delivery_tag: tag, headers: headers} = meta } ->
       delivery_count = get_delivery_count(headers)
       Logger.info("Received message: '#{msg}' delivered: #{delivery_count} times")
       case msg do
           "reject me" -> Logger.info("Rejected message")
                          :ok = Basic.reject(consumer_chan, tag)
           _ ->
                          Logger.info("Acked message")
                          :ok = Basic.ack(consumer_chan, tag)
        end
end

First we publish the message, “this is a good message”:
13:10:15.717 [info] Received message: 'this is a good message' delivered: 0 times
13:10:15.717 [info] Acked message

Then we publish a message which we reject:
13:10:20.423 [info] Received message: 'reject me' delivered: 0 times
13:10:20.423 [info] Rejected message
13:10:20.447 [info] Received message: 'reject me' delivered: 1 times
13:10:20.447 [info] Rejected message
13:10:20.470 [info] Received message: 'reject me' delivered: 2 times
13:10:20.470 [info] Rejected message

And after it’s delivered three times, it is routed to the unprocessed-messages queue.
We can see on the Management Interface that the message is routed to the queue:

Controlling the members of the quorum

Quorum queues do not automatically change the group of followers/leaders. This means that adding a new node to the cluster will not automatically ensure that the new node is being used for hosting quorum queues. Classic Queues in previous versions handled adding queues on new cluster nodes through the policy interface. However, this could be problematic as cluster sizes were scaled up or down. An important new feature in the 3.8.x series for Quorum Queues and Classic Queues is the in-built queue master rebalancing, which improves throughput and resource utilisation across nodes. Previously, this was only attainable using external scripts and plugins.

Adding a new member to the quorum can be achieved using the grow command:

rabbitmq-queues grow rabbit@$NEW_HOST all
Removing a now stale, for example, deleted, host from the members can be done through the shrink command:
rabbitmq-queues shrink rabbit@$OLD_HOST
We can also rebalance the queue masters so that the load is equal on the nodes:
rabbitmq-queues rebalance all
Which (in bash) will display a nice table with statistics on the number of masters on nodes. On Windows, use the --formatter json flag to get a readable output.
.

Summary

RabbitMQ 3.8.x comes with a lot of new features. Quorum Queues are just one of them. They provide a more understandable, in some cases less resource-intensive, new implementation for achieving replicated queues and high availability. They are based on the Raft consensus algorithm and support features not found in Classic Queues, which fundamentally are based on the custom Guaranteed Multicast protocol3 (a variant of Paxos). As this type and class of queues are still quite new, only time will tell if they become the more used and preferred queue type for the majority of distributed installations of RabbitMQ in comparison to their counterparts, the Classic Mirrored Queues. Until then, use both as best fitting for your RabbitMQ needs! 🙂

Need help with your RabbitMQ?

Our world-leading RabbitMQ team offers a variety of RabbitMQ service options to suit your needs. We have everything, from health checks to RabbitMQ support and monitoring, to help you ensure an efficient and reliable RabbitMQ system. Get in touch here.