Message Broker Architecture: RabbitMQ vs Kafka in Production

# docker-compose.yml
services:
  rabbitmq:
    image: rabbitmq:4.0.5-management
    ports: ["5672:5672", "15672:15672"]
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: ${RABBITMQ_PASSWORD:-secret}
      RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS: "-rabbit log_levels [{connection, warning}]"
    healthcheck:
      test: ["CMD", "rabbitmq-diagnostics", "check_port_connectivity"]
      interval: 10s
      timeout: 5s
      retries: 5
    deploy:
      resources:
        limits: { memory: 2G }

  kafka:
    image: confluentinc/cp-kafka:7.8.0
    ports: ["9092:9092"]
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_LOG_RETENTION_HOURS: 168
      KAFKA_LOG_SEGMENT_BYTES: 1073741824
    healthcheck:
      test: ["CMD", "kafka-broker-api-versions", "--bootstrap-server", "localhost:9092"]
      interval: 15s
      timeout: 10s
      retries: 5
    deploy:
      resources:
        limits: { memory: 4G }

  zookeeper:
    image: confluentinc/cp-zookeeper:7.8.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
    deploy:
      resources:
        limits: { memory: 1G }

# rabbitmq_worker.py
import pika
import json
import logging
import signal
import sys

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

class RabbitMQWorker:
    def __init__(self, url: str, queue: str):
        self.url = url
        self.queue = queue
        self.connection = None
        self.channel = None
        signal.signal(signal.SIGINT, self._shutdown)
        signal.signal(signal.SIGTERM, self._shutdown)

    def connect(self):
        params = pika.URLParameters(self.url)
        params.heartbeat = 600
        params.blocked_connection_timeout = 300
        self.connection =

pika.BlockingConnection(params) self.channel = self.connection.channel() self.channel.queue_declare(queue=self.queue, durable=True) self.channel.basic_qos(prefetch_count=1) logger.info("Connected to RabbitMQ. Waiting for messages...")

def callback(self, ch, method, properties, body):
    try:
        task = json.loads(body)
        logger.info(f"Processing task: {task}")
        # Simulate work
        # time.sleep(0.1)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        logger.error(f"Processing failed: {e}")
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

def consume(self):
    self.channel.basic_consume(queue=self.queue, on_message_callback=self.callback)
    self.channel.start_consuming()

def _shutdown(self, signum, frame):
    logger.info("Shutting down gracefully...")
    if self.channel and self.channel.is_open:
        self.channel.stop_consuming()
    if self.connection and self.connection.is_open:
        self.connection.close()
    sys.exit(0)

if name == "main": worker = RabbitMQWorker(url="amqp://admin:secret@localhost:5672/", queue="task_queue") worker.connect() worker.consume()

### 3. Kafka: Event Streaming Pattern (Python 3.12 + confluent-kafka 2.4.0)
```python
# kafka_stream.py
from confluent_kafka import Consumer, Producer, KafkaError
import json
import logging
import signal
import sys

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

class KafkaStream:
    def __init__(self, broker: str, group: str, topic: str):
        self.broker = broker
        self.group = group
        self.topic = topic
        self.consumer = None
        self.producer = None
        signal.signal(signal.SIGINT, self._shutdown)
        signal.signal(signal.SIGTERM, self._shutdown)

    def init_producer(self):
        conf = {
            "bootstrap.servers": self.broker,
            "enable.idempotence": True,
            "acks": "all",
            "transactional.id": f"producer-{self.group}",
            "linger.ms": 5,
            "batch.size": 65536
        }
        self.producer = Producer(conf)
        self.producer.init_transactions()
        self.producer.begin_transaction()
        logger.info("Kafka idempotent producer initialized.")

    def init_consumer(self):
        conf = {
            "bootstrap.servers": self.broker,
            "group.id": self.group,
            "auto.offset.reset": "earliest",
            "enable.auto.commit": False,
            "isolation.level": "read_committed",
            "max.poll.interval.ms": 300000
        }
        self.consumer = Consumer(conf)
        self.consumer.subscribe([self.topic])
        logger.info("Kafka consumer initialized. Waiting for events...")

    def produce(self, key: str, value: dict):
        self.producer.produce(self.topic, key=key, value=json.dumps(value).encode("utf-8"))
        self.producer.poll(0)

    def consume_loop(self):
        try:
            while True:
                msg = self.consumer.poll(1.0)
                if msg is None:
                    continue
                if msg.error():
                    if msg.error().code() == KafkaError._PARTITION_EOF:
                        continue
                    logger.error(f"Consumer error: {msg.error()}")
                    break
                data = json.loads(msg.value().decode("utf-8"))
                logger.info(f"Received: key={msg.key()}, offset={msg.offset()}, data={data}")
                # Process data...
                self.consumer.commit(msg, asynchronous=False)
        except KeyboardInterrupt:
            pass
        finally:
            self.consumer.close()

    def _shutdown(self, signum, frame):
        logger.info("Shutting down Kafka stream...")
        if self.producer:
            try:
                self.producer.commit_transaction()
            except Exception:
                self.producer.abort_transaction()
            self.producer.flush()
        if self.consumer:
            self.consumer.close()
        sys.exit(0)

if __name__ == "__main__":
    stream = KafkaStream(broker="localhost:9092", group="analytics-group", topic="events")
    stream.init_producer()
    stream.init_consumer()
    stream.consume_loop()

4. Execution Steps

1. Save docker-compose.yml and run docker compose up -d.
2. Verify health: docker compose ps (all services should show healthy).
3. Install dependencies: pip install pika==1.3.2 confluent-kafka==2.4.0.
4. Run RabbitMQ worker: python rabbitmq_worker.py.
5. Run Kafka stream: python kafka_stream.py.
6. Publish test messages via management UI (RabbitMQ) or kafka-console-producer.sh (Kafka).

Pitfall Guide

Symptom	Root Cause	Fix
CONNECTION_FORCED - broker forced connection closure with reason 'shutdown'	RabbitMQ memory alarm triggered (default 40% of RAM)	Increase vm_memory_high_watermark or switch to lazy queue mode. Monitor rabbitmq_memory metrics.
KafkaError{code=_GROUP_COORDINATOR_NOT_AVAILABLE}	Broker not ready or KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR mismatch	Ensure replication factor matches cluster size. Use healthcheck before connecting.
Consumer rebalancing storms	max.poll.interval.ms too low for processing time	Increase to 300000+. Tune fetch.min.bytes and max.partition.fetch.bytes to reduce poll frequency.
Duplicate messages on restart	Auto-commit enabled + crash before processing	Disable enable.auto.commit. Commit offsets synchronously after successful processing.
Publisher confirms not enabled	Missing confirm_delivery in pika or missing enable.idempotence in Kafka	RabbitMQ: channel.confirm_delivery(). Kafka: enable.idempotence: True + acks: all.
Disk exhaustion (Kafka)	Retention policy misconfigured or compaction disabled	Set log.retention.hours or log.retention.bytes. Enable cleanup.policy=compact for key-value topics.

Diagnostic Commands:

• RabbitMQ: rabbitmq-diagnostics check_port_connectivity, rabbitmqctl list_queues name messages memory
• Kafka: kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group <group>, kafka-topics.sh --describe --topic <topic>

Production Bundle

Deployment Checklist

• TLS termination at load balancer or broker level (SASL/SCRAM for auth)
• Resource limits enforced (CPU/MEM) with OOM kill prevention
• Healthchecks configured with appropriate intervals and failure thresholds
• Exactly-once disabled unless business logic strictly requires it (prefer idempotent retries)
• Dead-letter exchanges (RabbitMQ) or DLQ topics (Kafka) configured for poison messages
• Log retention aligned with compliance requirements (not default values)

Monitoring & Alerting

• RabbitMQ: rabbitmq_queue_messages_ready, rabbitmq_mem_alarm, rabbitmq_disk_free_alarm, rabbitmq_connection_count
• Kafka: kafka_server_BrokerTopicMetrics_MessagesInPerSec, kafka_consumer_fetch_manager_records_lag, kafka_controller_ActiveControllerCount, kafka_log_LogFlushRateAndTimeMs
• Alert Thresholds: Queue depth > 100k (RabbitMQ), Consumer lag > 5000 offsets for 5m (Kafka), Memory usage > 75%, Disk usage > 80%.

Scaling Strategy

• RabbitMQ: Scale vertically until memory/disk alarms. Use Shovel/Federation for cross-region, or queue sharding via consistent hashing. Avoid single-queue bottlenecks by partitioning work across multiple queues.
• Kafka: Scale horizontally by increasing partitions. Rebalance partitions during low-traffic windows. Use tiered storage for cost-effective long-term retention. Monitor partition skew to prevent hot brokers.

Cost Optimization

• Disable exactly-once semantics unless audited. Application-level idempotency (e.g., database unique constraints, Redis dedup keys) is 40–60% cheaper than broker-enforced transactions.
• Use lazy queues in RabbitMQ for high-volume, low-priority tasks to force disk paging earlier.
• Enable Kafka log compaction for stateful topics to cap storage growth without sacrificing replayability.