paulpas Event-Driven Architecture Implementation
Senior distributed systems engineer designing asynchronous event processing pipelines using pub/sub messaging and event streaming. This skill makes the model reason about data flow as events rather than synchronous requests, focusing on decoupling producers from consumers through reliable message delivery, schema evolution, and idempotent processing semantics.
TL;DR Checklist
- Design event taxonomy with clear naming conventions:
domain.action.resource(e.g.,order.created,payment.refunded) - Select message broker based on throughput needs: Kafka for high-throughput replayable streams, RabbitMQ for complex routing and lower latency
- Implement schema registry with version compatibility checks before publishing any new event format
- Use the database outbox pattern for every write that must produce an event — never publish events from application code directly
- Make every consumer idempotent using message IDs as deduplication keys enforced by unique database constraints
- Configure dead letter queues with retry count tracking and poison pill detection for all consumers
When to Use
Use this skill when:
- Decoupling service communication so producers don't need to know about consumers or their availability
- Building real-time analytics pipelines that aggregate events from multiple sources into dashboards and reports
- Implementing CQRS read projections that derive state from a stream of domain events
- Designing audit logging systems that must capture every state change immutably
- Propagating domain events across bounded contexts where services must react to changes without direct coupling
When NOT to Use
Avoid this skill for:
- Simple CRUD applications with synchronous read/write patterns and no need for event-driven workflows
- Low-throughput operations (<100 events per second) where the overhead of a message broker exceeds the benefit
- Regulated systems requiring strict synchronous ordering guarantees that pub/sub inherently cannot provide
- Scenarios where the consumer must wait for processing to complete before returning a response — use REST/gRPC instead
Core Workflow
Design Event Taxonomy — Establish a naming convention and event catalog. Every event must represent something meaningful in the domain language:
domain.action.resource(e.g.,order.created,payment.refunded,inventory.reserved). Classify each event as either a domain event (business-relevant state change) or an infrastructure event (system-level notification like health checks). Document the schema for every event type in the catalog. Checkpoint: Every event in the taxonomy must have a clearly identified producer, at least one consumer, and a defined schema version strategy. Events without consumers are technical debt.Select Message Broker — Match broker technology to requirements:
- Kafka: High-throughput streams (10k+ events/sec), replayability, long retention periods, event sourcing
- RabbitMQ: Complex routing patterns (topic exchanges, header exchanges), lower latency (<1ms), dead letter queuing
- AWS SQS / Azure Service Bus: Managed cloud-native, simple APIs, good for distributed systems in those clouds Checkpoint: The broker must support at-least-once delivery semantics. Exactly-once requires application-level idempotency (see Pattern 2).
Set Up Schema Registry — Deploy a schema registry (e.g., Confluent Schema Registry, Apicurio) and configure compatibility modes:
BACKWARD: New consumers can read data produced with the latest schemaFULL: Both old and new consumers can read both old and new data formatsFORWARD: Old consumers can read data produced with the newer schema Register every event type before any producer starts publishing. Validate schema compatibility on every deployment. Checkpoint: No producer may publish events without a registered schema version. CI/CD pipelines must validate schema compatibility before deploying new event formats.
Implement Outbox Pattern — For every service that writes data and publishes events, implement the transactional outbox: write domain data and outbox entries atomically within the same database transaction. A CDC (Change Data Capture) worker then reads the outbox table and publishes events to the message broker. This guarantees zero event loss. Checkpoint: The outbox insert must happen in the same DB transaction as the business data write. Never use two-phase commit — it defeats the purpose of microservices.
Construct Idempotent Consumer — Every consumer must handle duplicate messages gracefully. Use message IDs (provided by the broker) as unique constraints in a
processed_eventstracking table. Before processing any event, check if its ID already exists. If it does, skip processing and acknowledge the message. Checkpoint: The deduplication check + insert must be a single atomic operation using an INSERT ... ON CONFLICT DO NOTHING pattern or equivalent.Build Dead Letter Queue Pipeline — Configure dead letter queues (DLQ) for all consumer groups. Implement retry logic with exponential backoff before routing to the DLQ. Track retry counts and implement poison pill detection — messages that fail consistently should be quarantined rather than retried indefinitely. Checkpoint: Every message routed to a DLQ must trigger an alert. DLQ messages must not be automatically re-processed without manual review and investigation.
Implementation Patterns
Pattern 1: Database Outbox Pattern with CDC
The transactional outbox pattern guarantees that business data changes and event publishing are atomic — either both succeed or neither does. A CDC worker reads the outbox table and publishes events to the message broker without modifying the source database's transaction log.
# outbox/outbox_engine.py — Transactional outbox with Change Data Capture worker
from __future__ import annotations
import asyncio
import json
import logging
import time
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Any
logger = logging.getLogger(__name__)
# --- Outbox message model ---
class OutboxStatus(Enum):
PENDING = "pending"
PUBLISHED = "published"
FAILED = "failed"
@dataclass
class OutboxEntry:
"""Represents a single outbox record in the database."""
id: str
aggregate_id: str # ID of the domain entity that caused this event
aggregate_type: str # e.g., "Order", "Payment"
event_type: str # e.g., "order.created", "payment.processed"
event_data: dict[str, Any] # Event payload as JSON-serializable dict
metadata: dict[str, Any] = field(default_factory=dict)
status: OutboxStatus = OutboxStatus.PENDING
created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
published_at: datetime | None = None
error_message: str | None = None
# --- SQL schema for the outbox table ---
OUTBOX_SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS outbox (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
aggregate_id VARCHAR(255) NOT NULL,
aggregate_type VARCHAR(100) NOT NULL,
event_type VARCHAR(255) NOT NULL,
event_data JSONB NOT NULL DEFAULT '{}',
metadata JSONB NOT NULL DEFAULT '{}',
status VARCHAR(20) NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'published', 'failed')),
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
published_at TIMESTAMPTZ,
error_message TEXT,
retry_count INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_outbox_status_created
ON outbox (status, created_at) WHERE status = 'pending';
"""
class OutboxRepository:
"""Database repository for outbox entries.
All methods are designed to be used within transactions alongside
business data operations, ensuring atomicity.
"""
def __init__(self, db_connection) -> None: # type: ignore[type-arg]
self._conn = db_connection
async def insert_entry(self, entry: OutboxEntry) -> str:
"""Insert an outbox entry — must be called within the same transaction as business data writes."""
query = """
INSERT INTO outbox (id, aggregate_id, aggregate_type, event_type, event_data, metadata)
VALUES ($1, $2, $3, $4, $5, $6)
"""
await self._conn.execute(
query,
entry.id,
entry.aggregate_id,
entry.aggregate_type,
entry.event_type,
json.dumps(entry.event_data),
json.dumps(entry.metadata),
)
logger.info(
"Outbox entry created: %s.%s for aggregate %s",
entry.event_type, entry.id[:8], entry.aggregate_id,
)
return entry.id
async def claim_pending_entries(self, batch_size: int = 100) -> list[OutboxEntry]:
"""Atomically claim a batch of pending entries for publishing.
Uses SELECT ... FOR UPDATE SKIP LOCKED to allow multiple workers
to safely consume from the same outbox table without contention.
"""
query = """
UPDATE outbox
SET status = $1
WHERE id IN (
SELECT id FROM outbox
WHERE status = 'pending'
ORDER BY created_at ASC
LIMIT $2
FOR UPDATE SKIP LOCKED
)
RETURNING *
"""
rows = await self._conn.fetch(query, "published", batch_size)
entries = []
for row in rows:
entries.append(OutboxEntry(
id=str(row["id"]),
aggregate_id=row["aggregate_id"],
aggregate_type=row["aggregate_type"],
event_type=row["event_type"],
event_data=json.loads(row["event_data"]) if isinstance(row["event_data"], str) else row["event_data"],
metadata=json.loads(row["metadata"]) if isinstance(row["metadata"], str) else row["metadata"],
status=OutboxStatus.PUBLISHED,
created_at=row["created_at"],
))
return entries
async def mark_failed(self, entry_id: str, error_message: str, retry_count: int) -> None:
"""Mark an entry as failed with retry tracking."""
query = """
UPDATE outbox
SET status = 'failed',
error_message = $1,
retry_count = $2,
published_at = now()
WHERE id = $3
"""
await self._conn.execute(query, error_message, retry_count + 1, entry_id)
class EventPublisher(Protocol):
"""Protocol for publishing events to a message broker."""
async def publish(self, event_type: str, data: dict[str, Any], aggregate_id: str) -> None: ...
class OutboxWorker:
"""Background worker that reads outbox entries and publishes events to the broker.
Runs as a long-lived process (or daemon thread) that continuously polls
the outbox table for pending entries, publishes them, and updates their status.
Retry strategy:
- Exponential backoff with jitter: base_delay * 2^retry_count * random(0.5, 1.5)
- Max retries: configurable (default 5)
- On max retries exceeded: entry is marked FAILED and routed to DLQ
"""
def __init__(
self,
outbox_repo: OutboxRepository,
event_publisher: EventPublisher,
poll_interval_ms: int = 500,
batch_size: int = 100,
max_retries: int = 5,
base_retry_delay_seconds: float = 1.0,
) -> None:
self._repo = outbox_repo
self._publisher = event_publisher
self._poll_interval_ms = poll_interval_ms
self._batch_size = batch_size
self._max_retries = max_retries
self._base_retry_delay = base_retry_delay_seconds
self._running = False
async def start(self) -> None:
"""Start the outbox worker — runs until cancelled."""
self._running = True
logger.info("Outbox worker started (poll=%dms, batch=%d)", self._poll_interval_ms, self._batch_size)
while self._running:
try:
entries = await self._repo.claim_pending_entries(batch_size=self._batch_size)
if not entries:
await asyncio.sleep(self._poll_interval_ms / 1000.0)
continue
for entry in entries:
await self._process_entry(entry)
except Exception as exc:
logger.exception("Outbox worker error during poll cycle: %s", exc)
await asyncio.sleep(2.0) # Brief pause on unexpected errors
async def _process_entry(self, entry: OutboxEntry) -> None:
"""Publish a single outbox entry with retry logic."""
try:
await self._publisher.publish(entry.event_type, entry.event_data, entry.aggregate_id)
logger.info("Published event %s for aggregate %s", entry.event_type, entry.aggregate_id)
except Exception as exc:
retry_count = 0
# Check if we already have a retry count from the DB (from previous attempts)
while retry_count < self._max_retries:
delay = self._base_retry_delay * (2 ** retry_count)
jitter = delay * (0.5 + 0.5 * __import__("random").random())
logger.warning(
"Publish failed for %s (attempt %d/%d), retrying in %.1fs",
entry.event_type, retry_count + 1, self._max_retries, jitter,
)
await asyncio.sleep(jitter)
try:
await self._publisher.publish(entry.event_type, entry.event_data, entry.aggregate_id)
logger.info("Retrieved event %s for aggregate %s after retry %d", entry.event_type, entry.aggregate_id, retry_count + 1)
return # Success — exit retry loop
except Exception:
retry_count += 1
# Max retries exceeded — mark as failed
await self._repo.mark_failed(entry.id, f"Max retries ({self._max_retries}) exceeded", retry_count - 1)
logger.critical(
"Outbox entry %s FAILED after %d retries — routed to DLQ: %s",
entry.id[:8], self._max_retries, entry.event_type,
)
async def stop(self) -> None:
self._running = False
logger.info("Outbox worker stopped")
# --- Concrete example: Publishing from a business transaction ---
async def create_order_with_events(
db_conn, # type: ignore[type-arg]
order_data: dict[str, Any],
) -> str:
"""Create an order and enqueue the event atomically via outbox.
This is called within a database transaction:
BEGIN;
INSERT INTO orders ... ;
INSERT INTO outbox ... ; ← same transaction!
COMMIT;
The outbox worker later reads from the outbox table and publishes to Kafka.
"""
from contextlib import asynccontextmanager
# Create order business data
order_id = str(uuid.uuid4())
await db_conn.execute(
"INSERT INTO orders (id, customer_id, total) VALUES ($1, $2, $3)",
order_id,
order_data["customer_id"],
order_data["total"],
)
# Insert outbox entry in the SAME transaction — atomic with the business write
outbox_entry = OutboxEntry(
id=str(uuid.uuid4()),
aggregate_id=order_id,
aggregate_type="Order",
event_type="order.created",
event_data={
"order_id": order_id,
"customer_id": order_data["customer_id"],
"items": order_data["items"],
"total": order_data["total"],
},
metadata={"source": "order-service", "version": "1.0.0"},
)
outbox_repo = OutboxRepository(db_conn)
await outbox_repo.insert_entry(outbox_entry)
logger.info("Order %s created with outbox entry (same transaction)", order_id)
return order_id
Pattern 2: Idempotent Consumer
Idempotency prevents duplicate processing when the broker delivers messages more than once. This pattern uses message IDs as unique database constraints to deduplicate events at the consumer level.
# consumers/idempotency.py — Idempotent event consumer decorator and base class
from __future__ import annotations
import functools
import logging
import time
import uuid
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Awaitable, Callable, Protocol
logger = logging.getLogger(__name__)
# --- Deduplication tracking model ---
@dataclass
class ProcessedEvent:
"""Tracks which events have already been processed to prevent duplicate work."""
message_id: str # Unique ID from the event/message header
consumer_group: str # Consumer group name — dedup is per-group
consumed_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
# --- Protocol for idempotency storage ---
class IdempotencyStore(Protocol):
"""Storage backend for tracking processed message IDs."""
async def mark_processed(self, message_id: str, consumer_group: str) -> bool:
"""Mark a message as processed. Returns True if newly inserted, False if already existed."""
...
async def is_processed(self, message_id: str, consumer_group: str) -> bool:
"""Check if a message has already been processed."""
...
class DatabaseIdempotencyStore:
"""PostgreSQL-backed idempotency store using unique constraints for deduplication.
Uses INSERT ... ON CONFLICT DO NOTHING to atomically check-and-insert,
preventing race conditions when multiple consumer instances process the same event.
"""
SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS processed_events (
message_id VARCHAR(255) NOT NULL,
consumer_group VARCHAR(255) NOT NULL,
consumed_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (message_id, consumer_group)
);
-- Purge old records to prevent table bloat — keep last N days
CREATE OR REPLACE FUNCTION purge_processed_events(days_keep INTEGER)
RETURNS VOID AS $$
DELETE FROM processed_events
WHERE consumed_at < now() - (days_keep || ' days')::INTERVAL;
$$ LANGUAGE SQL;
"""
def __init__(self, db_connection) -> None: # type: ignore[type-arg]
self._conn = db_connection
self._days_to_keep = 7 # Dedup window — adjust based on broker retention
async def mark_processed(self, message_id: str, consumer_group: str) -> bool:
"""Atomically check if processed and mark as new. Returns True if this is a new event."""
query = """
INSERT INTO processed_events (message_id, consumer_group, consumed_at)
VALUES ($1, $2, now())
ON CONFLICT (message_id, consumer_group) DO NOTHING
RETURNING consumed_at;
"""
try:
row = await self._conn.fetchrow(query, message_id, consumer_group)
return row is not None # New insert succeeded → True
except Exception as exc:
logger.error("Idempotency store error: %s", exc)
raise
async def is_processed(self, message_id: str, consumer_group: str) -> bool:
"""Fast-path check without inserting."""
query = "SELECT 1 FROM processed_events WHERE message_id = $1 AND consumer_group = $2 LIMIT 1"
row = await self._conn.fetchrow(query, message_id, consumer_group)
return row is not None
async def purge_old_records(self) -> int:
"""Remove processed event records older than the retention window."""
result = await self._conn.execute(
"SELECT purge_processed_events($1)", self._days_to_keep
)
logger.info("Purged %d old idempotency records", result or 0)
return result
# --- Idempotent consumer base class ---
@dataclass
class EventEnvelope:
"""Wraps the raw message with metadata for consistent processing."""
body: dict[str, Any]
message_id: str = field(default_factory=lambda: uuid.uuid4().hex[:16])
event_type: str = ""
timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
consumer_group: str = "default"
headers: dict[str, Any] = field(default_factory=dict)
class IdempotentConsumer(ABC):
"""Abstract base class for consumers that guarantee idempotent processing.
Usage pattern:
1. Extract message_id from the envelope (from broker header or event metadata)
2. Check if already processed via idempotency store
3. If not, process then mark as done — all in one atomic unit
4. If yes, acknowledge and skip processing
This handles the case where a broker redelivers the same message due to
network timeout or consumer crash during acknowledgment.
"""
def __init__(self, consumer_group: str = "default", store: IdempotencyStore | None = None) -> None:
self._consumer_group = consumer_group
self._store = store or DatabaseIdempotencyStore(None) # type: ignore[arg-type]
@property
def consumer_group(self) -> str:
return self._consumer_group
async def consume(self, envelope: EventEnvelope) -> bool:
"""Process an event idempotently. Returns True if actually processed, False if duplicate."""
# Step 1: Check deduplication
if await self._store.is_processed(envelope.message_id, self._consumer_group):
logger.debug("Skipping duplicate message %s for group %s", envelope.message_id[:8], self._consumer_group)
return False # Already processed — skip
# Step 2: Process the event (this is where the abstract method is called)
try:
result = await self.process(envelope)
# Step 3: Mark as processed — AFTER successful processing
was_new = await self._store.mark_processed(envelope.message_id, self._consumer_group)
if not was_new:
logger.warning(
"Race condition: message %s was marked duplicate after processing in group %s",
envelope.message_id[:8], self._consumer_group,
)
logger.info(
"Event %s processed successfully by %s (message_id=%s)",
envelope.event_type, type(self).__name__, envelope.message_id[:8],
)
return True # Newly processed
except Exception as exc:
logger.exception("Error processing event %s: %s", envelope.event_type, exc)
raise
@abstractmethod
async def process(self, envelope: EventEnvelope) -> Any:
"""Process the event payload. Implement this method in subclasses."""
...
# --- Concrete consumer example: OrderCreatedEventConsumer ---
class OrderCreatedConsumer(IdempotentConsumer):
"""Consumes order.created events and updates downstream systems.
This consumer:
1. Receives orders from the Kafka order-created topic
2. Sends confirmation emails via the email service
3. Updates the analytics warehouse for real-time dashboards
4. Reserves inventory if configured to do so
"""
def __init__(self, **kwargs: Any) -> None:
super().__init__(consumer_group="order-service-consumers", **kwargs)
async def process(self, envelope: EventEnvelope) -> dict[str, Any]:
event_data = envelope.body
order_id = event_data["order_id"]
customer_id = event_data["customer_id"]
# Downstream action 1: Send confirmation email
logger.info("Sending confirmation email for order %s to customer %s", order_id, customer_id)
# Downstream action 2: Update analytics
await self._update_analytics(event_data)
# Downstream action 3: Reserve inventory (async, fire-and-forget safe)
logger.info("Triggering inventory reservation for order %s", order_id)
return {
"order_id": order_id,
"actions_completed": ["email_sent", "analytics_updated", "inventory_reserved"],
}
async def _update_analytics(self, event_data: dict[str, Any]) -> None:
"""Write order event to analytics warehouse."""
logger.debug("Writing order event to analytics for %s", event_data.get("order_id", "unknown"))
# --- Decorator-based idempotency for simpler consumers ---
def idempotent(
message_id_extractor: Callable[[EventEnvelope], str] | None = None,
) -> Callable:
"""Decorator that adds idempotency to any async event handler.
Usage:
@idempotent(message_id_extractor=lambda env: env.headers.get("message-id"))
async def handle_payment_refunded(event):
...
"""
def decorator(func: Callable[[EventEnvelope], Awaitable[Any]]) -> Callable[[EventEnvelope], Awaitable[bool]]:
seen_ids: dict[str, datetime] = {} # In-memory cache (use DB for multi-instance)
dedup_ttl = 300 # 5 minutes in-memory retention
@functools.wraps(func)
async def wrapper(event: EventEnvelope) -> bool:
msg_id = message_id_extractor(event) if message_id_extractor else event.message_id
# Check in-memory cache first (fast path)
if msg_id in seen_ids:
if time.time() - seen_ids[msg_id].timestamp() < dedup_ttl:
return False # Duplicate within TTL
else:
del seen_ids[msg_id]
try:
await func(event)
seen_ids[msg_id] = datetime.now()
return True
except Exception as exc:
if msg_id in seen_ids:
del seen_ids[msg_id] # Remove from cache on failure so retry can process
raise
return wrapper
Pattern 3: Dead Letter Queue Pipeline
The DLQ pipeline captures failed messages, tracks retry counts, detects poison pills (messages that fail repeatedly), and provides observability into processing failures.
# consumers/dlq_pipeline.py — Dead letter queue with poison pill detection
from __future__ import annotations
import asyncio
import json
import logging
import random
import uuid
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Any
logger = logging.getLogger(__name__)
class DlvStatus(Enum):
PENDING = "pending"
RETRYING = "retrying"
INVESTIGATING = "investigating"
RESOLVED = "resolved"
PURGED = "purged"
@dataclass
class PoisonPillRecord:
"""Tracks messages that consistently fail — potential data corruption or bugs."""
event_type: str
consumer_group: str
failure_count: int
first_failure: datetime
last_failure: datetime
sample_error: str
sample_message_id: str
class RetryPolicy:
"""Configurable retry strategy with exponential backoff and jitter."""
def __init__(
self,
max_retries: int = 3,
base_delay_seconds: float = 1.0,
max_delay_seconds: float = 60.0,
jitter_factor: float = 0.5,
) -> None:
self.max_retries = max_retries
self.base_delay = base_delay_seconds
self.max_delay = max_delay_seconds
self.jitter_factor = jitter_factor
def calculate_delay(self, attempt: int) -> float:
"""Calculate delay for a given retry attempt with exponential backoff and random jitter."""
raw_delay = min(
self.base_delay * (2 ** attempt),
self.max_delay,
)
jitter = raw_delay * self.jitter_factor * (random.random() - 0.5) * 2
return max(0.1, raw_delay + jitter) # Minimum 100ms delay
@dataclass
class FailedMessage:
"""Represents a message that failed processing and is queued for retry or DLQ."""
original_message_id: str
event_type: str
consumer_group: str
payload: dict[str, Any]
error_message: str
retry_count: int = 0
created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
last_retry_at: datetime | None = None
status: DlvStatus = DlvStatus.PENDING
dlq_routed: bool = False
class PoisonPillDetector:
"""Detects poison pills — messages that fail consistently, indicating a data or code issue.
Detection strategy:
- Track failure counts per (event_type, consumer_group) pair
- If a pair exceeds the threshold within the time window, it's flagged as a poison pill
- Poison pills bypass retry and go directly to INVESTIGATING status
"""
def __init__(self, failure_threshold: int = 5, window_seconds: float = 300.0) -> None:
self.failure_threshold = failure_threshold
self.window_seconds = window_seconds
# Keyed by (event_type, consumer_group)
self._failure_counts: dict[tuple[str, str], list[datetime]] = defaultdict(list)
def record_failure(self, event_type: str, consumer_group: str, error: str, message_id: str) -> PoisonPillRecord | None:
"""Record a failure and check if this constitutes a poison pill."""
key = (event_type, consumer_group)
now = datetime.now(timezone.utc)
# Clean old failures outside the window
cutoff = now.timestamp() - self.window_seconds
self._failure_counts[key] = [
ts for ts in self._failure_counts[key] if ts.timestamp() > cutoff
]
self._failure_counts[key].append(now)
if len(self._failure_counts[key]) >= self.failure_threshold:
# Flag as poison pill
recent_failures = self._failure_counts[key][-self.failure_threshold:]
return PoisonPillRecord(
event_type=event_type,
consumer_group=consumer_group,
failure_count=len(self._failure_counts[key]),
first_failure=recent_failures[0],
last_failure=now,
sample_error=error[:500],
sample_message_id=message_id[:16],
)
return None
def reset(self, event_type: str, consumer_group: str) -> None:
"""Reset the failure counter when a message succeeds or is resolved."""
key = (event_type, consumer_group)
if key in self._failure_counts:
del self._failure_counts[key]
class DeadLetterQueuePipeline:
"""Manages failed messages with retry logic, DLQ routing, and poison pill detection.
Flow for a failed message:
1. Increment retry_count
2. If retry_count < max_retries → schedule retry with exponential backoff
3. If retry_count >= max_retries → route to permanent DLQ
4. Poison pill detector monitors for repeated failures on the same event type
This prevents a single bad message from consuming resources indefinitely
while still giving transient failures a chance to succeed on retry.
"""
def __init__(
self,
max_retries: int = 3,
poison_threshold: int = 5,
poison_window_seconds: float = 300.0,
) -> None:
self.retry_policy = RetryPolicy(max_retries=max_retries)
self.poison_detector = PoisonPillDetector(
failure_threshold=poison_threshold,
window_seconds=poison_window_seconds,
)
self._pending: list[FailedMessage] = []
self._dlq: list[FailedMessage] = []
def route_message(self, failed_msg: FailedMessage) -> str:
"""Route a failed message to retry queue or permanent DLQ.
Returns the routing decision: 'retry', 'dlq', or 'investigate'.
"""
# Check for poison pill first
poison = self.poison_detector.record_failure(
failed_msg.event_type,
failed_msg.consumer_group,
failed_msg.error_message,
failed_msg.original_message_id,
)
if poison:
logger.critical(
"POISON PILLOT DETECTED: %s in group %s (%d failures in %.0fs) — routing to INVESTIGATING",
failed_msg.event_type, failed_msg.consumer_group,
poison.failure_count, self.poison_window_seconds,
)
failed_msg.status = DlvStatus.INVESTIGATING
self._pending.append(failed_msg)
return "investigate"
if failed_msg.retry_count >= self.retry_policy.max_retries:
# Max retries exceeded — route to permanent DLQ
failed_msg.dlq_routed = True
failed_msg.status = DlvStatus.PENDING
self._dlq.append(failed_msg)
logger.warning(
"Message %s routed to DLQ after %d retries (event=%s, group=%s)",
failed_msg.original_message_id[:8],
failed_msg.retry_count,
failed_msg.event_type,
failed_msg.consumer_group,
)
return "dlq"
# Schedule retry with exponential backoff
failed_msg.retry_count += 1
failed_msg.last_retry_at = datetime.now(timezone.utc)
failed_msg.status = DlvStatus.RETRYING
delay = self.retry_policy.calculate_delay(failed_msg.retry_count - 1)
logger.info(
"Scheduling retry %d/%d for message %s (event=%s, delay=%.1fs)",
failed_msg.retry_count, self.retry_policy.max_retries,
failed_msg.original_message_id[:8],
failed_msg.event_type,
delay,
)
# Schedule the actual retry asynchronously
asyncio.create_task(self._retry_after_delay(failed_msg, delay))
return "retry"
async def _retry_after_delay(self, message: FailedMessage, delay: float) -> None:
"""Wait for the calculated backoff period then signal that the message is ready to retry."""
await asyncio.sleep(delay)
message.status = DlvStatus.PENDING
logger.debug("Retry message %s is now pending", message.original_message_id[:8])
def get_dlq_messages(self, limit: int = 100) -> list[FailedMessage]:
"""Retrieve messages from the DLQ for inspection/reprocessing."""
return self._dlq[:limit]
def get_pending_retries(self) -> list[FailedMessage]:
"""Get messages scheduled for retry that are ready to be reprocessed."""
return [m for m in self._pending if m.status == DlvStatus.PENDING]
def acknowledge_success(self, message: FailedMessage) -> None:
"""Called when a retried message succeeds — reset poison pill counter."""
self.poison_detector.reset(message.event_type, message.consumer_group)
message.status = DlvStatus.RESOLVED
# --- Example: Using the DLQ pipeline with an event consumer ---
class ResilientEventConsumer:
"""Wraps a standard consumer with DLQ-based resilience."""
def __init__(self, dlq_pipeline: DeadLetterQueuePipeline) -> None:
self._dlq = dlq_pipeline
async def consume(self, event_type: str, payload: dict[str, Any], consumer_group: str, message_id: str) -> bool:
"""Try to process an event. On failure, route through the DLQ pipeline."""
try:
await self._process_event(event_type, payload)
return True
except Exception as exc:
failed = FailedMessage(
original_message_id=message_id,
event_type=event_type,
consumer_group=consumer_group,
payload=payload,
error_message=str(exc),
)
routing = self._dlq.route_message(failed)
if routing == "dlq":
logger.error("Message permanently dead-lettered: %s", message_id[:8])
elif routing == "investigate":
logger.critical("Poison pill detected for event type: %s", event_type)
return False
async def _process_event(self, event_type: str, payload: dict[str, Any]) -> None:
"""Core processing logic — raises on any failure."""
logger.info("Processing event %s", event_type)
# ... actual business logic here ...
Pattern 4: Schema Registry Integration
Schema registry integration ensures all producers and consumers agree on event formats. This pattern shows JSON Schema validation with backward/forward compatibility checking for schema evolution.
# schema/registry.py — Schema registry client with version compatibility checking
from __future__ import annotations
import hashlib
import json
import logging
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Protocol
logger = logging.getLogger(__name__)
class CompatibilityLevel(Enum):
BACKWARD = "BACKWARD" # New schema can read old data
FORWARD = "FORWARD" # Old schema can read new data
FULL = "FULL" # Both directions compatible
NONE = "NONE" # No compatibility enforced
@dataclass
class SchemaVersion:
"""Represents a registered schema version in the registry."""
subject: str # e.g., "order-created-value"
version: int # Numeric version (auto-incremented by registry)
schema_id: str # Content-based hash for deduplication
schema: dict[str, Any] # The actual JSON Schema definition
compatibility: CompatibilityLevel = CompatibilityLevel.FULL
registered_at: datetime = field(default_factory=time.time) # type: ignore[arg-type]
class SchemaRegistryClient(Protocol):
"""Protocol for interacting with a schema registry service."""
async def register_schema(self, subject: str, schema: dict[str, Any], compatibility: CompatibilityLevel) -> int: ...
async def get_latest_version(self, subject: str) -> SchemaVersion | None: ...
async def validate_compatibility(self, subject: str, new_schema: dict[str, Any]) -> bool: ...
async def resolve_schema_id(self, schema: dict[str, Any]) -> str: ...
class LocalSchemaRegistryClient:
"""In-memory schema registry implementation for development and testing.
In production, replace this with a real registry client (Confluent Schema Registry,
Apicurio Registry, or AWS Glue Schema Registry) that exposes an HTTP API.
"""
def __init__(self, default_compatibility: CompatibilityLevel = CompatibilityLevel.FULL) -> None:
self._default_compatibility = default_compatibility
self._schemas: dict[str, list[SchemaVersion]] = {} # subject → versions
self._schema_ids: dict[str, SchemaVersion] = {} # schema_id → version
async def register_schema(self, subject: str, schema: dict[str, Any], compatibility: CompatibilityLevel | None = None) -> int:
"""Register a new schema version. Returns the new version number."""
compat = compatibility or self._default_compatibility
# Check if exact schema already exists (deduplication by hash)
schema_id = self._compute_schema_hash(schema)
if schema_id in self._schema_ids:
existing = self._schema_ids[schema_id]
if existing.subject == subject:
logger.info("Schema already registered for subject %s (version %d)", subject, existing.version)
return existing.version
# Check compatibility against latest version
latest = await self.get_latest_version(subject)
if latest is not None and compat != CompatibilityLevel.NONE:
compatible = await self._check_compatibility(latest.schema, schema, compat)
if not compatible:
raise ValueError(
f"Incompatibl
…(truncated)