Circuit Breaker for Message Queues — Resilient Message Processing with Backpressure

In this tutorial, you will learn about Circuit Breaker for Message Queues. We cover key concepts, practical examples, and best practices to help you master this topic.

Message queue circuit breakers protect consumers from downstream service failures by stopping message consumption, routing messages to dead letter queues, handling poison messages, and providing backpressure to prevent consumer overload during downstream degradation.

flowchart LR
    Q[(Message Queue)] --> CB{Consumer CB}
    CB -->|Closed| Consumer[Process Message]
    CB -->|Open| DLQ[(Dead Letter Queue)]
    Consumer -->|Success| Ack[Acknowledge]
    Consumer -->|Fail| Count[Count Failure]
    Count -->|Threshold| Open[Open Circuit]
    Open -->|Recovery| Half[Half-Open]
    Half -->|Probe OK| Close[Close Circuit]

What You'll Learn

• Consumer-side circuit breaker patterns
• Dead letter queue fallback routing
• Poison message handling
• Consumer health monitoring
• RabbitMQ and Kafka circuit breaking

Why It Matters

Message consumers that fail to process messages (due to downstream service outage) can cause infinite retry loops, consumer crashes, and message pile-up. Circuit breakers stop consumption, route messages to DLQs, and protect both the consumer and downstream services.

Real-World Use

DodaTech's RabbitMQ consumers use circuit breakers per downstream service. When the email service fails, the consumer circuit opens after 5 consecutive failures. Messages route to a dead letter queue for later reprocessing. The consumer restarts health checks every 30 seconds.

RabbitMQ Consumer Circuit Breaker

import pika
import time
import json

class ConsumerCircuitBreaker:
    def __init__(self, consumer, fail_threshold=5, recovery_timeout=30):
        self.consumer = consumer
        self.fail_threshold = fail_threshold
        self.recovery_timeout = recovery_timeout
        self.failures = 0
        self.state = 'CLOSED'
        self.last_failure = 0

    def process_message(self, channel, method, properties, body):
        if self.state == 'OPEN':
            if time.time() - self.last_failure > self.recovery_timeout:
                self.state = 'HALF_OPEN'
                print("[CB] Half-open probe")
            else:
                print("[CB] Circuit open, routing to DLQ")
                channel.basic_reject(delivery_tag=method.delivery_tag, requeue=False)
                return

        try:
            self.consumer(body)
            channel.basic_ack(delivery_tag=method.delivery_tag)
            self.failures = 0
            if self.state == 'HALF_OPEN':
                self.state = 'CLOSED'
                print("[CB] Recovered, circuit closed")
        except Exception as e:
            self.failures += 1
            self.last_failure = time.time()
            if self.failures >= self.fail_threshold:
                self.state = 'OPEN'
                print(f"[CB] Circuit open ({self.failures} failures)")
            channel.basic_reject(delivery_tag=method.delivery_tag, requeue=False)
            print(f"[CB] Rejected to DLQ: {e}")

def process_email(message):
    data = json.loads(message)
    print(f"Sending email to {data['to']}")
    if time.time() % 3 == 0:
        raise ConnectionError("Email service unavailable")

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='email_tasks', arguments={
    'x-dead-letter-exchange': 'dlx',
    'x-dead-letter-routing-key': 'email_failed',
})
channel.queue_declare(queue='email_failed')

cb = ConsumerCircuitBreaker(process_email)
channel.basic_consume(queue='email_tasks',
    on_message_callback=cb.process_message)

print("Consumer starting with circuit breaker")

Expected output:

Consumer starting with circuit breaker
Sending email to user@example.com
Sending email to user2@example.com
[CB] Circuit open (5 failures)
[CB] Circuit open, routing to DLQ
[CB] Rejected to DLQ: Email service unavailable

Kafka Consumer Circuit Breaker

from kafka import KafkaConsumer, KafkaProducer
import json
import time

class KafkaConsumerCircuitBreaker:
    def __init__(self, consumer, dlq_producer, dlq_topic,
                 fail_threshold=5, recovery_timeout=30):
        self.consumer = consumer
        self.dlq_producer = dlq_producer
        self.dlq_topic = dlq_topic
        self.fail_threshold = fail_threshold
        self.recovery_timeout = recovery_timeout
        self.failures = 0
        self.state = 'CLOSED'
        self.last_failure = 0

    def poll(self):
        messages = self.consumer.poll(timeout_ms=1000)
        for tp, records in messages.items():
            for record in records:
                self._process(record)

    def _process(self, record):
        if self.state == 'OPEN':
            if time.time() - self.last_failure > self.recovery_timeout:
                self.state = 'HALF_OPEN'
            else:
                self._send_to_dlq(record)
                return

        try:
            self._handle_message(record.value)
            self.failures = 0
            if self.state == 'HALF_OPEN':
                self.state = 'CLOSED'
                print("[Kafka CB] Recovered")
        except Exception as e:
            self.failures += 1
            self.last_failure = time.time()
            if self.failures >= self.fail_threshold:
                self.state = 'OPEN'
                print(f"[Kafka CB] Open ({self.failures} failures)")
            self._send_to_dlq(record)

    def _send_to_dlq(self, record):
        self.dlq_producer.send(self.dlq_topic, value=record.value)
        print(f"[Kafka CB] Sent to DLQ: {record.value}")

    def _handle_message(self, value):
        data = json.loads(value)
        print(f"Processing: {data['id']}")
        if data.get('fail'):
            raise ValueError("Processing failed")

consumer = KafkaConsumer('orders', bootstrap_servers='localhost:9092')
producer = KafkaProducer(bootstrap_servers='localhost:9092')
cb = KafkaConsumerCircuitBreaker(consumer, producer, 'orders_dlq')

Expected output:

Processing: order-1
Processing: order-2
[Kafka CB] Open (5 failures)
[Kafka CB] Sent to DLQ: {"id": "order-3"}

Common Mistakes

• Requeuing failed messages indefinitely -- infinite requeueing creates infinite retry loops. Use a dead letter queue after N failures. Set redelivery limits on the queue (RabbitMQ x-delivery-count, Kafka max.poll.interval).
• Circuit breaker per consumer instead of per downstream service -- one circuit breaker per consumer opens for all message types. Create breakers per downstream dependency. Email messages and analytics messages use different downstream services and need separate breakers.
• Not stopping consumption when circuit is open -- continuing to consume messages when the circuit is open creates a backlog of unprocessed messages. Stop the consumer or reject messages to DLQ when open.
• Half-open probes on actual messages -- don't use real messages for half-open probes. Create dedicated health check messages or probe the downstream service directly. Real messages may have side effects.
• Ignoring message processing time in timeout config -- a message that takes 30 seconds to process should not trigger the circuit breaker after 5 seconds. Set failure timeout higher than the maximum expected processing time of any message.

Practice Questions

1. How does a consumer circuit breaker differ from a client circuit breaker?
2. What is the role of a dead letter queue in circuit breaker patterns?
3. Why should failed messages not be requeued indefinitely?
4. How do you perform half-open probes in a message consumer?
5. How do you handle poison messages that always fail?

Challenge

Build a resilient message processing system: (1) RabbitMQ/Kafka consumer with circuit breakers per downstream service, (2) separate circuit breakers for email, payment, and analytics processing, (3) dead letter queue with TTL (1 hour) and automatic retry from DLQ, (4) poison message detection: if a message fails 3 times, route to a manual review queue, (5) consumer health monitoring: pause consumption when circuit is open, resume when closed, (6) Prometheus metrics: messages consumed, failed, DLQ-routed, poison messages, circuit state, (7) half-open probes using synthetic health check messages every 30 seconds.

FAQ

Mini Project

Build a resilient message processing framework: (1) consumer with circuit breakers per downstream service (RabbitMQ + Kafka support), (2) configurable failure thresholds and recovery timeouts per message type, (3) dead letter queue with automatic retry (DLQ consumer with exponential backoff), (4) poison message detection (3 failures = quarantine), (5) circuit-aware message rejection (when open, reject with "circuit open" reason), (6) health checker that sends probe messages periodically, (7) Prometheus metrics: message count by status, circuit state, DLQ depth, poison message count, (8) Grafana dashboard showing message processing health per queue.

What's Next

Continue with Event-Driven Architecture to learn event-driven circuit breaker patterns. Then explore Chaos Testing for Chaos Engineering circuit breaker tests.