RabbitMQ for Queued Model Inference on Kubernetes

RabbitMQ has an acknowledgement timeout, it defaults to thirty minutes, and when a consumer exceeds it the broker does not just requeue that delivery — it closes the channel and requeues everything outstanding on it, from every consumer sharing that channel.

The operator, not a hand-written StatefulSet

Running RabbitMQ on Kubernetes by writing your own StatefulSet means owning peer discovery, the Erlang cookie, quorum membership during rolling updates and the ordering of a version upgrade. The RabbitMQ Cluster Operator owns all of that, and the reason to use it is not convenience but that those are the failure modes that lose messages.

Install the operator first, then describe the cluster you want as a custom resource. The operator reconciles it into a StatefulSet, Services, a ConfigMap and a Secret.

The RabbitmqCluster resource and its credentials

apiVersion: rabbitmq.com/v1beta1
kind: RabbitmqCluster
metadata:
  name: inference
spec:
  replicas: 3
  resources:
    requests:
      cpu: 1000m
      memory: 2Gi
    limits:
      cpu: 1000m
      memory: 2Gi

Three replicas rather than one, because a single-node broker on Kubernetes is a broker that loses your queue every time the node is drained. The operator creates a Service named after the instance, which exposes 5672 for AMQP, 15672 for the management UI and 15692 for Prometheus metrics.

Credentials come from a Secret named after the instance with a -default-user suffix — for the cluster above, that is inference-default-user — holding username and password keys. Mount that Secret into your consumer rather than copying the values into your own Secret, because the operator manages its lifecycle and a copy is a copy that goes stale.

Declare the queue itself as a quorum queue rather than a classic one, by setting the x-queue-type argument to quorum at declaration. Quorum queues replicate through a consensus protocol and are the type designed to survive node loss, which is the entire reason you asked for three replicas.

Prefetch and manual acknowledgement

Two consumer settings decide whether this works. Acknowledge manually, after the result is durably written, so a consumer that dies mid-call releases its delivery for redelivery instead of having already confirmed it. And set a small prefetch through basic.qos, because a consumer that has claimed a hundred slow deliveries has claimed a hundred deliveries nobody else can take.

import json, pika

params = pika.URLParameters(os.environ["AMQP_URL"])
conn = pika.BlockingConnection(params)
ch = conn.channel()

ch.queue_declare(queue="inference", durable=True,
                 arguments={"x-queue-type": "quorum"})
ch.basic_qos(prefetch_count=2)

def on_message(ch, method, properties, body):
    job = json.loads(body)
    try:
        write_result(job["job_id"], call_model(job["prompt"]))
        ch.basic_ack(method.delivery_tag)
    except PermanentError:
        ch.basic_nack(method.delivery_tag, requeue=False)   # to the DLX

ch.basic_consume(queue="inference", on_message_callback=on_message)
ch.start_consuming()

The requeue=False on basic_nack is what sends a permanently bad message to a dead-letter exchange instead of round-tripping it forever. Declare the queue with a x-dead-letter-exchange argument and a bound dead-letter queue, or that nack simply discards the message.

One caution specific to this client shape: a blocking consumer that spends ninety seconds inside its callback is not reading the socket, and RabbitMQ’s heartbeat can time out the connection while your model call is perfectly healthy. Either run the model call on a separate thread and acknowledge back on the connection’s thread, or use an asynchronous client. This is the same class of problem as the acknowledgement timeout below and people frequently diagnose one as the other.

consumer_timeout, and how it takes down a channel

RabbitMQ documents a delivery acknowledgement timeout with a default value of 30 minutes. If a consumer does not acknowledge within it, the node closes the channel with a PRECONDITION_FAILED channel exception and logs a message naming the consumer tag, the channel, the queue and the delivery tag, along with the timeout used.

The consequence is worse than a redelivery. RabbitMQ documents that all the following deliveries on that channel, from all consumers, are then requeued. One slow model call therefore returns every in-flight message on the same channel to the queue, and if you were using one channel for a pool of consumers you have just duplicated all of their work at once.

Two mitigations, and use both. Give each consumer its own channel, so the blast radius of a timeout is one delivery. And if a legitimate unit of work can genuinely exceed the timeout, raise consumer_timeout in the broker configuration — it is a node-level setting, not a per-queue one, so raising it affects everything on that cluster and is a decision to make once and write down.

If your consumer is Celery rather than a hand-written one, the same arithmetic applies with different names; the settings that matter there are in Celery and Redis for queued inference on Kubernetes.

Memory alarms and a publisher that stops silently

A queue in front of a model is a queue that gets deep. That is its purpose: the producer accepts requests faster than a provider will serve them, and the difference accumulates. RabbitMQ has an opinion about that accumulation, and the way it expresses the opinion catches people out.

RabbitMQ documents that by default a node will use about 60% of available RAM, and that when it passes that threshold it raises a memory alarm and blocks all connections that are publishing messages. Blocks, not rejects. Your producer does not get an error it can log; it gets a publish that never returns. A synchronous intake handler in that state stops responding, its own timeouts fire upstream, and the reported symptom is “the API is down” with nothing wrong in the API.

Three things reduce the chance of getting there. Keep message bodies small, for the same reason as everywhere else in this cluster: publish a reference to the document rather than the document. Give the pod a memory request and limit that are consistent with the watermark, because a node calculating 60% of a container’s limit behaves very differently from one calculating 60% of the host’s RAM, and the operator needs to know which it is looking at. And bound the queue itself — declare it with a maximum length and an overflow behaviour, so a runaway producer sheds load or dead-letters rather than filling the broker.

The AMQP protocol does expose this state: a client can subscribe to connection-blocked notifications and react rather than hang. Almost no application does, which is why this reads as a mystery outage the first time. If you write nothing else defensive, at least set a publish timeout on the producer so a blocked connection surfaces as an error you can see instead of a request that never finishes.

Building it

1. Install the RabbitMQ Cluster Operator into the cluster, then apply a RabbitmqCluster with three replicas and explicit resource requests.
2. Wait for the Secret with the -default-user suffix to appear and reference it from your consumer Deployment with secretKeyRef. Do not copy the values.
3. Declare the work queue as durable with x-queue-type: quorum and an x-dead-letter-exchange pointing at a bound dead-letter queue.
4. Write the consumer with manual acks, a prefetch of 1 or 2, and one channel per consumer.
5. Decide about consumer_timeout deliberately: either keep the 30-minute default and guarantee your work fits, or raise it in the broker config and record why.
6. Delete a broker pod while work is in flight and confirm the consumer reconnects and no result row is missing. That is the test the whole three-replica setup exists to pass.