# Event Driven Architecture > Implements event-driven architecture patterns including pub/sub messaging with RabbitMQ and Kafka, schema registry management, dead letter queue handling, idempotent consumer design, database outbox pattern, and event streaming for asynchronous system coordination. - Skill: `paulpas/event-driven-architecture` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/event-driven-architecture` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/event-driven-architecture/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/event-driven-architecture --- # 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 1. **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. 2. **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). 3. **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 schema - `FULL`: Both old and new consumers can read both old and new data formats - `FORWARD`: 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. 4. **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. 5. **Construct Idempotent Consumer** — Every consumer must handle duplicate messages gracefully. Use message IDs (provided by the broker) as unique constraints in a `processed_events` tracking 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. 6. **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. ```python # 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. ```python # 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. ```python # 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. ```python # 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)