backend-message-queueUse this skill when the user says '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', or when designing asynchronous messaging…
Independent GitHub snapshot · Automated classification · Original documentation governs
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.
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".
Before activating, verify:
No file output unless the user requests it. Produces messaging topology specs as text.
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.
Per topic/queue: 8 lines. Per message type: 6 lines.
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
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
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
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)
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"
}
}
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.
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
}
}
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 },
}),
}],
});
}
}
},
});
}
}
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);
});
}
// 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 },
}],
});
}
| Metric | What It Tells | Alert Threshold |
|---|---|---|
| Consumer lag | How far behind consumers are | Lag > 1000 for > 5 min |
| Messages in DLQ | Permanent failures | > 0, alert immediately |
| Processing time | Consumer health | p99 > 10s |
| Throughput (msg/s) | System load | Compare to baseline |
| Failed deliveries | Broker connectivity | > 1% for > 1 min |
| Queue depth (SQS/Rabbit) | Backlog | Depth > 10000 |
// 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 });
}
}
}
| Concern | Practice |
|---|---|
| Message ordering | Use key-based partitioning. Same key = same partition = ordered |
| Schema evolution | Add fields only (backward compat). Version in envelope. Never mutate existing fields |
| Large messages | >1MB: store reference (S3 URL) in message, not the payload itself |
| Rebalancing (Kafka) | Static group membership to reduce rebalance frequency |
| Connection security | TLS for all brokers. SASL/SCRAM or mTLS for auth |
| Geo-distribution | Kafka MirrorMaker for cross-region replication. Pulsar has native geo-replication |
| Risk | Mitigation |
|---|---|
| Unauthorized produce/consume | ACLs per topic (Kafka), IAM policies (SQS), Vhost permissions (RabbitMQ) |
| Message tampering | TLS in transit. Optional: message-level HMAC or encryption |
| Sensitive data in messages | Encrypt payload at application level before producing |
| DoS via large messages | Enforce max message size at broker level |
| Credential exposure | Use IAM roles (AWS), service accounts, or vault, never hardcoded creds |
| Anti-Pattern | Why It's Bad | Fix |
|---|---|---|
| Using MQ as a database | Storage grows unbounded, no query capability | Define retention limits, use DB for persistence |
| Infinite retention | Storage explosion, slow rebalances | Set retention by time and size |
| No DLQ monitoring | Silent data loss | Alert on DLQ message production |
| Committing offset before processing | Lost messages on crash | Commit after processing (at-least-once) |
| Too many partitions | Rebalance overhead, connection overhead | Partitions = consumers × 2-3 max |
| Synchronous producing | Increases latency, reduces throughput | Batch or async produce |
| Single consumer on partitioned topic | N-1 idle partitions | Match consumer count to partitions |
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.
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)
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-Pattern | Why It Fails | Correct Approach |
|---|---|---|
| Infinite retention | Storage explosion, slow rebalances | Set retention by time (7d default) and size |
| No DLQ monitoring | Silent data loss | Alert on DLQ message production |
| Committing offset before processing | Lost messages on crash | Commit after processing (at-least-once) |
| Too many partitions | Rebalance overhead, connection overhead | Partitions = consumers x 2-3 max |
produce() with flush() at intervals or count. Reduces network round-trips by 10x compared to per-message flush.