Backend Message Queue

Backend Message Queue

Purpose

Design consistent, production-grade message-driven systems. Every message flow must follow the same conventions for broker selection, topic/queue topology, message schema, delivery guarantees, consumer idempotency, retry, and dead-letter handling.

Agent Protocol

Trigger

Exact user phrases: "message queue", "Kafka", "RabbitMQ", "SQS", "pub-sub", "event bus", "consumer group", "topic", "queue", "at-least-once", "exactly-once", "idempotent consumer", "event sourcing", "dead letter queue", "DLQ", "message broker", "producer", "consumer", "event-driven", "async processing", "design a message flow".

Input Context

Before activating, verify:

• The business event or asynchronous task is known.
• The delivery guarantee requirement (at-most-once / at-least-once / exactly-once) is known. If not, ask: "What delivery guarantee do you need?"
• The throughput requirement is known.
• The broker preference (Kafka / RabbitMQ / SQS) is known. If not, ask: "Which broker? Kafka (high throughput, replay), RabbitMQ (routing, flexibility), or SQS (managed, simple)?"

Output Artifact

No file output unless the user requests it. Produces messaging topology specs as text.

Response Format

For each topic/queue:

Broker: {Kafka | RabbitMQ | SQS}
Name: {topic or queue name}
Type: {topic | queue | exchange + queue}
Partitions: {number} / Shards: {number}
Retention: {TTL or size limit}
Consumers: {consumer group or worker pool}
Dead-letter: {DLQ name}

For each message type:

Schema: {event_name} v{version}
Key: {partition key field}
Payload: {field list}
Guarantee: {at-most-once | at-least-once | exactly-once}
Idempotency key: {field name}

No preamble. No postamble. No explanations. No filler/hedging/transitions. Compress output — why use many token when few do trick.

Completion Criteria

• Broker is selected with justification.
• Every topic/queue has: name, type, partitions, retention, consumers, DLQ.
• Every message has: schema with version, key, payload, delivery guarantee, idempotency key.
• Consumer failure handling (retry policy, DLQ routing) is defined.
• Ordering requirements are documented (key-based partitioning).
• Schema evolution strategy is defined.

Max Response Length

Per topic/queue: 8 lines. Per message type: 6 lines.

Decision Tree

Which Broker?

What are your requirements?
  ├── High throughput (>100K msg/s), replay, log compaction, analytics
  │   └── Apache Kafka — partitioning, consumer groups, long retention
  ├── Flexible routing (topic/direct/fanout), per-message ack, task queues
  │   └── RabbitMQ — exchanges, bindings, TTL, dead-lettering
  ├── Fully managed, no ops, Lambda triggers, simple
  │   └── AWS SQS — auto-scale, limited features, 256KB max
  ├── High throughput, low latency, JVM-free
  │   └── Pulsar — geo-replication, multi-tenancy, segment-based storage
  └── Pub/sub with push delivery, mobile/web integration
      └── Google Pub/Sub — managed, exactly-once, push subscriptions

Which Delivery Guarantee?

What happens if a message is lost?
  ├── Data loss is acceptable (metrics, non-critical logs)
  │   └── At-most-once — fire and forget, lowest overhead
  ├── Duplicates are OK, data loss is not
  │   └── At-least-once — retry on failure, idempotent consumer required
  ├── Neither loss nor duplicates allowed (financial, inventory)
  │   └── Exactly-once — transactional producer + dedup consumer + idempotent processing
  └── I don't know
      └── Default to at-least-once — safest choice for most use cases

Workflow

Step 1: Select Broker

Kafka:      high throughput (100k+ msg/s), replay, log compaction, multi-consumer
            Best for: event sourcing, analytics pipelines, audit logs, CDC

RabbitMQ:   flexible routing (exchanges, bindings), low latency, per-message ack
            Best for: task queues, RPC, complex routing, lower throughput

SQS:        fully managed, no ops, auto-scaled, simple
            Best for: AWS-native, simple queues, Lambda triggers, small teams

Step 2: Define Topic / Queue Topology

Kafka:
  topic: user.events.v1 (partitions: 6, replication: 3, retention: 7d)
  consumer group: user-service-group (6 consumers = 1 per partition)

RabbitMQ:
  exchange: user.events (type: topic)
  queue: user.created (bind key: user.created.*)
  queue: user.updated (bind key: user.updated.*)
  dead-letter exchange: user.events.dlx

SQS:
  queue: user-events-queue (visibility timeout: 30s, retention: 4d)
  DLQ: user-events-dlq (max receives: 3)

Step 3: Define Message Schema

Every message has a consistent envelope:

{
  "id": "uuid",
  "type": "UserCreated",
  "version": 1,
  "timestamp": "2026-05-18T10:00:00Z",
  "producer": "user-service",
  "key": "user_abc123",
  "data": {
    "user_id": "abc123",
    "name": "Jane Doe",
    "email": "jane@example.com"
  }
}

Step 4: Delivery Guarantees

At-most-once:     fire and forget. Lowest overhead, possible data loss.
                  Use: metrics, non-critical logs.

At-least-once:    retry on failure. Lower overhead, no data loss, possible duplicates.
                  Use: order processing, notifications, data sync.
                  Required: idempotent consumer.

Exactly-once:     transactional producers + idempotent consumers + dedup.
                  High overhead. Only when mandated by compliance.
                  Use: financial transactions, inventory deduction.

Step 5: Idempotent Consumer Pattern

On receive message:
  1. Check if idempotency_key exists in processed set (Redis / DB).
  2. If exists → ack and skip (duplicate).
  3. If not exists → process, store idempotency_key, commit offset / ack.

Idempotency key = message.id or business_key + event_type
Processed set TTL: match broker retention period

class IdempotentConsumer {
  private processed = new Set<string>();

  async process<T>(message: Message<T>, handler: (data: T) => Promise<void>): Promise<void> {
    if (this.processed.has(message.id)) {
      logger.info('Duplicate message skipped', { id: message.id });
      return;
    }
    await handler(message.data);
    this.processed.add(message.id);
    // For persistence: store in Redis with TTL
  }
}

Step 6: Retry and Dead-Letter Queue

Kafka:
  - Retry topic: user.events.v1.retry (consumers reprocess after delay)
  - DLQ topic: user.events.v1.dlq (after 3 retries)
  - Alert on DLQ message production

RabbitMQ:
  - Retry policy: reject → DLX → retry queue (with TTL) → requeue
  - Max retries: 3 with exponential backoff (10s, 30s, 60s)
  - After max: route to DLQ permanently

SQS:
  - redrive policy: maxReceiveCount = 3
  - DLQ for failed messages
  - Lambda DLQ destinations for async invocation failures

// Kafka consumer with retry (Node.js)
class RetryableConsumer {
  private maxRetries = 3;
  private retryTopics: Record<string, string> = {};

  async consume(topic: string, handler: (msg: Message) => Promise<void>) {
    const consumer = this.kafka.consumer({ groupId: 'order-service' });
    await consumer.subscribe({ topic });

    await consumer.run({
      eachMessage: async ({ topic: originTopic, partition, message }) => {
        const parsed = JSON.parse(message.value!.toString());
        const retryCount = parsed._metadata?.retryCount ?? 0;

        try {
          await handler(parsed);
          await consumer.commitOffsets([{ topic: originTopic, partition, offset: message.offset }]);
        } catch (err) {
          if (retryCount >= this.maxRetries) {
            await this.sendToDLQ(parsed);
            logger.error('Message sent to DLQ after max retries', { id: parsed.id, error: err });
          } else {
            const delayTopic = `${topic}.retry-${retryCount + 1}`;
            await this.producer.send({
              topic: delayTopic,
              messages: [{
                value: JSON.stringify({
                  ...parsed,
                  _metadata: { retryCount: retryCount + 1, lastError: (err as Error).message },
                }),
              }],
            });
          }
        }
      },
    });
  }
}

Step 7: Consumer Group Scaling

Kafka partitions vs consumers:
  Partitions = 6, Consumers = 6  → Each consumer gets 1 partition (ideal)
  Partitions = 6, Consumers = 3  → Each consumer gets 2 partitions (balanced)
  Partitions = 6, Consumers = 10 → 4 consumers idle (waste)
  
  Rule: consumer count <= partition count

// Graceful shutdown for consumer
async function shutdownGracefully(consumer: Consumer) {
  process.on('SIGTERM', async () => {
    logger.info('Shutting down consumer...');
    await consumer.disconnect();
    process.exit(0);
  });
}

Step 8: Producer Patterns

// Kafka producer with idempotency
const producer = kafka.producer({
  idempotent: true,                   // exactly-once production
  maxInFlightRequests: 5,             // limit concurrency
  retries: 3,
});

async function publishEvent(event: DomainEvent) {
  await producer.send({
    topic: event.type.replace(/([a-z])([A-Z])/g, '$1.$2').toLowerCase() + '.v1',
    messages: [{
      key: event.aggregateId,         // ordering by aggregate
      value: JSON.stringify({
        id: uuidv4(),
        type: event.constructor.name,
        version: 1,
        timestamp: new Date().toISOString(),
        producer: serviceName,
        key: event.aggregateId,
        data: event,
      }),
      headers: { 'event-type': event.constructor.name },
    }],
  });
}

Step 9: Monitoring and Observability

// Kafka lag monitoring
async function checkConsumerLag(admin: Admin, groupId: string): Promise<void> {
  const lag = await admin.fetchOffsets({ groupId });
  for (const partition of lag) {
    const topicLag = partition.offset ?? 0;
    if (topicLag > 1000) {
      logger.warn('High consumer lag', { groupId, partition: partition.partition, lag: topicLag });
    }
  }
}

Production Considerations

Security

Anti-Patterns

Rules

• Never consume from production topics without a consumer group id.
• Always set a retention limit (time or size). Never use infinite retention.
• Every message must have a unique id and timestamp.
• Always use key-based partitioning when message ordering matters.
• Schema evolve via new version — never mutate existing message schemas.
• DLQ must have monitoring and alerting. Unattended DLQ = silent data loss.
• Consumer lag must be monitored. Set alerts for lag > threshold.
• Never commit offsets before processing is complete (at-least-once).
• Max message size: 1MB for Kafka, 256KB for SQS, unlimited for RabbitMQ (practical: 10MB).
• Never produce to a topic that doesn't exist — create topics with proper config first.
• Use idempotent producers for Kafka (exactly-once semantics to broker).

References

• references/broker-comparison.md — Message Broker Comparison
• references/consumer-patterns.md — Consumer Patterns
• references/kafka-patterns.md — Kafka Patterns
• references/message-design.md — Message Schema Design
• references/message-queue-monitoring.md — Message Queue Monitoring
• references/message-queue-security.md — Message Queue Security
• references/producer-patterns.md — Producer Patterns
• references/rabbitmq-patterns.md — RabbitMQ Patterns

Handoff

No artifact produced unless requested. Next skill: backend-caching — if the event-driven system needs to cache materialized views or read models. Carry forward: topic/queue topology, message schemas, consumer group configs, retry/DLQ policies.

Implementation Patterns

Kafka Producer/Consumer

from typing import Dict, Callable, Any, Optional
import json
import asyncio
from confluent_kafka import Producer, Consumer, KafkaError

class KafkaMessageProducer:
    def __init__(self, bootstrap_servers: str, client_id: str = "producer-1"):
        self.producer = Producer({
            "bootstrap.servers": bootstrap_servers,
            "client.id": client_id,
            "acks": "all",
            "enable.idempotence": True,
            "compression.type": "snappy",
        })

    def produce(self, topic: str, key: str, value: Dict, headers: Optional[Dict] = None):
        headers_list = [("content-type", "application/json")]
        if headers:
            headers_list.extend((k, v.encode()) for k, v in headers.items())

        self.producer.produce(
            topic=topic,
            key=key.encode(),
            value=json.dumps(value).encode(),
            headers=headers_list,
            callback=self._delivery_report,
        )
        self.producer.poll(0)

    def flush(self):
        self.producer.flush()

    def _delivery_report(self, err, msg):
        if err:
            print(f"Delivery failed: {err}")
        else:
            print(f"Delivered to {msg.topic()}[{msg.partition()}] @ {msg.offset()}")


class KafkaMessageConsumer:
    def __init__(self, bootstrap_servers: str, group_id: str, topics: list[str]):
        self.consumer = Consumer({
            "bootstrap.servers": bootstrap_servers,
            "group.id": group_id,
            "auto.offset.reset": "earliest",
            "enable.auto.commit": False,
            "max.poll.interval.ms": 300000,
        })
        self.consumer.subscribe(topics)

    def process_messages(self, handler: Callable, timeout: float = 1.0):
        try:
            while True:
                msg = self.consumer.poll(timeout)
                if msg is None:
                    continue
                if msg.error():
                    if msg.error().code() == KafkaError._PARTITION_EOF:
                        continue
                    print(f"Consumer error: {msg.error()}")
                    continue

                try:
                    value = json.loads(msg.value().decode())
                    handler(value, msg.key().decode() if msg.key() else None, msg.headers() or [])
                    self.consumer.commit(msg)
                except Exception as e:
                    print(f"Processing error: {e}")
                    # Send to DLQ
                    self._send_to_dlq(msg)
        except KeyboardInterrupt:
            pass
        finally:
            self.consumer.close()

    def _send_to_dlq(self, msg):
        dlq_topic = f"{msg.topic()}.dlq"
        self.consumer.produce(dlq_topic, key=msg.key(), value=msg.value())
        self.consumer.commit(msg)

Architecture Decision Trees

Broker Selection

What are the requirements?
├── High throughput (>100K msg/s), replay, long retention
│   └── Apache Kafka
│       ├── Persistent, replayable, ordered within partition
│       ├── Consumer group scaling
│       └── Schema Registry required
│
├── Simple messaging, lower throughput
│   └── RabbitMQ
│       ├── Direct, topic, fanout exchanges
│       ├── Priority queues, TTL, DLX
│       └── Good for RPC and work queues
│
├── Serverless, fully managed, no ops
│   └── AWS SQS / SNS or GCP Pub/Sub
│       ├── Automatic scaling
│       ├── No broker management
│       └── Pay per request, not per throughput
│
└── Exactly once, ordered delivery
    └── Kafka with idempotent producer + transactional API

Anti-Patterns

Performance Optimization

• Batched production: Use produce() with flush() at intervals or count. Reduces network round-trips by 10x compared to per-message flush.
• Compression: Use Snappy or Zstd compression at the producer. Reduces storage and network by 60-70% for text-based messages.
• Partition-aware consumers: Match consumer count to partition count. Avoid idle partitions and rebalance overhead. Use cooperative rebalancing for sticky partition assignment.