The tax service was down for twenty minutes and is back. orders.dlq holds 230 messages: 41 of them failed with TaxServiceUnavailableException, the rest are missing orders and declined payments that a replay would only send round again. RabbitMQ has no "move the messages where header X equals Y": a queue hands out messages from the head, one after the other, and the management UI's "Move messages" and a shovel take all of them. Here is how to take only the 41, measured on RabbitMQ 4.3.6 with classic and quorum queues.
The obvious script reads a message, checks it, and puts the ones it does not want straight back:
method, props, body = ch.basic_get("orders.dlq", auto_ack=False)
if not wanted(props):
ch.basic_nack(method.delivery_tag, requeue=True) # do not do thisA message requeued this way goes back to the head of the queue, where it was. The next basic_get returns the same message again, and the script spins on the first unwanted message forever. Classic and quorum queues behave the same.
The way through the queue is to keep the skipped messages unacknowledged while you read on. As long as a delivery is outstanding on the channel, basic_get hands out the next message. At the end one basic_nack gives all of them back:
import sys, pika
QUEUE = "orders.dlq"
DEATH = ("x-death", "x-first-death-", "x-last-death-", "x-delivery-count")
def wanted(method, headers):
# your filter; Spring's RepublishMessageRecoverer writes x-exception-*
exc = headers.get("x-exception-type", "")
return exc.endswith("TaxServiceUnavailableException")
def route(method, headers):
if headers.get("x-death"): # dead-lettered by the broker
death = headers["x-death"][0]
return death["exchange"], death["routing-keys"][0]
# republished by Spring's recoverer
return headers["x-original-exchange"], headers["x-original-routingKey"]
conn = pika.BlockingConnection(pika.URLParameters(sys.argv[1]))
ch = conn.channel()
ch.confirm_delivery() # 1.
total = ch.queue_declare(QUEUE, passive=True).method.message_count # 2.
moved = 0
try:
for _ in range(total):
method, props, body = ch.basic_get(QUEUE, auto_ack=False) # 3.
if method is None:
break
headers = dict(props.headers or {})
if not wanted(method, headers):
continue # 4. hold it, unacked
exchange, routing_key = route(method, headers)
props.headers = {k: v for k, v in headers.items()
if not k.startswith(DEATH)}
ch.basic_publish(exchange, routing_key, body, props,
mandatory=True) # raises if not routed
ch.basic_ack(method.delivery_tag) # only after the confirm
moved += 1
finally:
ch.basic_nack(0, multiple=True, requeue=True) # 5. hand the rest back
conn.close()
print(f"moved {moved} of {total}")We ran it against a dead-letter queue with nine messages: six dead-lettered by the broker, three republished by a Spring-style recoverer without x-death. It moved the three that matched to the work queue, without death headers, and left the other six in the dead-letter queue in their original order. Same result on a classic and a quorum queue. With a filter that raised an exception on the fifth message, the one message already moved stayed moved and the other eight were back in the dead-letter queue, their delivery count unchanged.
What each numbered comment is for:
confirm_delivery(), pika's basic_publish waits for the broker and raises if the message was not taken; with mandatory=True it also raises when no queue is bound for the routing key. Ack the original only after that.auto_ack=True a skipped message is gone the moment you look at it.basic_nack(0, multiple=True) returns every delivery still outstanding on the channel, in the finally block so that an exception in your filter or a failed publish hands the messages back the same way. The skipped messages go back to the head of the queue in their old order.Afterwards every message that was skipped is marked redelivered. That is harmless in a dead-letter queue, but worth knowing if a consumer treats the flag as suspicious.
Closing the channel also returns every unacknowledged message, and many scripts rely on it. On a classic queue that is fine. On a quorum queue it is not the same thing as a nack:
| How the skipped messages go back (quorum queue, 4.3.6) | x-delivery-count | 29 scans of one message, default delivery limit |
|---|---|---|
basic_nack(requeue=True) | unchanged | message still there |
| connection closed with the message unacked | +1 per close | message gone at scan 22 |
RabbitMQ 4 gives every quorum queue a delivery limit of 20. A return by channel close counts as a failed delivery; an explicit nack does not. Twenty-one scans that end with conn.close(), or twenty-one crashes of the script, and the message is dropped, or dead-lettered further if the dead-letter queue has a dead-letter exchange of its own. On a quorum dead-letter queue that is fed by a quorum queue the margin is smaller still, because the message arrives with the deliveries it already used up: in our test a single reject on the work queue brought it in with x-delivery-count 1, and the quorum DLQ article shows a poison message arriving with 21. So end every scan with the nack, in a finally block as above, and never leave the messages to the channel close.
What you can filter on depends on who put the message into the dead-letter queue:
x-death[0] has reason, queue, exchange and routing-keys. The exception is not there, because the broker never saw it. Your own publish headers are kept, so a header like x-tenant or x-order-type works.RepublishMessageRecoverer, MassTransit's _error queues): the exception sits in the headers, x-exception-type and x-exception-message for Spring, MT-Fault-ExceptionType for MassTransit, and there is no x-death. Spring keeps the route in x-original-exchange and x-original-routingKey.Exception messages usually contain ids ("Order 4711 not found"). Compare the exception type, or the message with digits replaced, not the full text.
If the same question comes up every week, let the broker sort the dead letters as they arrive. A dead-lettered message keeps its routing key unless the queue sets x-dead-letter-routing-key, so a topic exchange as dead-letter exchange can send each kind to its own queue:
ch.exchange_declare("orders.dlx", "topic", durable=True)
ch.queue_declare("orders.dlq.created", durable=True)
ch.queue_bind("orders.dlq.created", "orders.dlx", "orders.created")
ch.queue_declare("orders.dlq.cancelled", durable=True)
ch.queue_bind("orders.dlq.cancelled", "orders.dlx", "orders.cancelled")
ch.queue_declare("orders.work", durable=True, arguments={"x-dead-letter-exchange": "orders.dlx"})Then "replay the created orders" is "replay one queue", and "Move messages" or a shovel can do it. Two cautions from the test: a catch-all binding with # next to the specific ones receives a copy of every message, so the two queues overlap; and a routing key the bindings do not cover is dropped silently. Bind every key you use, or put an alternate exchange on the dead-letter exchange. Splitting works for routing keys and your own headers (a headers exchange), not for the exception, which the broker does not know.
finally block, not with a connection close, above all on quorum queues.mandatory=True on the publish, ack after the confirm.x-death for broker dead letters, exception headers for republished ones.