Using Message Queues

Message Queues

Implement asynchronous communication patterns for event-driven architectures, background job processing, and service decoupling.

When to Use This Skill

Use message queues when:

• Long-running operations block HTTP requests (report generation, video processing)
• Service decoupling required (microservices, event-driven architecture)
• Guaranteed delivery needed (payment processing, order fulfillment)
• Event streaming for analytics (log aggregation, metrics pipelines)
• Workflow orchestration for complex processes (multi-step sagas, human-in-the-loop)
• Background job processing (email sending, image resizing)

Broker Selection Decision Tree

Choose message broker based on primary need:

Event Streaming / Log Aggregation

→ Apache Kafka

• Throughput: 500K-1M msg/s
• Replay events (event sourcing)
• Exactly-once semantics
• Long-term retention
• Use: Analytics pipelines, CQRS, event sourcing

Simple Background Jobs

→ Task Queues

• Python → Celery + Redis
• TypeScript → BullMQ + Redis
• Go → Asynq + Redis
• Use: Email sending, report generation, webhooks

Complex Workflows / Sagas

→ Temporal

• Durable execution (survives restarts)
• Saga pattern support
• Human-in-the-loop workflows
• Use: Order processing, AI agent orchestration

Request-Reply / RPC Patterns

→ NATS

• Built-in request-reply
• Sub-millisecond latency
• Cloud-native, simple operations
• Use: Microservices RPC, IoT command/control

Complex Message Routing

→ RabbitMQ

• Exchanges (direct, topic, fanout, headers)
• Dead letter exchanges
• Message TTL, priorities
• Use: Multi-consumer patterns, pub/sub

Already Using Redis

→ Redis Streams

• No new infrastructure
• Simple consumer groups
• Moderate throughput (100K+ msg/s)
• Use: Notification queues, simple job queues

Performance Comparison

Quick Start Examples

Kafka Producer/Consumer (Python)

See examples/kafka-python/ for working code.

from confluent_kafka import Producer, Consumer

# Producer
producer = Producer({'bootstrap.servers': 'localhost:9092'})
producer.produce('orders', key='order_123', value='{"status": "created"}')
producer.flush()

# Consumer
consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'order-processors',
    'auto.offset.reset': 'earliest'
})
consumer.subscribe(['orders'])

while True:
    msg = consumer.poll(1.0)
    if msg is not None:
        process_order(msg.value())

Celery Background Jobs (Python)

See examples/celery-image-processing/ for full implementation.

from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379')

@app.task(bind=True, max_retries=3)
def process_image(self, image_url: str):
    try:
        result = expensive_image_processing(image_url)
        return result
    except RecoverableError as e:
        raise self.retry(exc=e, countdown=60)

BullMQ Job Processing (TypeScript)

See examples/bullmq-webhook-processor/ for full implementation.

import { Queue, Worker } from 'bullmq'

const queue = new Queue('webhooks', {
  connection: { host: 'localhost', port: 6379 }
})

// Enqueue job
await queue.add('send-webhook', {
  url: 'https://example.com/webhook',
  payload: { event: 'order.created' }
})

// Process jobs
const worker = new Worker('webhooks', async job => {
  await fetch(job.data.url, {
    method: 'POST',
    body: JSON.stringify(job.data.payload)
  })
}, { connection: { host: 'localhost', port: 6379 } })

Temporal Workflow Orchestration

See examples/temporal-order-saga/ for saga pattern implementation.

from temporalio import workflow, activity
from datetime import timedelta

@workflow.defn
class OrderSagaWorkflow:
    @workflow.run
    async def run(self, order_id: str) -> str:
        # Step 1: Reserve inventory
        inventory_id = await workflow.execute_activity(
            reserve_inventory,
            order_id,
            start_to_close_timeout=timedelta(seconds=10),
        )

        # Step 2: Charge payment
        payment_id = await workflow.execute_activity(
            charge_payment,
            order_id,
            start_to_close_timeout=timedelta(seconds=30),
        )

        return f"Order {order_id} completed"

Core Patterns

Event Naming Convention

Use: Domain.Entity.Action.Version

Examples:

• order.created.v1
• user.profile.updated.v2
• payment.failed.v1

Event Schema Structure

{
  "event_type": "order.created.v2",
  "event_id": "uuid-here",
  "timestamp": "2025-12-02T10:00:00Z",
  "version": "2.0",
  "data": {
    "order_id": "ord_123",
    "customer_id": "cus_456"
  },
  "metadata": {
    "producer": "order-service",
    "trace_id": "abc123",
    "correlation_id": "xyz789"
  }
}

Dead Letter Queue Pattern

Route failed messages to dead letter queue (DLQ) after max retries:

@app.task(bind=True, max_retries=3)
def process_order(self, order_id: str):
    try:
        result = perform_processing(order_id)
        return result
    except UnrecoverableError as e:
        send_to_dlq(order_id, str(e))
        raise Reject(e, requeue=False)

Idempotency for Exactly-Once Processing

@app.post("/process")
async def process_payment(
    payment_data: dict,
    idempotency_key: str = Header(None)
):
    # Check if already processed
    cached_result = redis_client.get(f"idempotency:{idempotency_key}")
    if cached_result:
        return {"status": "already_processed"}

    result = process_payment_logic(payment_data)
    redis_client.setex(f"idempotency:{idempotency_key}", 86400, result)
    return {"status": "processed", "result": result}

Frontend Integration

Job Status Updates via SSE

# FastAPI endpoint for real-time job status
@app.get("/status/{task_id}")
async def task_status_stream(task_id: str):
    async def event_generator():
        while True:
            task = celery_app.AsyncResult(task_id)

            if task.state == 'PROGRESS':
                yield {"event": "progress", "data": task.info.get('progress', 0)}
            elif task.state == 'SUCCESS':
                yield {"event": "complete", "data": task.result}
                break

            await asyncio.sleep(0.5)

    return EventSourceResponse(event_generator())

React Component

export function JobStatus({ jobId }: { jobId: string }) {
  const [progress, setProgress] = useState(0)

  useEffect(() => {
    const eventSource = new EventSource(`/api/status/${jobId}`)

    eventSource.addEventListener('progress', (e) => {
      setProgress(JSON.parse(e.data))
    })

    eventSource.addEventListener('complete', (e) => {
      toast({ title: 'Job complete', description: JSON.parse(e.data) })
      eventSource.close()
    })

    return () => eventSource.close()
  }, [jobId])

  return <ProgressBar value={progress} />
}

Detailed Guides

For comprehensive documentation, see reference files:

Broker-Specific Guides

• Kafka: See references/kafka.md for partitioning, consumer groups, exactly-once semantics
• RabbitMQ: See references/rabbitmq.md for exchanges, bindings, routing patterns
• NATS: See references/nats.md for JetStream, request-reply patterns
• Redis Streams: See references/redis-streams.md for consumer groups, acknowledgments

Task Queue Guides

• Celery: See references/celery.md for periodic tasks, canvas (workflows), monitoring
• BullMQ: See references/bullmq.md for job prioritization, flows, Bull Board monitoring
• Temporal: See references/temporal-workflows.md for saga patterns, signals, queries

Pattern Guides

• Event Patterns: See references/event-patterns.md for event sourcing, CQRS, outbox pattern

Common Anti-Patterns to Avoid

1. Synchronous API for Long Operations

# ❌ BAD: Blocks request thread
@app.post("/generate-report")
def generate_report(user_id: str):
    report = expensive_computation(user_id)  # 5 minutes!
    return report

# ✅ GOOD: Enqueue background job
@app.post("/generate-report")
async def generate_report(user_id: str):
    task = generate_report_task.delay(user_id)
    return {"task_id": task.id}

2. Non-Idempotent Consumers

# ❌ BAD: Processes duplicates
@app.task
def send_email(email: str):
    send_email_service(email)  # Sends twice if retried!

# ✅ GOOD: Idempotent with deduplication
@app.task
def send_email(email: str, idempotency_key: str):
    if redis.exists(f"sent:{idempotency_key}"):
        return "already_sent"
    send_email_service(email)
    redis.setex(f"sent:{idempotency_key}", 86400, "1")

3. Ignoring Dead Letter Queues

# ❌ BAD: Failed messages lost forever
@app.task(max_retries=3)
def risky_task(data):
    process(data)  # If all retries fail, data disappears

# ✅ GOOD: DLQ for manual inspection
@app.task(max_retries=3)
def risky_task(data):
    try:
        process(data)
    except Exception as e:
        if self.request.retries >= 3:
            send_to_dlq(data, str(e))
        raise

4. Using Kafka for Request-Reply

# ❌ BAD: Kafka is not designed for RPC
def get_user_profile(user_id: str):
    kafka_producer.send("user_requests", {"user_id": user_id})
    # How to correlate response? Kafka is asynchronous!

# ✅ GOOD: Use NATS request-reply or HTTP/gRPC
response = await nats.request("user.profile", user_id.encode())

Library Recommendations

Context7 Research

Confluent Kafka (Python)

• Context7 ID: /confluentinc/confluent-kafka-python
• Trust Score: 68.8/100
• Code Snippets: 192+
• Production-ready Python Kafka client

Temporal

• Context7 ID: /websites/temporal_io
• Trust Score: 80.9/100
• Code Snippets: 3,769+
• Workflow orchestration for durable execution

Installation

Python:

pip install confluent-kafka celery[redis] temporalio aio-pika redis

TypeScript/Node.js:

npm install kafkajs bullmq @temporalio/client amqplib ioredis

Rust:

cargo add rdkafka lapin async-nats redis

Go:

go get github.com/confluentinc/confluent-kafka-go
go get github.com/hibiken/asynq
go get go.temporal.io/sdk

Utilities

Use scripts for setup automation:

• Kafka setup: Run python scripts/kafka_producer_consumer.py for test utilities
• Schema validation: Run python scripts/validate_message_schema.py to validate event schemas

Related Skills

• api-patterns: API design for async job submission
• realtime-sync: WebSocket/SSE for job status updates
• feedback: Toast notifications for job completion
• databases-*: Persistent storage for event logs
• observability: Tracing and metrics for queue operations

BullMQ Webhook Processor Example

Resilient webhook processing system using BullMQ with retry logic, dead letter queue, and monitoring.

Use Case

Process incoming webhooks asynchronously with:

• Immediate HTTP 200 response (webhook doesn't timeout)
• Retry failed webhooks with exponential backoff
• Dead letter queue for permanently failed webhooks
• Rate limiting to external APIs
• Monitoring dashboard

Architecture

Webhook POST → Express API → BullMQ Queue → Worker → External API
                   ↓                              ↓
               200 OK (instant)            Process async

Files

bullmq-webhook-processor/
├── src/
│   ├── server.ts            # Express API
│   ├── queues/
│   │   └── webhookQueue.ts  # Queue setup
│   ├── workers/
│   │   └── webhookWorker.ts # Job processor
│   └── processors/
│       └── stripeProcessor.ts
├── package.json
└── .env.example

Quick Start

npm install
docker run -p 6379:6379 redis  # Start Redis
npm run dev

Implementation

Queue Setup

// queues/webhookQueue.ts
import { Queue } from 'bullmq';

export const webhookQueue = new Queue('webhooks', {
  connection: { host: 'localhost', port: 6379 },
  defaultJobOptions: {
    attempts: 5,
    backoff: { type: 'exponential', delay: 2000 },
    removeOnComplete: { age: 86400 },  // Keep 24 hours
    removeOnFail: { age: 604800 },     // Keep 7 days
  },
});

Worker

// workers/webhookWorker.ts
import { Worker } from 'bullmq';

const worker = new Worker('webhooks', async (job) => {
  const { provider, event } = job.data;

  console.log(`Processing ${provider} webhook:`, job.id);

  switch (provider) {
    case 'stripe':
      return await processStripeWebhook(event);
    case 'github':
      return await processGitHubWebhook(event);
    default:
      throw new Error(`Unknown provider: ${provider}`);
  }
}, {
  connection: { host: 'localhost', port: 6379 },
  concurrency: 10,
  limiter: { max: 100, duration: 60000 },  // 100 jobs/min
});

worker.on('completed', (job) => {
  console.log(`✓ Job ${job.id} completed`);
});

worker.on('failed', (job, err) => {
  console.error(`✗ Job ${job.id} failed:`, err.message);
});

API Endpoint

// server.ts
import express from 'express';
import { webhookQueue } from './queues/webhookQueue';

const app = express();
app.use(express.json());

app.post('/webhooks/stripe', async (req, res) => {
  // Validate webhook signature
  const signature = req.headers['stripe-signature'];
  // ... validation logic ...

  // Queue for async processing
  await webhookQueue.add('stripe-event', {
    provider: 'stripe',
    event: req.body,
    receivedAt: new Date().toISOString(),
  }, {
    jobId: req.body.id,  // Idempotency (prevent duplicates)
  });

  // Immediate response
  res.json({ received: true });
});

app.listen(3000);

Features

• Automatic retries with exponential backoff
• Job deduplication via jobId
• Progress tracking
• Failed job monitoring
• Bull Board dashboard integration

Monitoring

import { createBullBoard } from '@bull-board/api';
import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
import { ExpressAdapter } from '@bull-board/express';

const serverAdapter = new ExpressAdapter();
serverAdapter.setBasePath('/admin/queues');

createBullBoard({
  queues: [new BullMQAdapter(webhookQueue)],
  serverAdapter,
});

app.use('/admin/queues', serverAdapter.getRouter());

Access: http://localhost:3000/admin/queues

Celery Image Processing Pipeline

Distributed image processing using Celery with FastAPI, including optimization, thumbnail generation, and CDN upload.

Use Case

Handle image uploads asynchronously: 1. User uploads image → immediate response 2. Background: Optimize, generate thumbnails, upload to S3 3. Notify user when complete

Files

celery-image-processing/
├── app/
│   ├── main.py              # FastAPI app
│   ├── celery_app.py        # Celery configuration
│   ├── tasks/
│   │   └── image_tasks.py   # Image processing tasks
│   └── routes/
│       └── upload.py        # Upload endpoint
├── requirements.txt
└── .env.example

Quick Start

# Install
pip install -r requirements.txt

# Start Redis
docker run -p 6379:6379 redis

# Start Celery worker
celery -A app.celery_app worker --loglevel=info

# Start FastAPI
uvicorn app.main:app --reload

Implementation

Celery Tasks

# app/tasks/image_tasks.py
from celery import chain
from PIL import Image
import boto3

@celery_app.task
def optimize_image(image_path):
    """Optimize image (reduce file size)"""
    img = Image.open(image_path)

    # Optimize
    img.save(
        image_path,
        optimize=True,
        quality=85,
    )

    return image_path

@celery_app.task
def generate_thumbnail(image_path):
    """Generate 200x200 thumbnail"""
    img = Image.open(image_path)
    img.thumbnail((200, 200))

    thumb_path = image_path.replace('.jpg', '_thumb.jpg')
    img.save(thumb_path)

    return thumb_path

@celery_app.task
def upload_to_s3(file_paths):
    """Upload files to S3"""
    s3 = boto3.client('s3')
    urls = []

    for path in file_paths:
        key = f"images/{Path(path).name}"
        s3.upload_file(path, 'my-bucket', key)
        urls.append(f"https://my-bucket.s3.amazonaws.com/{key}")

    return urls

@celery_app.task
def notify_user(user_id, image_urls):
    """Send notification to user"""
    # Send email/push notification
    return {'notified': user_id, 'urls': image_urls}

FastAPI Endpoint

# app/routes/upload.py
from fastapi import APIRouter, UploadFile, File
from celery import chain

router = APIRouter()

@router.post("/upload")
async def upload_image(file: UploadFile = File(...), user_id: int):
    # Save temp file
    temp_path = f"/tmp/{file.filename}"
    with open(temp_path, "wb") as f:
        content = await file.read()
        f.write(content)

    # Create task chain
    workflow = chain(
        optimize_image.s(temp_path),
        generate_thumbnail.s(),
        upload_to_s3.s(),
        notify_user.s(user_id),
    )

    task = workflow.apply_async()

    return {
        "message": "Processing started",
        "task_id": task.id,
        "status_url": f"/status/{task.id}"
    }

@router.get("/status/{task_id}")
async def get_status(task_id: str):
    from celery.result import AsyncResult

    task = AsyncResult(task_id)

    return {
        "task_id": task_id,
        "status": task.status,
        "result": task.result if task.ready() else None,
    }

Error Handling

@celery_app.task(bind=True, max_retries=3)
def upload_to_s3(self, file_path):
    try:
        s3.upload_file(file_path, 'bucket', 'key')
    except ClientError as exc:
        if exc.response['Error']['Code'] == 'SlowDown':
            # Retry with backoff
            raise self.retry(exc=exc, countdown=60)
        else:
            # Don't retry (permanent failure)
            raise

Monitoring

# Start Flower dashboard
celery -A app.celery_app flower --port=5555

Features:

• Task progress tracking
• Failed task inspection
• Worker health monitoring
• Task rate graphs

Kafka Python Producer/Consumer Example

Production-ready Kafka producer and consumer implementations using confluent-kafka-python.

Prerequisites

# Install confluent-kafka
pip install confluent-kafka

# Start Kafka (Docker)
docker run -d \
  --name kafka \
  -p 9092:9092 \
  apache/kafka:latest

Files

• producer.py - Producer examples (simple, batching, partitioning, error handling)
• consumer.py - Consumer examples (manual commit, batching, multiple topics)

Running Examples

Producer

python producer.py

Output:

🚀 Kafka Producer Examples
==================================================

=== Example 1: Simple Message ===
✅ Message delivered to orders [0] @ offset 42

=== Example 2: Message with Headers ===
✅ Message delivered to payments [1] @ offset 15

=== Example 3: Batch Production ===
Produced 100 messages in 0.23 seconds (435 msg/s)

Consumer

python consumer.py

Output:

🚀 Kafka Consumer Examples
==================================================

=== Example 1: Manual Commit ===
📦 Processing order: ord_123
✅ Committed offset 42
📦 Processing order: ord_456
✅ Committed offset 43

Key Features

Producer

• ✅ Idempotent producer (prevents duplicates)
• ✅ Batching for efficiency
• ✅ Custom partitioning
• ✅ Headers for metadata
• ✅ Error handling with retries

Consumer

• ✅ Manual offset commits (exactly-once)
• ✅ Batch processing
• ✅ Error handling (recoverable vs unrecoverable)
• ✅ Dead letter queue pattern
• ✅ Multiple topic subscriptions

Configuration

Producer Config

config = {
    'bootstrap.servers': 'localhost:9092',
    'acks': 'all',  # Wait for all replicas
    'enable.idempotence': True,  # Prevent duplicates
    'compression.type': 'lz4',  # Fast compression
}

Consumer Config

config = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'python-consumer-group',
    'enable.auto.commit': False,  # Manual commit
    'auto.offset.reset': 'earliest',
}

Topics Used

• orders - Order events
• payments - Payment events
• logs - Log messages
• inventory - Inventory updates
• alerts - Critical alerts

Monitoring

Check Topics

docker exec kafka kafka-topics --list --bootstrap-server localhost:9092

Consumer Group Lag

docker exec kafka kafka-consumer-groups \
  --describe \
  --group python-consumer-group \
  --bootstrap-server localhost:9092

References

• Confluent Kafka Python: /confluentinc/confluent-kafka-python
• Trust Score: 68.8/100
• Code Snippets: 192+

Temporal Order Saga Example

Distributed saga pattern for order processing using Temporal workflow orchestration.

Use Case

E-commerce order processing with multiple steps that need coordination: 1. Reserve inventory 2. Charge payment 3. Create shipment 4. Send confirmation

If any step fails, compensating transactions rollback previous steps.

What is Temporal?

Durable workflow engine that persists execution state. If worker crashes, workflow resumes from last checkpoint.

Files

temporal-order-saga/
├── workflows/
│   └── orderWorkflow.ts     # Order saga workflow
├── activities/
│   ├── inventory.ts         # Inventory operations
│   ├── payment.ts           # Payment operations
│   └── shipping.ts          # Shipping operations
├── worker.ts                # Temporal worker
├── client.ts                # Start workflow
└── package.json

Quick Start

# Install Temporal CLI
brew install temporal

# Start Temporal server
temporal server start-dev

# Install dependencies
npm install

# Run worker
npm run worker

# Execute workflow (separate terminal)
npm run start-order

Workflow Implementation

// workflows/orderWorkflow.ts
import { proxyActivities } from '@temporalio/workflow';
import * as activities from '../activities';

const { reserveInventory, chargePayment, createShipment, sendConfirmation } =
  proxyActivities<typeof activities>({
    startToCloseTimeout: '1 minute',
  });

export async function orderWorkflow(orderId: string): Promise<string> {
  let inventoryReserved = false;
  let paymentCharged = false;
  let shipmentCreated = false;

  try {
    // Step 1: Reserve inventory
    await reserveInventory(orderId);
    inventoryReserved = true;

    // Step 2: Charge payment
    await chargePayment(orderId);
    paymentCharged = true;

    // Step 3: Create shipment
    await createShipment(orderId);
    shipmentCreated = true;

    // Step 4: Send confirmation
    await sendConfirmation(orderId);

    return 'Order completed successfully';

  } catch (error) {
    // Compensating transactions (rollback)
    if (shipmentCreated) {
      await cancelShipment(orderId);
    }
    if (paymentCharged) {
      await refundPayment(orderId);
    }
    if (inventoryReserved) {
      await releaseInventory(orderId);
    }

    throw error;
  }
}

Activities (Actual Work)

// activities/inventory.ts
export async function reserveInventory(orderId: string): Promise<void> {
  const order = await db.order.findUnique({ where: { id: orderId } });

  for (const item of order.items) {
    const updated = await db.product.update({
      where: { id: item.productId, stock: { gte: item.quantity } },
      data: { stock: { decrement: item.quantity } },
    });

    if (!updated) {
      throw new Error(`Insufficient stock for product ${item.productId}`);
    }
  }
}

export async function releaseInventory(orderId: string): Promise<void> {
  const order = await db.order.findUnique({ where: { id: orderId } });

  for (const item of order.items) {
    await db.product.update({
      where: { id: item.productId },
      data: { stock: { increment: item.quantity } },
    });
  }
}

// activities/payment.ts
export async function chargePayment(orderId: string): Promise<void> {
  const order = await db.order.findUnique({ where: { id: orderId } });

  const charge = await stripe.charges.create({
    amount: order.total * 100,
    currency: 'usd',
    source: order.paymentMethodId,
  });

  await db.order.update({
    where: { id: orderId },
    data: { stripeChargeId: charge.id },
  });
}

export async function refundPayment(orderId: string): Promise<void> {
  const order = await db.order.findUnique({ where: { id: orderId } });

  await stripe.refunds.create({
    charge: order.stripeChargeId,
  });
}

Starting Workflow

// client.ts
import { Client } from '@temporalio/client';
import { orderWorkflow } from './workflows/orderWorkflow';

const client = new Client();

async function processOrder(orderId: string) {
  const handle = await client.workflow.start(orderWorkflow, {
    taskQueue: 'orders',
    workflowId: `order-${orderId}`,  // Unique workflow ID
    args: [orderId],
  });

  console.log(`Started workflow: ${handle.workflowId}`);

  // Wait for result
  const result = await handle.result();
  console.log('Workflow result:', result);
}

processOrder('order-123');

Error Handling Benefits

Without Temporal (fragile):

• Worker crashes mid-process → orphaned payment charge
• Network error → need manual cleanup
• Hard to track saga state

With Temporal:

• Automatic retry on failures
• Guaranteed compensation execution
• Full workflow history
• Crash recovery (resume from checkpoint)

Monitoring

Access Temporal Web UI: http://localhost:8233

View:

• Workflow execution history
• Failed workflows
• Retry attempts
• Execution timeline
• Event history

When to Use Temporal vs Celery

Use Temporal when:

• Multi-step workflows with rollback logic
• Long-running processes (hours/days)
• Need execution history and replay
• Critical workflows requiring guarantees

Use Celery when:

• Simple background jobs
• No complex orchestration needed
• Python-only ecosystem
• Lower infrastructure complexity

BullMQ Reference Guide

Modern Node.js/TypeScript job queue built on Redis with advanced features like delayed jobs, rate limiting, and job prioritization.

Table of Contents

• When to Use BullMQ
• Installation
• Core Concepts
• Queue, Worker, Job
• Advanced Features
• Delayed Jobs
• Repeatable Jobs (Cron)
• Job Prioritization
• Rate Limiting
• Retry and Error Handling
• Job Events
• Horizontal Scaling
• Monitoring with Bull Board
• Job Patterns
• Fire-and-Forget
• Wait for Completion
• Job Dependencies (Flow)
• Best Practices
• Common Patterns
• Webhook Processing
• Image Processing Pipeline
• Resources

When to Use BullMQ

• TypeScript/Node.js ecosystem
• Need advanced scheduling (cron, delayed, repeatable jobs)
• Rate limiting per queue
• Job prioritization
• Horizontal scaling with workers

Installation

npm install bullmq ioredis

Core Concepts

Queue, Worker, Job

import { Queue, Worker } from 'bullmq';

// 1. Queue (producer)
const emailQueue = new Queue('emails', {
  connection: { host: 'localhost', port: 6379 }
});

// 2. Add job
await emailQueue.add('send-welcome', {
  to: 'user@example.com',
  template: 'welcome',
});

// 3. Worker (consumer)
const worker = new Worker('emails', async (job) => {
  console.log(`Processing job ${job.id}:`, job.data);

  // Simulate email sending
  await sendEmail(job.data.to, job.data.template);

  return { sent: true };
}, { connection: { host: 'localhost', port: 6379 } });

Advanced Features

Delayed Jobs

// Process in 1 hour
await queue.add('reminder', { userId: 123 }, {
  delay: 3600000,  // 1 hour in ms
});

// Process at specific time
await queue.add('scheduled-report', { reportId: 456 }, {
  delay: new Date('2025-12-04T09:00:00').getTime() - Date.now(),
});

Repeatable Jobs (Cron)

import { Queue } from 'bullmq';

await queue.add('daily-report', {}, {
  repeat: {
    pattern: '0 9 * * *',  // Every day at 9 AM
    tz: 'America/New_York',
  },
});

// Every 5 minutes
await queue.add('health-check', {}, {
  repeat: { every: 300000 },  // 5 minutes
});

Job Prioritization

// High priority (processed first)
await queue.add('critical-alert', { ... }, { priority: 1 });

// Normal priority
await queue.add('email', { ... }, { priority: 5 });

// Low priority (processed last)
await queue.add('cleanup', { ... }, { priority: 10 });

Rate Limiting

const worker = new Worker('api-calls', async (job) => {
  await callExternalAPI(job.data);
}, {
  connection: { host: 'localhost', port: 6379 },
  limiter: {
    max: 100,       // Max 100 jobs
    duration: 60000 // Per 60 seconds
  },
});

Retry and Error Handling

await queue.add('flaky-api-call', { url: '...' }, {
  attempts: 5,              // Retry up to 5 times
  backoff: {
    type: 'exponential',
    delay: 2000,            // Start with 2s, then 4s, 8s, 16s, 32s
  },
});

// Worker error handling
const worker = new Worker('api-calls', async (job) => {
  try {
    await callAPI(job.data.url);
  } catch (error) {
    if (error.code === 'RATE_LIMIT') {
      throw error;  // Retry
    }
    // Don't retry for other errors
    await job.moveToFailed({ message: error.message }, token);
  }
});

Job Events

import { QueueEvents } from 'bullmq';

const queueEvents = new QueueEvents('emails');

queueEvents.on('completed', ({ jobId, returnvalue }) => {
  console.log(`Job ${jobId} completed:`, returnvalue);
});

queueEvents.on('failed', ({ jobId, failedReason }) => {
  console.error(`Job ${jobId} failed:`, failedReason);
});

queueEvents.on('progress', ({ jobId, data }) => {
  console.log(`Job ${jobId} progress:`, data);
});

Horizontal Scaling

// Run multiple workers (same queue, different processes)

// worker1.ts
const worker1 = new Worker('emails', processEmail, {
  connection: redis,
  concurrency: 5,  // Process 5 jobs concurrently
});

// worker2.ts (separate server)
const worker2 = new Worker('emails', processEmail, {
  connection: redis,
  concurrency: 5,
});

// Jobs automatically distributed across workers

Monitoring with Bull Board

import { createBullBoard } from '@bull-board/api';
import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
import { ExpressAdapter } from '@bull-board/express';

const serverAdapter = new ExpressAdapter();
serverAdapter.setBasePath('/admin/queues');

createBullBoard({
  queues: [
    new BullMQAdapter(emailQueue),
    new BullMQAdapter(webhookQueue),
  ],
  serverAdapter,
});

app.use('/admin/queues', serverAdapter.getRouter());

Access dashboard: http://localhost:3000/admin/queues

Job Patterns

Fire-and-Forget

await queue.add('send-email', { to: 'user@example.com' });
// Don't wait for completion

Wait for Completion

const job = await queue.add('generate-report', { userId: 123 });
const result = await job.waitUntilFinished(queueEvents);
console.log('Report:', result);

Job Dependencies (Flow)

import { FlowProducer } from 'bullmq';

const flow = new FlowProducer({ connection: redis });

await flow.add({
  name: 'process-video',
  queueName: 'videos',
  data: { videoId: 123 },
  children: [
    {
      name: 'extract-thumbnail',
      queueName: 'images',
      data: { videoId: 123 },
    },
    {
      name: 'generate-subtitles',
      queueName: 'transcription',
      data: { videoId: 123 },
    },
  ],
});

Best Practices

1. Use job IDs for idempotency - Prevent duplicate processing 2. Set reasonable timeouts - Prevent stuck jobs 3. Monitor failed jobs - Set up alerting 4. Use separate queues - Different priorities/workers 5. Clean completed jobs - Prevent Redis memory growth 6. Use concurrency wisely - Based on I/O vs CPU 7. Handle errors gracefully - Distinguish retryable vs fatal errors

Common Patterns

Webhook Processing

// Receive webhook
app.post('/webhooks/stripe', async (req, res) => {
  await webhookQueue.add('stripe-event', req.body, {
    attempts: 3,
    backoff: { type: 'exponential', delay: 1000 },
  });

  res.json({ received: true });
});

// Process async
const worker = new Worker('stripe-webhooks', async (job) => {
  await processStripeEvent(job.data);
});

Image Processing Pipeline

// Upload endpoint
app.post('/upload', async (req, res) => {
  const jobId = await imageQueue.add('process-image', {
    imageUrl: req.file.url,
    userId: req.user.id,
  }, {
    priority: req.user.isPremium ? 1 : 5,
  });

  res.json({ jobId });
});

// Worker pipeline
const worker = new Worker('images', async (job) => {
  const { imageUrl } = job.data;

  // Update progress
  await job.updateProgress(10);

  const optimized = await optimizeImage(imageUrl);
  await job.updateProgress(50);

  const thumbnail = await generateThumbnail(optimized);
  await job.updateProgress(90);

  await uploadToS3(optimized, thumbnail);
  await job.updateProgress(100);

  return { optimized, thumbnail };
});

Resources

• BullMQ Docs: https://docs.bullmq.io/
• Bull Board: https://github.com/felixmosh/bull-board
• GitHub: https://github.com/taskforcesh/bullmq

Celery Reference Guide

Distributed task queue for Python with support for scheduling, retries, and result backends.

Table of Contents

• When to Use Celery
• Installation
• Basic Setup
• celery_app.py
• Running Workers
• Task Invocation
• Task Routing and Prioritization
• Periodic Tasks (Beat)
• Task Workflows
• Chain (Sequential)
• Group (Parallel)
• Chord (Parallel + Callback)
• Retry Configuration
• Task Progress Tracking
• FastAPI Integration
• Monitoring with Flower
• Result Backends
• Best Practices
• Common Patterns
• Image Processing
• Email Campaigns
• Resources

When to Use Celery

• Python ecosystem (FastAPI, Django, Flask)
• Complex task workflows (chains, groups, chords)
• Scheduled tasks (cron-like)
• Long-running background jobs
• Need multiple queue backends (Redis, RabbitMQ, SQS)

Installation

# With Redis backend
pip install celery[redis]

# With RabbitMQ backend
pip install celery[amqp]

# With SQS backend
pip install celery[sqs]

Basic Setup

celery_app.py

from celery import Celery

app = Celery(
    'myapp',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1',
)

app.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='UTC',
    enable_utc=True,
)

@app.task
def add(x, y):
    return x + y

@app.task(bind=True, max_retries=3)
def send_email(self, to, subject, body):
    try:
        # Send email logic
        return {'sent': True, 'to': to}
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60)  # Retry after 1 minute

Running Workers

# Start worker (single process)
celery -A celery_app worker --loglevel=info

# Multiple workers with concurrency
celery -A celery_app worker --concurrency=4 --loglevel=info

# Specific queue
celery -A celery_app worker -Q high-priority,default --loglevel=info

Task Invocation

# Fire and forget
result = add.delay(4, 5)

# Wait for result
result = add.delay(4, 5)
print(result.get(timeout=10))  # Blocks until complete

# Async with callback
add.apply_async((4, 5), link=notify_completion.s())

# ETA (execute at specific time)
from datetime import datetime, timedelta
add.apply_async((4, 5), eta=datetime.now() + timedelta(hours=1))

# Countdown (delay in seconds)
add.apply_async((4, 5), countdown=300)  # Run in 5 minutes

Task Routing and Prioritization

# Route tasks to specific queues
app.conf.task_routes = {
    'myapp.tasks.send_email': {'queue': 'high-priority'},
    'myapp.tasks.generate_report': {'queue': 'low-priority'},
}

# Priority (0-9, higher = more important)
send_email.apply_async((to, subject, body), priority=9)

Periodic Tasks (Beat)

from celery.schedules import crontab

app.conf.beat_schedule = {
    'send-daily-report': {
        'task': 'myapp.tasks.generate_daily_report',
        'schedule': crontab(hour=9, minute=0),  # 9 AM daily
    },
    'cleanup-old-files': {
        'task': 'myapp.tasks.cleanup_files',
        'schedule': crontab(hour=2, minute=0, day_of_week='sunday'),
    },
    'check-every-5-minutes': {
        'task': 'myapp.tasks.health_check',
        'schedule': 300.0,  # Every 5 minutes (seconds)
    },
}

Start beat scheduler:

celery -A celery_app beat --loglevel=info

Task Workflows

Chain (Sequential)

from celery import chain

# download → process → upload (sequential)
workflow = chain(
    download_image.s(url),
    process_image.s(),
    upload_to_s3.s(),
)

result = workflow.apply_async()

Group (Parallel)

from celery import group

# Process multiple images in parallel
job = group(
    process_image.s(image1),
    process_image.s(image2),
    process_image.s(image3),
)

result = job.apply_async()
results = result.get()  # Wait for all to complete

Chord (Parallel + Callback)

from celery import chord

# Process in parallel, then aggregate
job = chord(
    [process_image.s(img) for img in images],
    aggregate_results.s(),  # Called with all results
)

result = job.apply_async()

Retry Configuration

@app.task(
    bind=True,
    autoretry_for=(ConnectionError, TimeoutError),
    retry_kwargs={'max_retries': 5},
    retry_backoff=True,     # Exponential backoff
    retry_backoff_max=600,  # Max 10 minutes
    retry_jitter=True,      # Add randomness
)
def flaky_api_call(self, url):
    response = requests.get(url, timeout=10)
    return response.json()

Task Progress Tracking

@app.task(bind=True)
def long_running_task(self, items):
    total = len(items)

    for i, item in enumerate(items):
        process_item(item)

        # Update progress
        self.update_state(
            state='PROGRESS',
            meta={'current': i + 1, 'total': total, 'percent': (i + 1) / total * 100}
        )

    return {'status': 'Complete', 'processed': total}

# Check progress
result = long_running_task.delay(items)
while not result.ready():
    info = result.info
    if isinstance(info, dict):
        print(f"Progress: {info.get('percent')}%")
    time.sleep(1)

FastAPI Integration

from fastapi import FastAPI, BackgroundTasks
from celery.result import AsyncResult

app = FastAPI()

@app.post("/process")
async def process_endpoint(data: dict):
    # Queue task
    task = process_data.delay(data)

    return {"task_id": task.id, "status": "queued"}

@app.get("/status/{task_id}")
async def get_status(task_id: str):
    task = AsyncResult(task_id, app=celery_app)

    return {
        "task_id": task_id,
        "status": task.status,
        "result": task.result if task.ready() else None,
    }

Monitoring with Flower

# Install
pip install flower

# Run dashboard
celery -A celery_app flower --port=5555

Access: http://localhost:5555

Features:

• Real-time task monitoring
• Worker management
• Task history and stats
• Retry/revoke tasks
• Task rate graphs

Result Backends

# Redis (default)
app = Celery(broker='redis://localhost', backend='redis://localhost')

# PostgreSQL (persistent)
app = Celery(
    broker='redis://localhost',
    backend='db+postgresql://user:pass@localhost/celery_results'
)

# Disable backend (fire-and-forget)
app = Celery(broker='redis://localhost', backend=None)

Best Practices

1. Keep tasks small - Break large jobs into smaller tasks 2. Use task routing - Separate queues for different priorities 3. Set timeouts - Prevent hanging tasks 4. Monitor task states - Failed tasks need attention 5. Clean old results - result_expires setting 6. Use idempotent tasks - Safe to retry 7. Avoid task coupling - Tasks shouldn't depend on each other's state 8. Use serializers wisely - JSON (safe), pickle (faster but unsafe)

Common Patterns

Image Processing

@app.task
def process_uploaded_image(image_path):
    # Optimize
    optimized = optimize_image(image_path)

    # Generate variants
    thumbnail = generate_thumbnail(optimized)
    webp = convert_to_webp(optimized)

    # Upload to CDN
    urls = upload_to_cdn([optimized, thumbnail, webp])

    return {'urls': urls}

# Usage in upload endpoint
@app.post("/upload")
async def upload_image(file: UploadFile):
    path = await save_temp_file(file)
    task = process_uploaded_image.delay(path)
    return {"task_id": task.id}

Email Campaigns

@app.task
def send_campaign_emails(campaign_id):
    campaign = Campaign.objects.get(id=campaign_id)
    users = campaign.get_target_users()

    # Create group of tasks
    job = group(send_email.s(user.email, campaign.template) for user in users)
    return job.apply_async()

@app.task(rate_limit='100/m')  # Max 100 emails/minute
def send_email(to, template):
    # Actual email sending
    pass

Resources

• Celery Docs: https://docs.celeryq.dev/
• Flower: https://flower.readthedocs.io/
• GitHub: https://github.com/celery/celery

Event Patterns - Event Sourcing, CQRS, Outbox

Advanced event-driven architecture patterns for building scalable, reliable systems.

Table of Contents

• Event Sourcing
• Implementation
• Persistence with Kafka
• CQRS (Command Query Responsibility Segregation)
• Implementation
• Outbox Pattern
• Implementation
• Outbox Table Schema
• Event Versioning
• Upcasting Pattern
• Saga Pattern (Distributed Transactions)
• Event-Based Saga (Choreography)
• Best Practices
• 1. Event Naming Convention
• 2. Event Schema Evolution
• 3. Idempotent Event Handlers
• 4. Event Metadata
• Related Patterns

Event Sourcing

Pattern: Store state changes as immutable events instead of current state.

Benefits:

• Complete audit log (every change recorded)
• Time travel (replay to any point)
• Event replay for new projections
• Natural fit for event-driven systems

Trade-offs:

• More complex than CRUD
• Requires event versioning strategy
• Eventual consistency

Implementation

from dataclasses import dataclass
from typing import List
from datetime import datetime
import json

@dataclass
class Event:
    """Base event"""
    event_id: str
    event_type: str
    aggregate_id: str
    timestamp: datetime
    version: int
    data: dict

@dataclass
class OrderCreated(Event):
    event_type: str = "order.created.v1"

@dataclass
class ItemAdded(Event):
    event_type: str = "item.added.v1"

@dataclass
class OrderSubmitted(Event):
    event_type: str = "order.submitted.v1"

class OrderAggregate:
    """Event-sourced aggregate"""
    def __init__(self, order_id: str):
        self.order_id = order_id
        self.events: List[Event] = []
        self.version = 0

        # State (derived from events)
        self.status = "draft"
        self.items = []
        self.total = 0.0

    def create_order(self, customer_id: str):
        """Command: Create order"""
        event = OrderCreated(
            event_id=str(uuid.uuid4()),
            event_type="order.created.v1",
            aggregate_id=self.order_id,
            timestamp=datetime.utcnow(),
            version=self.version + 1,
            data={"customer_id": customer_id}
        )
        self.apply(event)
        self.events.append(event)

    def add_item(self, item_id: str, quantity: int, price: float):
        """Command: Add item"""
        if self.status != "draft":
            raise ValueError("Can only add items to draft orders")

        event = ItemAdded(
            event_id=str(uuid.uuid4()),
            event_type="item.added.v1",
            aggregate_id=self.order_id,
            timestamp=datetime.utcnow(),
            version=self.version + 1,
            data={"item_id": item_id, "quantity": quantity, "price": price}
        )
        self.apply(event)
        self.events.append(event)

    def submit_order(self):
        """Command: Submit order"""
        if not self.items:
            raise ValueError("Cannot submit empty order")

        event = OrderSubmitted(
            event_id=str(uuid.uuid4()),
            event_type="order.submitted.v1",
            aggregate_id=self.order_id,
            timestamp=datetime.utcnow(),
            version=self.version + 1,
            data={"total": self.total}
        )
        self.apply(event)
        self.events.append(event)

    def apply(self, event: Event):
        """Apply event to update state"""
        if event.event_type == "order.created.v1":
            self.status = "draft"
        elif event.event_type == "item.added.v1":
            self.items.append(event.data)
            self.total += event.data['price'] * event.data['quantity']
        elif event.event_type == "order.submitted.v1":
            self.status = "submitted"

        self.version = event.version

    @classmethod
    def from_events(cls, order_id: str, events: List[Event]):
        """Rebuild aggregate from events (event replay)"""
        aggregate = cls(order_id)
        for event in events:
            aggregate.apply(event)
            aggregate.events.append(event)
        return aggregate

Persistence with Kafka

from confluent_kafka import Producer, Consumer
import json
import uuid

class EventStore:
    """Kafka-based event store"""
    def __init__(self, bootstrap_servers: str):
        self.producer = Producer({'bootstrap.servers': bootstrap_servers})
        self.consumer = Consumer({
            'bootstrap.servers': bootstrap_servers,
            'group.id': f'event-replay-{uuid.uuid4()}',
            'auto.offset.reset': 'earliest'
        })

    def save_events(self, aggregate_id: str, events: List[Event]):
        """Append events to stream"""
        for event in events:
            self.producer.produce(
                topic='order-events',
                key=aggregate_id,  # All events for same aggregate in same partition
                value=json.dumps({
                    'event_id': event.event_id,
                    'event_type': event.event_type,
                    'aggregate_id': event.aggregate_id,
                    'timestamp': event.timestamp.isoformat(),
                    'version': event.version,
                    'data': event.data
                })
            )
        self.producer.flush()

    def load_events(self, aggregate_id: str) -> List[Event]:
        """Load all events for aggregate (event replay)"""
        self.consumer.subscribe(['order-events'])

        events = []
        while True:
            msg = self.consumer.poll(1.0)
            if msg is None:
                break

            if msg.key().decode() == aggregate_id:
                event_data = json.loads(msg.value().decode())
                events.append(Event(
                    event_id=event_data['event_id'],
                    event_type=event_data['event_type'],
                    aggregate_id=event_data['aggregate_id'],
                    timestamp=datetime.fromisoformat(event_data['timestamp']),
                    version=event_data['version'],
                    data=event_data['data']
                ))

        return sorted(events, key=lambda e: e.version)

CQRS (Command Query Responsibility Segregation)

Pattern: Separate write model (commands) from read model (queries).

Benefits:

• Optimize read and write independently
• Scale read and write separately
• Multiple read models from same events
• Complex queries without impacting writes

Trade-offs:

• Eventual consistency between models
• More complex than single model
• Need to synchronize models

Implementation

# WRITE MODEL (Command Side)
class OrderCommandHandler:
    """Handle write operations"""
    def __init__(self, event_store: EventStore):
        self.event_store = event_store

    def create_order(self, order_id: str, customer_id: str):
        """Command: Create order"""
        aggregate = OrderAggregate(order_id)
        aggregate.create_order(customer_id)

        # Save events
        self.event_store.save_events(order_id, aggregate.events)

    def add_item(self, order_id: str, item_id: str, quantity: int, price: float):
        """Command: Add item"""
        # Load events and rebuild aggregate
        events = self.event_store.load_events(order_id)
        aggregate = OrderAggregate.from_events(order_id, events)

        # Execute command
        aggregate.add_item(item_id, quantity, price)

        # Save new events
        self.event_store.save_events(order_id, aggregate.events[-1:])


# READ MODEL (Query Side)
class OrderReadModel:
    """Optimized for queries"""
    def __init__(self, db):
        self.db = db  # Could be PostgreSQL, MongoDB, etc.

    def get_order(self, order_id: str) -> dict:
        """Query: Get order details"""
        return self.db.query("SELECT * FROM orders WHERE id = ?", order_id)

    def get_customer_orders(self, customer_id: str) -> List[dict]:
        """Query: Get all orders for customer"""
        return self.db.query("SELECT * FROM orders WHERE customer_id = ?", customer_id)

    def get_order_summary(self, order_id: str) -> dict:
        """Query: Get order summary with items"""
        return self.db.query("""
            SELECT o.*, COUNT(i.id) as item_count, SUM(i.total) as total
            FROM orders o
            LEFT JOIN order_items i ON o.id = i.order_id
            WHERE o.id = ?
            GROUP BY o.id
        """, order_id)


# PROJECTOR (Sync write model → read model)
class OrderProjector:
    """Project events to read model"""
    def __init__(self, read_model: OrderReadModel):
        self.read_model = read_model

    def handle_event(self, event: Event):
        """Update read model based on event"""
        if event.event_type == "order.created.v1":
            self.read_model.db.execute(
                "INSERT INTO orders (id, customer_id, status) VALUES (?, ?, ?)",
                event.aggregate_id,
                event.data['customer_id'],
                'draft'
            )

        elif event.event_type == "item.added.v1":
            self.read_model.db.execute(
                "INSERT INTO order_items (order_id, item_id, quantity, price) VALUES (?, ?, ?, ?)",
                event.aggregate_id,
                event.data['item_id'],
                event.data['quantity'],
                event.data['price']
            )

        elif event.event_type == "order.submitted.v1":
            self.read_model.db.execute(
                "UPDATE orders SET status = ? WHERE id = ?",
                'submitted',
                event.aggregate_id
            )

    def project_all_events(self, stream: str):
        """Rebuild read model from scratch (event replay)"""
        consumer = Consumer({
            'bootstrap.servers': 'localhost:9092',
            'group.id': 'projector',
            'auto.offset.reset': 'earliest'
        })
        consumer.subscribe([stream])

        while True:
            msg = consumer.poll(1.0)
            if msg is None:
                break

            event_data = json.loads(msg.value().decode())
            event = Event(**event_data)
            self.handle_event(event)

Outbox Pattern

Pattern: Ensure database writes and event publishing are atomic (no partial failures).

Problem:

# ❌ NOT ATOMIC
db.insert_order(order)  # Succeeds
kafka.publish(event)     # FAILS → inconsistent state!

Solution: Write events to outbox table in same transaction, then publish asynchronously.

Implementation

import psycopg2
from psycopg2.extras import Json
from confluent_kafka import Producer
import json
import time

class OutboxPattern:
    """Transactional outbox pattern"""
    def __init__(self, db_conn, kafka_producer: Producer):
        self.db = db_conn
        self.producer = kafka_producer

    def create_order_with_event(self, order_id: str, customer_id: str):
        """Create order and event atomically"""
        cursor = self.db.cursor()

        try:
            # Begin transaction
            cursor.execute("BEGIN")

            # Insert order
            cursor.execute(
                "INSERT INTO orders (id, customer_id, status) VALUES (%s, %s, %s)",
                (order_id, customer_id, 'draft')
            )

            # Insert event to outbox (same transaction)
            event = {
                'event_type': 'order.created.v1',
                'aggregate_id': order_id,
                'timestamp': datetime.utcnow().isoformat(),
                'data': {'customer_id': customer_id}
            }

            cursor.execute(
                "INSERT INTO outbox (id, event_type, aggregate_id, payload) VALUES (%s, %s, %s, %s)",
                (str(uuid.uuid4()), event['event_type'], order_id, Json(event))
            )

            # Commit transaction (both writes succeed or both fail)
            cursor.execute("COMMIT")

        except Exception as e:
            cursor.execute("ROLLBACK")
            raise

    def publish_outbox_events(self):
        """Background worker: Publish events from outbox"""
        cursor = self.db.cursor()

        while True:
            # Fetch unpublished events
            cursor.execute(
                "SELECT id, aggregate_id, payload FROM outbox WHERE published = false ORDER BY created_at LIMIT 100"
            )

            events = cursor.fetchall()

            for event_id, aggregate_id, payload in events:
                try:
                    # Publish to Kafka
                    self.producer.produce(
                        topic='order-events',
                        key=aggregate_id,
                        value=json.dumps(payload)
                    )
                    self.producer.flush()

                    # Mark as published
                    cursor.execute(
                        "UPDATE outbox SET published = true WHERE id = %s",
                        (event_id,)
                    )
                    self.db.commit()

                except Exception as e:
                    print(f"Failed to publish {event_id}: {e}")
                    # Will retry on next iteration

            time.sleep(1)  # Poll every second

Outbox Table Schema

CREATE TABLE outbox (
    id UUID PRIMARY KEY,
    event_type VARCHAR(255) NOT NULL,
    aggregate_id VARCHAR(255) NOT NULL,
    payload JSONB NOT NULL,
    published BOOLEAN DEFAULT false,
    created_at TIMESTAMP DEFAULT NOW(),
    published_at TIMESTAMP,
    INDEX idx_unpublished (published, created_at)
);

-- Clean up old published events periodically
DELETE FROM outbox WHERE published = true AND published_at < NOW() - INTERVAL '7 days';

Event Versioning

Challenge: Events are immutable but business logic evolves.

Upcasting Pattern

class EventUpcaster:
    """Convert old event versions to new versions"""
    def upcast(self, event: dict) -> dict:
        """Convert event to latest version"""
        event_type = event['event_type']

        if event_type == "order.created.v1":
            # v1 → v2: Add currency field
            return {
                **event,
                'event_type': 'order.created.v2',
                'data': {
                    **event['data'],
                    'currency': 'USD'  # Default for old events
                }
            }

        elif event_type == "item.added.v1":
            # v1 → v2: Add discount field
            return {
                **event,
                'event_type': 'item.added.v2',
                'data': {
                    **event['data'],
                    'discount': 0.0  # Default for old events
                }
            }

        return event  # Already latest version

    def load_events_with_upcasting(self, aggregate_id: str) -> List[Event]:
        """Load and upcast events"""
        raw_events = self.event_store.load_events(aggregate_id)
        upcasted = [self.upcast(e) for e in raw_events]
        return [Event(**e) for e in upcasted]

Saga Pattern (Distributed Transactions)

Pattern: Coordinate transactions across services with compensation.

See temporal-workflows.md for comprehensive Temporal saga implementation.

Event-Based Saga (Choreography)

# Order Service: Emit event
kafka.publish('order-events', {
    'event_type': 'order.created',
    'order_id': 'ord_123',
    'customer_id': 'cus_456'
})

# Inventory Service: Listen and react
@kafka_consumer('order-events')
def handle_order_created(event):
    try:
        reserve_inventory(event['order_id'])
        kafka.publish('inventory-events', {
            'event_type': 'inventory.reserved',
            'order_id': event['order_id']
        })
    except InsufficientStockError:
        kafka.publish('inventory-events', {
            'event_type': 'inventory.reservation.failed',
            'order_id': event['order_id']
        })

# Payment Service: Listen and react
@kafka_consumer('inventory-events')
def handle_inventory_reserved(event):
    if event['event_type'] == 'inventory.reserved':
        charge_payment(event['order_id'])
        kafka.publish('payment-events', {
            'event_type': 'payment.charged',
            'order_id': event['order_id']
        })

# Order Service: Listen for completion or failure
@kafka_consumer('payment-events')
def handle_payment_charged(event):
    if event['event_type'] == 'payment.charged':
        complete_order(event['order_id'])
    elif event['event_type'] == 'payment.failed':
        # Compensate: Release inventory
        kafka.publish('compensation-events', {
            'event_type': 'inventory.release.requested',
            'order_id': event['order_id']
        })

Best Practices

1. Event Naming Convention

Domain.Entity.Action.Version

Examples:
- order.created.v1
- order.item.added.v2
- payment.charged.v1

2. Event Schema Evolution

# Include schema version in event
event = {
    'event_type': 'order.created.v2',
    'schema_version': 2,
    'data': {...}
}

# Handle multiple versions
def handle_order_created(event):
    if event['schema_version'] == 1:
        # Handle v1 schema
        pass
    elif event['schema_version'] == 2:
        # Handle v2 schema
        pass

3. Idempotent Event Handlers

@event_handler('order.created')
def handle_order_created(event):
    event_id = event['event_id']

    # Check if already processed
    if redis.exists(f'processed:{event_id}'):
        return  # Already handled

    # Process event
    create_order_projection(event)

    # Mark as processed
    redis.setex(f'processed:{event_id}', 86400, '1')

4. Event Metadata

event = {
    'event_type': 'order.created.v1',
    'event_id': str(uuid.uuid4()),
    'timestamp': datetime.utcnow().isoformat(),
    'data': {...},
    'metadata': {
        'correlation_id': 'request-123',  # Trace across services
        'causation_id': 'event-456',  # Which event caused this
        'user_id': 'user-789',  # Who triggered
        'service': 'order-service',  # Which service
        'version': '1.2.3'  # Service version
    }
}

Related Patterns

• Event Sourcing: State as event log
• CQRS: Separate read/write models
• Outbox: Atomic writes + events
• Saga: Distributed transactions with compensation
• Event Versioning: Handle schema evolution

Kafka - Event Streaming Platform

Apache Kafka is a distributed event streaming platform designed for high-throughput, fault-tolerant log aggregation and event sourcing.

Table of Contents

• When to Use Kafka
• Core Concepts
• Topics and Partitions
• Consumer Groups
• Offsets
• Python Implementation
• Producer with Confluent Kafka
• Consumer with Confluent Kafka
• Exactly-Once Semantics
• Idempotent Producer
• Transactional Producer
• Partitioning Strategies
• Key-Based Partitioning (Default)
• Custom Partitioner
• Consumer Rebalancing
• Graceful Shutdown
• Rebalance Listener
• Performance Tuning
• Producer Throughput Optimization
• Consumer Throughput Optimization
• Monitoring
• Key Metrics
• Prometheus Metrics Exporter
• Common Patterns
• Dead Letter Queue
• Event Sourcing
• Troubleshooting
• Consumer Not Receiving Messages
• Consumer Lag Growing
• Duplicate Messages
• Configuration Best Practices
• Production Producer Config
• Production Consumer Config
• Related Patterns

When to Use Kafka

Best for:

• Event streaming (500K-1M+ msg/s)
• Log aggregation across microservices
• Event sourcing / CQRS architectures
• Real-time analytics pipelines
• Long-term event retention (days/weeks)

Not ideal for:

• Simple background job queues (use Celery/BullMQ)
• Request-reply patterns (use NATS)
• Low-latency RPC (use gRPC/HTTP)

Core Concepts

Topics and Partitions

Topics are categories for messages. Partitions enable parallelism.

Topic: "orders"
├─ Partition 0: [msg1, msg4, msg7, ...]
├─ Partition 1: [msg2, msg5, msg8, ...]
└─ Partition 2: [msg3, msg6, msg9, ...]

Key points:

• Messages with same key go to same partition (ordering guarantee)
• More partitions = more parallelism
• Partition count cannot decrease (only increase)
• Recommended: Start with partitions = max expected consumers

Consumer Groups

Consumer group coordinates partition assignment across consumers.

Consumer Group: "order-processors"
├─ Consumer 1 → Partition 0, 1
├─ Consumer 2 → Partition 2, 3
└─ Consumer 3 → Partition 4, 5

Key points:

• One partition assigned to one consumer per group
• Adding consumers (up to partition count) increases parallelism
• Rebalancing occurs when consumers join/leave
• Different groups consume independently

Offsets

Offset tracks position in partition for each consumer group.

Partition 0: [0] [1] [2] [3] [4] [5] [6] [7] [8] [9]
                                    ↑
                            Consumer offset: 5

Commit strategies:

• Auto-commit (default): Periodic background commit (may lose messages on crash)
• Manual commit: Commit after processing (exactly-once semantics)

Python Implementation

Producer with Confluent Kafka

from confluent_kafka import Producer
import json
import uuid

# Producer configuration
config = {
    'bootstrap.servers': 'localhost:9092',
    'client.id': 'order-producer',
    'acks': 'all',  # Wait for all replicas (strongest guarantee)
    'compression.type': 'lz4',  # Fast compression
    'batch.size': 32768,  # 32KB batches
    'linger.ms': 10,  # Wait 10ms for batching
}

producer = Producer(config)

def delivery_callback(err, msg):
    """Called when message delivered or failed"""
    if err:
        print(f'Delivery failed: {err}')
    else:
        print(f'Message delivered to {msg.topic()} [{msg.partition()}] @ {msg.offset()}')

# Produce message with key for partitioning
order = {
    'order_id': 'ord_123',
    'customer_id': 'cus_456',
    'total': 99.99
}

producer.produce(
    topic='orders',
    key='ord_123',  # Messages with same key go to same partition
    value=json.dumps(order).encode('utf-8'),
    on_delivery=delivery_callback
)

# Flush ensures all messages sent before program exits
producer.flush()

Consumer with Confluent Kafka

from confluent_kafka import Consumer, KafkaException, KafkaError
import json

# Consumer configuration
config = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'order-processors',
    'auto.offset.reset': 'earliest',  # Start from beginning if no offset
    'enable.auto.commit': False,  # Manual commit for exactly-once
    'max.poll.interval.ms': 300000,  # 5 minutes max processing time
}

consumer = Consumer(config)
consumer.subscribe(['orders'])

try:
    while True:
        msg = consumer.poll(timeout=1.0)

        if msg is None:
            continue  # No message within timeout

        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                print(f'End of partition {msg.partition()}')
            else:
                raise KafkaException(msg.error())
        else:
            # Process message
            order = json.loads(msg.value().decode('utf-8'))
            print(f'Processing order: {order}')

            try:
                process_order(order)

                # Manual commit after successful processing
                consumer.commit(message=msg)

            except Exception as e:
                print(f'Processing failed: {e}')
                # Don't commit - message will be redelivered

finally:
    consumer.close()

Exactly-Once Semantics

Kafka supports exactly-once processing through idempotent producers and transactional writes.

Idempotent Producer

config = {
    'bootstrap.servers': 'localhost:9092',
    'enable.idempotence': True,  # Prevents duplicate messages
    'acks': 'all',
    'max.in.flight.requests.per.connection': 5,
}

producer = Producer(config)

How it works:

• Producer assigns sequence numbers to messages
• Broker detects and rejects duplicates
• Guarantees: No duplicates even on retries

Transactional Producer

from confluent_kafka import Producer

config = {
    'bootstrap.servers': 'localhost:9092',
    'transactional.id': 'order-processor-1',  # Unique per producer
    'enable.idempotence': True,
}

producer = Producer(config)
producer.init_transactions()

try:
    producer.begin_transaction()

    # Produce multiple messages atomically
    producer.produce('orders', key='ord_1', value='...')
    producer.produce('inventory', key='inv_1', value='...')
    producer.produce('notifications', key='ntf_1', value='...')

    # All messages committed together or none
    producer.commit_transaction()

except Exception as e:
    producer.abort_transaction()
    raise

Use case: Consuming from one topic, processing, producing to another (exactly-once end-to-end).

Partitioning Strategies

Key-Based Partitioning (Default)

# Messages with same key go to same partition
producer.produce(
    'orders',
    key='customer_456',  # All orders for this customer in same partition
    value=order_data
)

Guarantees: Order preserved per key.

Custom Partitioner

from confluent_kafka import Producer

def custom_partitioner(key, num_partitions):
    """Route VIP customers to partition 0"""
    if key.startswith('VIP_'):
        return 0
    else:
        # Default hash-based partitioning for others
        return hash(key) % num_partitions

config = {
    'bootstrap.servers': 'localhost:9092',
    'partitioner': custom_partitioner,
}

producer = Producer(config)

Consumer Rebalancing

When consumers join/leave, Kafka rebalances partition assignments.

Graceful Shutdown

import signal
import sys

def signal_handler(sig, frame):
    print('Shutting down gracefully...')
    consumer.close()  # Triggers rebalance, releases partitions
    sys.exit(0)

signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)

consumer = Consumer(config)
consumer.subscribe(['orders'])

while True:
    msg = consumer.poll(1.0)
    # Process messages...

Rebalance Listener

from confluent_kafka import Consumer

def on_assign(consumer, partitions):
    """Called when partitions assigned"""
    print(f'Assigned partitions: {[p.partition for p in partitions]}')

def on_revoke(consumer, partitions):
    """Called before partitions revoked"""
    print(f'Revoking partitions: {[p.partition for p in partitions]}')
    consumer.commit()  # Commit offsets before losing partitions

consumer = Consumer(config)
consumer.subscribe(['orders'], on_assign=on_assign, on_revoke=on_revoke)

Performance Tuning

Producer Throughput Optimization

config = {
    'bootstrap.servers': 'localhost:9092',
    'acks': '1',  # Wait for leader ACK only (faster than 'all')
    'compression.type': 'lz4',  # Fast compression
    'batch.size': 65536,  # 64KB batches (larger = more efficient)
    'linger.ms': 20,  # Wait 20ms for batching
    'buffer.memory': 67108864,  # 64MB send buffer
    'max.in.flight.requests.per.connection': 5,
}

Trade-offs:

• acks=1 faster but weaker durability than acks=all
• Larger batches = higher throughput, higher latency
• linger.ms adds latency but increases batching efficiency

Consumer Throughput Optimization

config = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'high-throughput-consumer',
    'fetch.min.bytes': 1048576,  # 1MB min fetch (reduce round trips)
    'fetch.max.wait.ms': 500,  # Max 500ms wait for fetch.min.bytes
    'max.partition.fetch.bytes': 2097152,  # 2MB per partition
    'session.timeout.ms': 30000,  # 30s session timeout
    'heartbeat.interval.ms': 10000,  # 10s heartbeat
}

Trade-offs:

• Larger fetch.min.bytes = fewer network calls but higher latency
• Increase max.partition.fetch.bytes for high-throughput topics
• Tune session.timeout.ms based on processing time per message

Monitoring

Key Metrics

from confluent_kafka import Consumer, TopicPartition

consumer = Consumer(config)
consumer.subscribe(['orders'])

# Get consumer lag (messages behind)
def get_consumer_lag():
    """Calculate lag for all assigned partitions"""
    assigned = consumer.assignment()
    lag = {}

    for partition in assigned:
        # Get current committed offset
        committed = consumer.committed([partition])[0].offset

        # Get high watermark (latest offset in partition)
        low, high = consumer.get_watermark_offsets(partition)

        lag[partition.partition] = high - committed

    return lag

# Print lag every 10 seconds
import time
while True:
    msg = consumer.poll(1.0)
    # Process message...

    if time.time() % 10 == 0:
        print(f'Consumer lag: {get_consumer_lag()}')

Prometheus Metrics Exporter

from prometheus_client import Counter, Histogram, Gauge

messages_consumed = Counter('kafka_messages_consumed_total', 'Total messages consumed', ['topic'])
processing_duration = Histogram('kafka_message_processing_seconds', 'Processing duration', ['topic'])
consumer_lag = Gauge('kafka_consumer_lag', 'Consumer lag', ['topic', 'partition'])

while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue

    start_time = time.time()

    # Process message
    process_order(msg.value())

    # Record metrics
    messages_consumed.labels(topic=msg.topic()).inc()
    processing_duration.labels(topic=msg.topic()).observe(time.time() - start_time)

    consumer.commit(message=msg)

Common Patterns

Dead Letter Queue

def process_with_dlq(msg):
    """Process message with DLQ fallback"""
    try:
        process_order(msg.value())
        consumer.commit(message=msg)
    except RecoverableError as e:
        # Retry by not committing (message will be redelivered)
        print(f'Recoverable error, will retry: {e}')
    except UnrecoverableError as e:
        # Send to DLQ
        dlq_producer.produce(
            'orders-dlq',
            key=msg.key(),
            value=msg.value(),
            headers={'error': str(e)}
        )
        consumer.commit(message=msg)  # Don't reprocess

Event Sourcing

# Write events to Kafka as source of truth
def create_order(order_data):
    """Event sourcing: Append event to log"""
    event = {
        'event_type': 'order.created.v1',
        'event_id': str(uuid.uuid4()),
        'timestamp': datetime.utcnow().isoformat(),
        'data': order_data
    }

    producer.produce(
        'order-events',
        key=order_data['order_id'],
        value=json.dumps(event).encode('utf-8')
    )
    producer.flush()

# Rebuild state by replaying events
def rebuild_order_state(order_id):
    """Replay all events for an order"""
    consumer = Consumer({
        'bootstrap.servers': 'localhost:9092',
        'group.id': f'rebuild-{uuid.uuid4()}',  # Unique group
        'auto.offset.reset': 'earliest',
    })

    consumer.subscribe(['order-events'])

    order_state = {}
    while True:
        msg = consumer.poll(1.0)
        if msg is None:
            break

        if msg.key().decode('utf-8') == order_id:
            event = json.loads(msg.value().decode('utf-8'))
            apply_event(order_state, event)

    return order_state

Troubleshooting

Consumer Not Receiving Messages

Check: 1. Topic exists: kafka-topics --list --bootstrap-server localhost:9092 2. Messages in topic: kafka-console-consumer --topic orders --from-beginning 3. Consumer group offset: kafka-consumer-groups --describe --group order-processors

Consumer Lag Growing

Causes:

• Processing too slow (scale horizontally by adding consumers)
• Too few partitions (repartition topic)
• Network issues (check broker health)

Fix:

# Scale by adding more consumers (up to partition count)
# If already at partition limit, repartition topic:
# kafka-topics --alter --topic orders --partitions 10

Duplicate Messages

Causes:

• Consumer crashed before committing offset
• Rebalancing occurred mid-processing

Fix: Implement idempotency:

def process_order_idempotent(order_id, data):
    """Idempotent processing with deduplication"""
    if redis.exists(f'processed:{order_id}'):
        return 'already_processed'

    result = process_order(data)
    redis.setex(f'processed:{order_id}', 86400, '1')  # 24h TTL
    return result

Configuration Best Practices

Production Producer Config

config = {
    'bootstrap.servers': 'broker1:9092,broker2:9092,broker3:9092',
    'client.id': 'order-producer',
    'acks': 'all',  # Strongest durability
    'retries': 2147483647,  # Max retries
    'max.in.flight.requests.per.connection': 5,
    'enable.idempotence': True,  # Prevent duplicates
    'compression.type': 'lz4',
    'batch.size': 32768,
    'linger.ms': 10,
}

Production Consumer Config

config = {
    'bootstrap.servers': 'broker1:9092,broker2:9092,broker3:9092',
    'group.id': 'order-processors',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': False,  # Manual commit
    'max.poll.interval.ms': 300000,  # 5 minutes
    'session.timeout.ms': 30000,  # 30 seconds
    'heartbeat.interval.ms': 10000,  # 10 seconds
    'isolation.level': 'read_committed',  # For transactional producers
}

Related Patterns

• CQRS: Command topics write events, query topics build read models
• Event Sourcing: Kafka as immutable event log
• Outbox Pattern: Ensure database writes and event publishing are atomic
• Saga Pattern: Coordinate distributed transactions via events (see Temporal integration)