Implementing Dead Letter Queues and Poison Message Handling in Data Pipelines

Introduction

Picture this: it's 2 AM, your on-call phone buzzes, and you stumble over to find that your order processing pipeline has completely stopped. The culprit? A single malformed JSON payload from a third-party vendor that your consumer couldn't parse. Because your pipeline had no failure isolation strategy, that one bad message caused the entire consumer to crash, which caused messages to pile up, which caused downstream services to time out. By the time you're fully awake, you're looking at six hours of backlog and a very unhappy engineering manager.

This is the exact problem that dead letter queues (DLQs) and poison message handling were designed to solve. A poison message is any message that causes a consumer to fail repeatedly — it could be malformed data, a schema mismatch, a message that triggers an unhandled edge case in your business logic, or data that's simply too large to process. A dead letter queue is the safety net you build to catch those messages gracefully instead of letting them bring down your pipeline. Done well, DLQs give you failure isolation, observability into what's going wrong, and the operational tools to recover without data loss.

By the end of this lesson, you'll have built a complete poison message handling system with real code you can adapt for production. We'll go end-to-end: detecting failures, routing to a DLQ, enriching dead letters with diagnostic metadata, setting up alerting, and building a replay mechanism so you can reprocess messages after fixing the underlying bug.

What you'll learn:

• How to identify and classify poison messages versus transient failures
• How to implement DLQ routing with retry logic in Python using Kafka and AWS SQS patterns
• How to enrich dead letters with diagnostic context so failures are debuggable
• How to build a replay pipeline that reprocesses DLQ messages safely
• How to monitor DLQ depth and set up alerting to catch failure spikes early

---

Prerequisites

This lesson assumes you're comfortable with:

• Python at an intermediate level (decorators, context managers, exception handling)
• Basic messaging concepts — producers, consumers, topics/queues, offsets
• Familiarity with either Kafka or SQS (we'll use both, but the patterns translate)
• Basic understanding of data pipeline architecture

You don't need hands-on experience with DLQs specifically — that's what we're building.

---

Understanding Poison Messages: Not All Failures Are Equal

Before writing a single line of code, you need to make a conceptual distinction that will drive every design decision in this lesson: the difference between a transient failure and a poison message.

A transient failure is a temporary problem. Your downstream database is momentarily unreachable. An external API is rate-limiting you. A network timeout occurred but would succeed if retried. These failures are expected in distributed systems and the correct response is to retry with backoff.

A poison message is fundamentally different. It's a message that will never succeed regardless of how many times you retry it. Retrying a poison message doesn't just waste compute — it actively harms your pipeline by blocking healthy messages from being processed. This is sometimes called head-of-line blocking, and it's one of the most insidious failure modes in streaming systems.

Here's how you can classify failures in practice:

from enum import Enum
from typing import Optional
import json

class FailureCategory(Enum):
    TRANSIENT = "transient"          # Retry immediately or with backoff
    POISON_MESSAGE = "poison_message"  # Route to DLQ immediately
    BUSINESS_RULE_VIOLATION = "business_rule"  # May need human review
    SCHEMA_VALIDATION = "schema_validation"    # Likely poison, log context

def classify_failure(exception: Exception, attempt_count: int) -> FailureCategory:
    """
    Classify a processing failure to determine the correct routing strategy.
    This logic should be tailored to your specific domain.
    """
    # Structural problems that retrying won't fix
    if isinstance(exception, (json.JSONDecodeError, UnicodeDecodeError)):
        return FailureCategory.SCHEMA_VALIDATION

    if isinstance(exception, SchemaValidationError):
        return FailureCategory.SCHEMA_VALIDATION

    if isinstance(exception, BusinessRuleViolationError):
        return FailureCategory.BUSINESS_RULE_VIOLATION

    # Connection-level issues are almost always transient
    if isinstance(exception, (ConnectionError, TimeoutError)):
        return FailureCategory.TRANSIENT

    # If we've retried many times and still failing, treat as poison
    if attempt_count >= MAX_RETRY_ATTEMPTS:
        return FailureCategory.POISON_MESSAGE

    return FailureCategory.TRANSIENT

Important: The attempt_count >= MAX_RETRY_ATTEMPTS branch is your safety valve. Even a message you initially classified as transient becomes a poison message in practice if it consistently fails. Define MAX_RETRY_ATTEMPTS based on your SLA tolerance — for most pipelines, 3-5 attempts with exponential backoff is appropriate before giving up.

The classification logic above is the policy layer of your DLQ system. Everything else — the routing, enrichment, monitoring — is mechanics. Get the policy right first.

---

Building the Core DLQ Infrastructure

We'll build our DLQ system around a pattern that works regardless of whether you're using Kafka, SQS, RabbitMQ, or Pub/Sub. The core abstraction is a DeadLetterRouter that knows how to accept a failed message and its diagnostic context, then route it somewhere appropriate.

Designing the Dead Letter Envelope

The most common mistake engineers make with DLQs is routing the raw failed message without any context. When you pull that message out of the DLQ three weeks later to investigate, you'll have no idea why it failed, how many times it was retried, or what the original error was.

A dead letter should always be wrapped in an envelope that carries full diagnostic context:

import json
import uuid
from datetime import datetime, timezone
from dataclasses import dataclass, field, asdict
from typing import Any, Dict, Optional

@dataclass
class DeadLetterEnvelope:
    """
    Wraps a failed message with full diagnostic context.
    This is what gets written to your DLQ — never the raw message alone.
    """
    # Envelope identity
    dead_letter_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    created_at: str = field(
        default_factory=lambda: datetime.now(timezone.utc).isoformat()
    )

    # Original message context
    original_message: Any = None          # The raw payload that failed
    original_topic: str = ""             # Where it came from
    original_partition: Optional[int] = None
    original_offset: Optional[int] = None
    original_message_id: Optional[str] = None

    # Failure context
    failure_reason: str = ""             # Human-readable summary
    failure_category: str = ""           # From our FailureCategory enum
    exception_type: str = ""            # e.g., "json.JSONDecodeError"
    exception_message: str = ""
    exception_traceback: str = ""

    # Retry history
    attempt_count: int = 0
    first_attempt_at: Optional[str] = None
    last_attempt_at: Optional[str] = None

    # Pipeline context (critical for debugging in microservice environments)
    pipeline_name: str = ""
    pipeline_version: str = ""
    consumer_group: str = ""
    hostname: str = ""

    # Replay tracking
    replay_count: int = 0
    last_replayed_at: Optional[str] = None
    replay_status: Optional[str] = None  # "success", "failed", "pending"

    def to_json(self) -> str:
        return json.dumps(asdict(self), default=str)

    @classmethod
    def from_json(cls, raw: str) -> "DeadLetterEnvelope":
        return cls(**json.loads(raw))

Notice the replay_count and replay_status fields. These are for when you eventually fix the bug and replay messages from the DLQ. Without them, it's very easy to accidentally replay the same message multiple times or lose track of which messages have been addressed.

Implementing the Dead Letter Router

Now let's build the router that your consumer will call when it encounters a failure:

import traceback
import socket
import logging
from typing import Protocol

logger = logging.getLogger(__name__)

class DLQBackend(Protocol):
    """
    Abstract interface for DLQ destinations.
    Implement this for Kafka, SQS, or any other backend.
    """
    def send(self, envelope: DeadLetterEnvelope) -> bool:
        ...

class DeadLetterRouter:
    def __init__(
        self,
        backend: DLQBackend,
        pipeline_name: str,
        pipeline_version: str,
        consumer_group: str,
        max_retry_attempts: int = 3,
    ):
        self.backend = backend
        self.pipeline_name = pipeline_name
        self.pipeline_version = pipeline_version
        self.consumer_group = consumer_group
        self.max_retry_attempts = max_retry_attempts
        self.hostname = socket.gethostname()

    def route_to_dlq(
        self,
        message: Any,
        exception: Exception,
        attempt_count: int,
        original_topic: str,
        original_partition: Optional[int] = None,
        original_offset: Optional[int] = None,
        original_message_id: Optional[str] = None,
        first_attempt_at: Optional[str] = None,
    ) -> bool:
        """
        Build an enriched dead letter envelope and send it to the DLQ backend.
        Returns True if the message was successfully routed, False otherwise.
        """
        category = classify_failure(exception, attempt_count)

        envelope = DeadLetterEnvelope(
            original_message=message,
            original_topic=original_topic,
            original_partition=original_partition,
            original_offset=original_offset,
            original_message_id=original_message_id,
            failure_reason=self._summarize_failure(exception, category),
            failure_category=category.value,
            exception_type=type(exception).__name__,
            exception_message=str(exception),
            exception_traceback=traceback.format_exc(),
            attempt_count=attempt_count,
            first_attempt_at=first_attempt_at,
            last_attempt_at=datetime.now(timezone.utc).isoformat(),
            pipeline_name=self.pipeline_name,
            pipeline_version=self.pipeline_version,
            consumer_group=self.consumer_group,
            hostname=self.hostname,
        )

        success = self.backend.send(envelope)

        if success:
            logger.warning(
                "Message routed to DLQ",
                extra={
                    "dead_letter_id": envelope.dead_letter_id,
                    "original_topic": original_topic,
                    "failure_category": category.value,
                    "exception_type": type(exception).__name__,
                    "attempt_count": attempt_count,
                }
            )
        else:
            # This is a critical failure — the DLQ itself is unavailable
            logger.error(
                "CRITICAL: Failed to route message to DLQ. Message may be lost.",
                extra={"original_topic": original_topic}
            )

        return success

    def _summarize_failure(
        self, exception: Exception, category: FailureCategory
    ) -> str:
        summaries = {
            FailureCategory.SCHEMA_VALIDATION: (
                f"Message failed schema validation: {type(exception).__name__}"
            ),
            FailureCategory.BUSINESS_RULE_VIOLATION: (
                f"Business rule violated: {str(exception)[:200]}"
            ),
            FailureCategory.POISON_MESSAGE: (
                f"Message exhausted {self.max_retry_attempts} retry attempts"
            ),
            FailureCategory.TRANSIENT: (
                f"Transient failure after max retries: {type(exception).__name__}"
            ),
        }
        return summaries.get(category, f"Unknown failure: {type(exception).__name__}")

Design note: The DLQBackend protocol using Protocol (structural subtyping) means you can swap backends without changing the router. You'll want this flexibility — today it might be Kafka, tomorrow SQS.

---

Kafka Implementation: Topic-Based DLQs

In Kafka-based pipelines, the standard pattern is to create a dedicated DLQ topic — usually named <original-topic>.dlq or <original-topic>.dead-letter. Let's implement the Kafka backend:

from confluent_kafka import Producer, KafkaError
from confluent_kafka.admin import AdminClient, NewTopic

class KafkaDLQBackend:
    def __init__(
        self,
        bootstrap_servers: str,
        dlq_topic: str,
        producer_config: Optional[Dict] = None,
    ):
        self.dlq_topic = dlq_topic
        config = {
            "bootstrap.servers": bootstrap_servers,
            "acks": "all",               # Wait for all replicas — no data loss
            "retries": 5,
            "retry.backoff.ms": 500,
            "enable.idempotence": True,  # Prevent duplicate DLQ entries on retry
        }
        if producer_config:
            config.update(producer_config)

        self.producer = Producer(config)

    def send(self, envelope: DeadLetterEnvelope) -> bool:
        try:
            self.producer.produce(
                topic=self.dlq_topic,
                key=envelope.dead_letter_id.encode("utf-8"),
                value=envelope.to_json().encode("utf-8"),
                headers={
                    "failure_category": envelope.failure_category,
                    "pipeline_name": envelope.pipeline_name,
                    "exception_type": envelope.exception_type,
                },
                on_delivery=self._delivery_callback,
            )
            self.producer.flush(timeout=10)
            return True
        except Exception as e:
            logger.error(f"Failed to write to Kafka DLQ: {e}")
            return False

    def _delivery_callback(self, err, msg):
        if err is not None:
            logger.error(f"DLQ delivery failed: {err}")
        else:
            logger.debug(
                f"Dead letter delivered to {msg.topic()} "
                f"partition {msg.partition()} offset {msg.offset()}"
            )

Now let's wire it into a real consumer with proper retry logic. This is the full consumer loop you'd actually run in production:

import time
from confluent_kafka import Consumer, KafkaError, Message

MAX_RETRY_ATTEMPTS = 3
RETRY_BACKOFF_SECONDS = [1, 5, 30]  # Exponential-ish backoff

def process_order_event(payload: dict) -> None:
    """Your actual business logic goes here."""
    if "order_id" not in payload:
        raise SchemaValidationError("Missing required field: order_id")
    if payload.get("total_amount", 0) < 0:
        raise BusinessRuleViolationError(
            f"Order {payload['order_id']} has negative total: {payload['total_amount']}"
        )
    # ... rest of processing

def run_consumer_with_dlq(
    bootstrap_servers: str,
    source_topic: str,
    dlq_topic: str,
    consumer_group: str,
):
    consumer = Consumer({
        "bootstrap.servers": bootstrap_servers,
        "group.id": consumer_group,
        "auto.offset.reset": "earliest",
        "enable.auto.commit": False,  # Manual commits only — critical for at-least-once
    })

    dlq_backend = KafkaDLQBackend(
        bootstrap_servers=bootstrap_servers,
        dlq_topic=dlq_topic,
    )
    dlq_router = DeadLetterRouter(
        backend=dlq_backend,
        pipeline_name="order-processing-pipeline",
        pipeline_version="2.4.1",
        consumer_group=consumer_group,
    )

    consumer.subscribe([source_topic])

    try:
        while True:
            msg = consumer.poll(timeout=1.0)
            if msg is None:
                continue
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    continue
                logger.error(f"Consumer error: {msg.error()}")
                continue

            _process_with_retry(msg, consumer, dlq_router, source_topic)

    finally:
        consumer.close()

def _process_with_retry(
    msg: Message,
    consumer: Consumer,
    dlq_router: DeadLetterRouter,
    source_topic: str,
) -> None:
    """
    Attempt to process a message with retries, routing to DLQ on final failure.
    """
    first_attempt_at = datetime.now(timezone.utc).isoformat()
    raw_value = msg.value().decode("utf-8")

    for attempt in range(1, MAX_RETRY_ATTEMPTS + 1):
        try:
            payload = json.loads(raw_value)
            process_order_event(payload)

            # Success — commit the offset
            consumer.commit(message=msg, asynchronous=False)
            logger.info(
                f"Successfully processed message at "
                f"partition={msg.partition()} offset={msg.offset()}"
            )
            return

        except (json.JSONDecodeError, SchemaValidationError) as e:
            # Poison immediately — no point retrying schema failures
            logger.warning(
                f"Poison message detected at offset {msg.offset()}: {e}"
            )
            _handle_final_failure(
                raw_value, e, attempt, msg, consumer,
                dlq_router, source_topic, first_attempt_at
            )
            return

        except Exception as e:
            category = classify_failure(e, attempt)
            if category == FailureCategory.TRANSIENT and attempt < MAX_RETRY_ATTEMPTS:
                backoff = RETRY_BACKOFF_SECONDS[min(attempt - 1, len(RETRY_BACKOFF_SECONDS) - 1)]
                logger.warning(
                    f"Transient failure on attempt {attempt}/{MAX_RETRY_ATTEMPTS}, "
                    f"retrying in {backoff}s: {e}"
                )
                time.sleep(backoff)
                continue

            # Final failure — route to DLQ
            _handle_final_failure(
                raw_value, e, attempt, msg, consumer,
                dlq_router, source_topic, first_attempt_at
            )
            return

def _handle_final_failure(
    raw_value: str,
    exception: Exception,
    attempt_count: int,
    msg: Message,
    consumer: Consumer,
    dlq_router: DeadLetterRouter,
    source_topic: str,
    first_attempt_at: str,
) -> None:
    """Route to DLQ and commit offset so we don't reprocess the poison message."""
    dlq_router.route_to_dlq(
        message=raw_value,
        exception=exception,
        attempt_count=attempt_count,
        original_topic=source_topic,
        original_partition=msg.partition(),
        original_offset=msg.offset(),
        original_message_id=None,
        first_attempt_at=first_attempt_at,
    )
    # CRITICAL: Commit even on failure, otherwise we'll process this message forever
    consumer.commit(message=msg, asynchronous=False)

The most important line in that entire consumer: consumer.commit(message=msg, asynchronous=False) after routing to the DLQ. If you don't commit the offset after a final failure, your consumer will restart and attempt the same poison message endlessly. The DLQ is where the message lives now — commit the offset and move on.

---

AWS SQS Implementation: Redrive Policies

SQS handles DLQs slightly differently through a native feature called a redrive policy. Rather than writing routing code yourself, you configure SQS to automatically move messages to a DLQ after N failed receive attempts. This is elegant, but it has tradeoffs worth understanding.

import boto3
import json
from typing import Optional

class SQSDLQBackend:
    """
    For SQS pipelines, this handles the case where you need to
    manually route to a DLQ (bypassing the native redrive policy)
    for immediate poison message detection.
    """
    def __init__(self, dlq_url: str, region_name: str = "us-east-1"):
        self.sqs = boto3.client("sqs", region_name=region_name)
        self.dlq_url = dlq_url

    def send(self, envelope: DeadLetterEnvelope) -> bool:
        try:
            self.sqs.send_message(
                QueueUrl=self.dlq_url,
                MessageBody=envelope.to_json(),
                MessageAttributes={
                    "FailureCategory": {
                        "StringValue": envelope.failure_category,
                        "DataType": "String",
                    },
                    "PipelineName": {
                        "StringValue": envelope.pipeline_name,
                        "DataType": "String",
                    },
                    "ExceptionType": {
                        "StringValue": envelope.exception_type,
                        "DataType": "String",
                    },
                    "AttemptCount": {
                        "StringValue": str(envelope.attempt_count),
                        "DataType": "Number",
                    },
                },
            )
            return True
        except Exception as e:
            logger.error(f"Failed to send to SQS DLQ: {e}")
            return False

def configure_sqs_dlq_via_terraform():
    """
    Infrastructure-as-code example for setting up SQS + DLQ with a redrive policy.
    This is Terraform HCL, shown here for reference.

    resource "aws_sqs_queue" "order_events_dlq" {
      name                      = "order-events-dlq"
      message_retention_seconds = 1209600  # 14 days — gives you time to debug
    }

    resource "aws_sqs_queue" "order_events" {
      name = "order-events"
      redrive_policy = jsonencode({
        deadLetterTargetArn = aws_sqs_queue.order_events_dlq.arn
        maxReceiveCount     = 3   # Move to DLQ after 3 failed attempts
      })
    }
    """
    pass

SQS redrive vs. manual routing: The native SQS redrive policy is great for catching failures you didn't anticipate. But it has a blind spot: it doesn't distinguish between transient and poison failures — it just counts receive attempts. For immediate poison detection (like schema validation failures), you still want manual routing as we showed in the Kafka example. The two approaches complement each other.

---

Building the Replay Pipeline

A DLQ without a replay mechanism is a dead end. The whole point is to fix the bug, then reprocess the messages that failed. Here's a production-ready replay system:

class DLQReplayEngine:
    """
    Replays messages from a DLQ back through the original processing pipeline.

    Key design decisions:
    - Replay in batches to avoid overwhelming the pipeline
    - Mark each message as "pending" before replay to prevent duplicate processing
    - Update replay status regardless of success or failure
    - Support filtering by failure category, date range, or pipeline version
    """
    def __init__(
        self,
        dlq_consumer: Consumer,
        dlq_topic: str,
        processing_function,
        dlq_backend: DLQBackend,
        batch_size: int = 50,
        dry_run: bool = False,
    ):
        self.dlq_consumer = dlq_consumer
        self.dlq_topic = dlq_topic
        self.process = processing_function
        self.dlq_backend = dlq_backend
        self.batch_size = batch_size
        self.dry_run = dry_run

    def replay_by_category(
        self,
        category: FailureCategory,
        max_messages: int = 1000,
    ) -> Dict[str, int]:
        """
        Replay all DLQ messages matching a specific failure category.
        Returns a summary of replay outcomes.
        """
        results = {"success": 0, "failed": 0, "skipped": 0, "dry_run": 0}

        self.dlq_consumer.subscribe([self.dlq_topic])
        processed = 0

        while processed < max_messages:
            msg = self.dlq_consumer.poll(timeout=5.0)
            if msg is None:
                break  # No more messages

            envelope = DeadLetterEnvelope.from_json(msg.value().decode("utf-8"))

            # Filter by category
            if envelope.failure_category != category.value:
                results["skipped"] += 1
                continue

            # Skip messages already successfully replayed
            if envelope.replay_status == "success":
                results["skipped"] += 1
                continue

            if self.dry_run:
                logger.info(
                    f"[DRY RUN] Would replay dead_letter_id={envelope.dead_letter_id} "
                    f"original_topic={envelope.original_topic} "
                    f"failure_reason={envelope.failure_reason}"
                )
                results["dry_run"] += 1
                processed += 1
                continue

            # Attempt replay
            outcome = self._replay_single(envelope)
            results[outcome] += 1
            processed += 1

            if processed % 100 == 0:
                logger.info(f"Replay progress: {processed}/{max_messages} — {results}")

        self.dlq_consumer.close()
        return results

    def _replay_single(self, envelope: DeadLetterEnvelope) -> str:
        """Replay one message and return 'success' or 'failed'."""
        envelope.replay_count += 1
        envelope.last_replayed_at = datetime.now(timezone.utc).isoformat()
        envelope.replay_status = "pending"

        try:
            payload = json.loads(envelope.original_message)
            self.process(payload)

            envelope.replay_status = "success"
            logger.info(
                f"Replay succeeded for dead_letter_id={envelope.dead_letter_id}"
            )
            return "success"

        except Exception as e:
            envelope.replay_status = "failed"
            logger.error(
                f"Replay failed for dead_letter_id={envelope.dead_letter_id}: {e}"
            )
            # Write the updated envelope back to the DLQ with new replay status
            self.dlq_backend.send(envelope)
            return "failed"

Before you run a replay in production, always do a dry run first:

# Always start with dry_run=True to see what you're about to replay
replay_engine = DLQReplayEngine(
    dlq_consumer=consumer,
    dlq_topic="order-events.dlq",
    processing_function=process_order_event,
    dlq_backend=dlq_backend,
    dry_run=True,  # <-- safe to run anytime
)

# See what schema validation failures look like before touching them
summary = replay_engine.replay_by_category(
    category=FailureCategory.SCHEMA_VALIDATION,
    max_messages=500,
)
print(f"Dry run summary: {summary}")
# Output: {'success': 0, 'failed': 0, 'skipped': 12, 'dry_run': 47, 'in_progress': 0}

# Once satisfied, flip dry_run=False
replay_engine.dry_run = False
summary = replay_engine.replay_by_category(
    category=FailureCategory.SCHEMA_VALIDATION,
    max_messages=500,
)
print(f"Replay summary: {summary}")

---

Monitoring and Alerting: The DLQ Depth Problem

A DLQ is only useful if someone is watching it. The most important metric is DLQ depth — the number of messages sitting in the DLQ that haven't been replayed. A sudden spike in DLQ depth means something upstream broke. A slowly growing DLQ depth means you have an intermittent issue you haven't noticed yet.

Here's a monitoring implementation using Prometheus metrics:

from prometheus_client import Counter, Gauge, Histogram
import threading
import time

# Define your metrics
dlq_messages_routed = Counter(
    "dlq_messages_routed_total",
    "Total number of messages routed to the DLQ",
    ["pipeline_name", "failure_category", "exception_type"],
)

dlq_depth = Gauge(
    "dlq_depth",
    "Current number of unprocessed messages in the DLQ",
    ["dlq_name"],
)

dlq_replay_duration = Histogram(
    "dlq_replay_duration_seconds",
    "Time taken to replay a single DLQ message",
    ["pipeline_name", "outcome"],
)

class InstrumentedDeadLetterRouter(DeadLetterRouter):
    """Extends DeadLetterRouter with Prometheus instrumentation."""

    def route_to_dlq(self, message, exception, attempt_count, original_topic, **kwargs):
        category = classify_failure(exception, attempt_count)
        result = super().route_to_dlq(
            message, exception, attempt_count, original_topic, **kwargs
        )

        if result:
            dlq_messages_routed.labels(
                pipeline_name=self.pipeline_name,
                failure_category=category.value,
                exception_type=type(exception).__name__,
            ).inc()

        return result

def poll_dlq_depth(kafka_admin_client, dlq_topics: list, interval_seconds: int = 30):
    """
    Background thread that periodically measures DLQ depth.
    In production, this feeds your Grafana dashboards.
    """
    def _poll():
        while True:
            for topic in dlq_topics:
                try:
                    # Get lag using confluent-kafka AdminClient
                    metadata = kafka_admin_client.list_topics(topic=topic, timeout=10)
                    partition_count = len(metadata.topics[topic].partitions)

                    # Simplified depth calculation — use a proper lag tool in production
                    dlq_depth.labels(dlq_name=topic).set(
                        _calculate_topic_lag(kafka_admin_client, topic, partition_count)
                    )
                except Exception as e:
                    logger.error(f"Failed to poll DLQ depth for {topic}: {e}")

            time.sleep(interval_seconds)

    thread = threading.Thread(target=_poll, daemon=True)
    thread.start()

For alerting, define two alert rules in your monitoring system:

Alert 1: DLQ spike (immediate action required)

# Prometheus alerting rule
- alert: DLQDepthSpike
  expr: increase(dlq_messages_routed_total[5m]) > 10
  for: 2m
  labels:
    severity: warning
  annotations:
    summary: "DLQ receiving elevated message volume"
    description: >
      Pipeline {{ $labels.pipeline_name }} routed {{ $value }} messages
      to DLQ in the last 5 minutes. Category: {{ $labels.failure_category }}.

- alert: DLQDepthCritical
  expr: dlq_depth > 1000
  for: 5m
  labels:
    severity: critical
  annotations:
    summary: "DLQ depth exceeds critical threshold"
    description: "{{ $labels.dlq_name }} has {{ $value }} unprocessed messages."

---

Hands-On Exercise

In this exercise, you'll build a complete DLQ system for a simulated e-commerce event pipeline. The scenario: your pipeline consumes customer_activity events from Kafka, and a recent vendor change is sending events in a new format that breaks your consumer.

Setup: You'll need Docker with a Kafka instance. Create a docker-compose.yml with a single-node Kafka broker and create two topics: customer-activity and customer-activity.dlq.

Step 1: Define your custom exceptions and implement the classify_failure function from earlier. Add a case for a new DataSizeError that should be treated as a poison message immediately (events over 1MB should never be retried).

Step 2: Create a CustomerActivityProcessor class with a process(payload: dict) method that:

• Validates the payload has user_id, event_type, and timestamp fields (raise SchemaValidationError if missing)
• Validates event_type is one of ["page_view", "add_to_cart", "purchase", "search"] (raise BusinessRuleViolationError if not)
• Simulates a transient failure 20% of the time by raising ConnectionError (use random.random() < 0.2)
• Prints a success message otherwise

Step 3: Wire up the full consumer loop using the run_consumer_with_dlq pattern from this lesson. Produce a mix of good messages, schema-invalid messages, and business rule violations to your source topic.

Step 4: Run your consumer and observe the DLQ filling with enriched envelopes. Inspect a few dead letter messages and verify the failure_category, exception_traceback, and attempt_count fields are populated correctly.

Step 5: Implement a simplified DLQReplayEngine that reads from the DLQ and prints each envelope's failure_reason and original_message. Run it with dry_run=True first.

Challenge: Extend the replay engine to filter messages by first_attempt_at date so you can replay only messages from the last 24 hours.

---

Common Mistakes & Troubleshooting

Mistake 1: Not Committing Offsets After DLQ Routing

This is the most common and most damaging mistake. If your DLQ write fails (say, the DLQ topic is unavailable) and you also don't commit the offset, you'll reprocess that message on restart. If you commit without writing to the DLQ, you lose the message entirely.

The correct approach: implement a fallback for DLQ write failures. Options include writing to a local file, a separate database, or an emergency S3 bucket. In the worst case, log the full envelope at ERROR level — structured logging means you can reconstruct it later.

def _handle_final_failure_with_fallback(raw_value, exception, ...):
    success = dlq_router.route_to_dlq(raw_value, exception, ...)

    if not success:
        # Emergency fallback: log the full envelope so it's recoverable
        logger.error(
            "EMERGENCY_DLQ_FALLBACK",
            extra={
                "raw_message": raw_value,
                "exception_type": type(exception).__name__,
                "exception_message": str(exception),
            }
        )
        # Increment an alert counter so this gets noticed immediately
        dlq_fallback_counter.inc()

    # Always commit the offset — we've done everything we can
    consumer.commit(message=msg, asynchronous=False)

Mistake 2: Infinite Retry Loops on Transient Failures

If you don't distinguish between transient and poison failures, you can end up retrying a message forever. Always cap your retry count. If a "transient" failure has persisted through 5 retries with exponential backoff, it's no longer transient — route it to the DLQ.

Mistake 3: DLQ Messages Expiring Before You Notice

Kafka topic retention is not infinite, and SQS message retention maxes out at 14 days. If a DLQ fills up during a vacation and nobody notices for three weeks, you've lost your recovery window.

The fix is two-fold: set up the DLQ depth alert we covered earlier, and consider writing DLQ messages to both Kafka and a durable store like S3 or a database table. The Kafka DLQ is for operational use; the S3 archive is your audit trail.

Mistake 4: Replaying Without Idempotency

Replay is only safe if your processing function is idempotent — meaning running it twice produces the same result as running it once. Before you build a replay engine, make sure your downstream writes are idempotent (upserts rather than inserts, deduplication by message ID, etc.).

Mistake 5: Storing Sensitive Data in DLQ Messages

Your dead letter envelopes contain the original payload. If that payload contains PII — customer names, email addresses, payment details — your DLQ becomes a compliance risk. Apply the same encryption and access controls to your DLQ topics that you apply to your source topics. Consider masking sensitive fields before storing in the envelope.

Debugging: "My DLQ is growing but I don't know why"

Start with the failure_category distribution:

# Quick analysis script for DLQ topic
from collections import Counter

def analyze_dlq(consumer, dlq_topic, max_messages=500):
    consumer.subscribe([dlq_topic])
    categories = Counter()
    exception_types = Counter()
    pipelines = Counter()

    for _ in range(max_messages):
        msg = consumer.poll(timeout=2.0)
        if msg is None:
            break
        envelope = DeadLetterEnvelope.from_json(msg.value().decode("utf-8"))
        categories[envelope.failure_category] += 1
        exception_types[envelope.exception_type] += 1
        pipelines[envelope.pipeline_name] += 1

    print("By failure category:", categories.most_common())
    print("By exception type:", exception_types.most_common())
    print("By pipeline:", pipelines.most_common())

This quick script will usually point you directly at the source of the problem within minutes.

---

Summary & Next Steps

Let's recap what you've built. You now have a complete poison message handling system that:

1. Classifies failures into transient vs. poison categories so you retry what's recoverable and isolate what isn't
2. Enriches dead letters with full diagnostic context — traceback, attempt history, pipeline metadata — so failures are debuggable without having to reproduce them
3. Routes intelligently using a pluggable backend that works with Kafka, SQS, or any other messaging system
4. Replays safely with dry-run support, category filtering, and idempotency awareness
5. Monitors continuously with Prometheus metrics and alert rules that catch DLQ spikes before they become incidents

The pattern you've implemented here is production-grade. The DeadLetterEnvelope design, the offset-commit discipline, and the classification logic are all things you'd find in mature data platform teams.

Where to go from here:

• Schema Registry integration: Rather than catching json.JSONDecodeError manually, integrate with Confluent Schema Registry to validate messages against Avro or Protobuf schemas before your processing logic ever sees them. This catches schema drift at the infrastructure layer.

• DLQ-driven alerting with PagerDuty: Wire your Prometheus DLQ depth metric to PagerDuty so on-call gets paged automatically when depth exceeds your SLA threshold.

• Exactly-once semantics: The approach in this lesson is at-least-once. For truly exactly-once processing, explore Kafka transactions combined with transactional writes to your sink. This is a significant architectural step up but eliminates the idempotency burden from your application code.

• Multi-hop DLQs: For pipelines with multiple processing stages, consider a DLQ per stage rather than a single global DLQ. This makes it much easier to identify which stage failed and replay only from that point forward.

The 2 AM incident from the introduction? With what you've built today, that's a 9 AM Slack message instead.