Dead Letters and Poison Pills.

{
  "order_id": "ord_123",
  "user_id": "usr_456",
  "items": "INVALID - should be array, got string",  // ← The poison
  "total": 99.99
}

{
  "original_message": { ... },
  "error": "ValueError: Invalid item quantity",
  "failed_at": "2024-01-15T10:30:00Z",
  "retry_count": 3,
  "consumer_id": "orders-consumer-1",
  "stack_trace": "..."
}

# Error Classification and Routing

from enum import Enum
from typing import Optional
import json


class ErrorCategory(Enum):
    VALIDATION = "validation"      # Bad data, won't retry
    TRANSIENT = "transient"        # Might succeed later
    DEPENDENCY = "dependency"      # External service issue
    POISON = "poison"              # Crashes consumer
    UNKNOWN = "unknown"            # Unclassified


ERROR_ROUTING = {
    ErrorCategory.VALIDATION: {
        "retry": False,
        "dlq": "orders.dlq.validation",
        "alert": False,  # Expected errors
    },
    ErrorCategory.TRANSIENT: {
        "retry": True,
        "max_retries": 5,
        "backoff": "exponential",
        "dlq": "orders.dlq.transient",
        "alert": True,  # Alert if DLQ'd
    },
    ErrorCategory.DEPENDENCY: {
        "retry": True,
        "max_retries": 10,
        "backoff": "linear",
        "dlq": "orders.dlq.dependency",
        "alert": True,
    },
    ErrorCategory.POISON: {
        "retry": False,  # Never retry poison
        "dlq": "orders.dlq.poison",
        "alert": True,   # Immediate alert
        "page_oncall": True,  # Wake someone up
    },
    ErrorCategory.UNKNOWN: {
        "retry": True,
        "max_retries": 3,
        "dlq": "orders.dlq.unknown",
        "alert": True,
    },
}


def classify_error(error: Exception) -> ErrorCategory:
    """Classify an exception into an error category."""
    
    # Validation errors
    if isinstance(error, (ValueError, json.JSONDecodeError, KeyError)):
        return ErrorCategory.VALIDATION
    
    # Transient errors
    if isinstance(error, (TimeoutError, ConnectionError)):
        return ErrorCategory.TRANSIENT
    
    # Dependency errors
    if "503" in str(error) or "rate limit" in str(error).lower():
        return ErrorCategory.DEPENDENCY
    
    # Poison indicators
    if isinstance(error, (MemoryError, RecursionError)):
        return ErrorCategory.POISON
    
    return ErrorCategory.UNKNOWN

# Basic DLQ Implementation
# WARNING: Not production-ready - for learning only

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


@dataclass
class DeadLetter:
    """A message that failed processing."""
    original_message: Dict[str, Any]
    error_type: str
    error_message: str
    stack_trace: str
    failed_at: str
    retry_count: int
    consumer_id: str
    topic: str
    partition: Optional[int] = None
    offset: Optional[int] = None


class SimpleDLQConsumer:
    """
    Consumer with basic dead letter queue support.
    """
    
    def __init__(
        self,
        kafka_consumer,
        kafka_producer,
        dlq_topic: str,
        max_retries: int = 3,
        consumer_id: str = "consumer-1"
    ):
        self.consumer = kafka_consumer
        self.producer = kafka_producer
        self.dlq_topic = dlq_topic
        self.max_retries = max_retries
        self.consumer_id = consumer_id
    
    async def process_message(self, message: Dict[str, Any]) -> None:
        """Override this to implement your processing logic."""
        raise NotImplementedError
    
    async def run(self):
        """Main consumer loop with DLQ handling."""
        async for msg in self.consumer:
            # Kafka headers are a list of (str, bytes) tuples
            retry_count = int(dict(msg.headers or []).get('retry_count', b'0'))
            
            try:
                # Parse message
                value = json.loads(msg.value)
                
                # Process
                await self.process_message(value)
                
                # Commit on success
                await self.consumer.commit()
                
            except Exception as e:
                await self._handle_failure(msg, e, retry_count)
    
    async def _handle_failure(self, msg, error: Exception, retry_count: int):
        """Handle a processing failure."""
        import traceback
        
        if retry_count < self.max_retries:
            # Retry: republish with incremented count
            print(f"Retrying message (attempt {retry_count + 1}/{self.max_retries})")
            await self._retry_message(msg, retry_count + 1)
        else:
            # Max retries exceeded: send to DLQ
            print(f"Max retries exceeded, sending to DLQ: {error}")
            await self._send_to_dlq(msg, error, retry_count)
        
        # Commit to move past this message
        await self.consumer.commit()
    
    async def _retry_message(self, msg, new_retry_count: int):
        """Republish message for retry."""
        headers = dict(msg.headers) if msg.headers else {}
        headers['retry_count'] = str(new_retry_count).encode()
        
        await self.producer.send(
            msg.topic,
            value=msg.value,
            key=msg.key,
            headers=list(headers.items())
        )
    
    async def _send_to_dlq(self, msg, error: Exception, retry_count: int):
        """Send failed message to dead letter queue."""
        import traceback
        
        try:
            original = json.loads(msg.value)
        except ValueError:
            # The failure may BE the bad JSON; don't crash while quarantining it
            original = {"raw": msg.value.decode('utf-8', errors='replace')}

        dead_letter = DeadLetter(
            original_message=original,
            error_type=type(error).__name__,
            error_message=str(error),
            stack_trace=traceback.format_exc(),
            failed_at=datetime.now(timezone.utc).isoformat(),
            retry_count=retry_count,
            consumer_id=self.consumer_id,
            topic=msg.topic,
            partition=msg.partition,
            offset=msg.offset
        )
        
        await self.producer.send(
            self.dlq_topic,
            value=json.dumps(asdict(dead_letter)).encode(),
            key=msg.key
        )


# Usage example
class OrderConsumer(SimpleDLQConsumer):
    async def process_message(self, message: Dict[str, Any]) -> None:
        order_id = message['order_id']
        items = message['items']  # May raise KeyError
        
        if not isinstance(items, list):
            raise ValueError(f"Items must be a list, got {type(items)}")
        
        # Process order...
        print(f"Processing order {order_id} with {len(items)} items")

# Production-Ready Dead Letter Queue Implementation

import asyncio
import json
import traceback
import hashlib
from dataclasses import dataclass, field, asdict
from datetime import datetime, timedelta, timezone
from typing import Optional, Dict, Any, List, Callable, Set
from enum import Enum
import logging
import time

from aiokafka import TopicPartition

logger = logging.getLogger(__name__)


# =============================================================================
# Configuration
# =============================================================================

@dataclass
class DLQConfig:
    """Configuration for dead letter queue handling."""
    
    # Retry configuration
    max_retries: int = 5  # the simple consumer in 4.1 used 3; tune per error category
    base_backoff_seconds: float = 1.0
    max_backoff_seconds: float = 300.0
    backoff_multiplier: float = 2.0
    
    # DLQ topics
    dlq_topic_suffix: str = ".dlq"
    validation_dlq_suffix: str = ".dlq.validation"
    poison_dlq_suffix: str = ".dlq.poison"
    
    # Poison pill detection
    poison_detection_enabled: bool = True
    poison_timeout_seconds: float = 30.0
    
    # Alerting thresholds
    dlq_alert_threshold: int = 100
    dlq_critical_threshold: int = 1000


class ErrorCategory(Enum):
    VALIDATION = "validation"
    TRANSIENT = "transient"
    DEPENDENCY = "dependency"
    POISON = "poison"
    UNKNOWN = "unknown"


@dataclass
class DeadLetter:
    """A message that failed processing."""
    id: str
    original_topic: str
    original_partition: int
    original_offset: int
    original_key: Optional[str]
    original_value: Dict[str, Any]
    original_headers: Dict[str, str]
    
    error_category: str
    error_type: str
    error_message: str
    stack_trace: str
    
    first_failed_at: str
    last_failed_at: str
    retry_count: int
    
    consumer_id: str
    consumer_version: str
    
    # For replay tracking
    replayed: bool = False
    replayed_at: Optional[str] = None
    replay_result: Optional[str] = None


# =============================================================================
# Error Classification
# =============================================================================

class ErrorClassifier:
    """Classifies exceptions into error categories."""
    
    # Exceptions that indicate validation errors
    VALIDATION_ERRORS = (
        ValueError,
        KeyError,
        TypeError,
        json.JSONDecodeError,
    )
    
    # Exceptions that indicate transient errors
    TRANSIENT_ERRORS = (
        TimeoutError,
        ConnectionError,
        ConnectionResetError,
    )
    
    # Error message patterns
    TRANSIENT_PATTERNS = [
        "timeout",
        "connection refused",
        "temporarily unavailable",
        "try again",
        "rate limit",
        "503",
        "504",
    ]
    
    POISON_PATTERNS = [
        "out of memory",
        "recursion",
        "stack overflow",
        "segmentation fault",
    ]
    
    @classmethod
    def classify(cls, error: Exception) -> ErrorCategory:
        """Classify an exception."""
        error_str = str(error).lower()
        
        # Check for poison indicators
        if isinstance(error, (MemoryError, RecursionError, PoisonPillError)):
            return ErrorCategory.POISON
        
        for pattern in cls.POISON_PATTERNS:
            if pattern in error_str:
                return ErrorCategory.POISON
        
        # Check for validation errors
        if isinstance(error, cls.VALIDATION_ERRORS):
            return ErrorCategory.VALIDATION
        
        # Check for transient errors
        if isinstance(error, cls.TRANSIENT_ERRORS):
            return ErrorCategory.TRANSIENT
        
        for pattern in cls.TRANSIENT_PATTERNS:
            if pattern in error_str:
                return ErrorCategory.TRANSIENT
        
        return ErrorCategory.UNKNOWN


# =============================================================================
# Core DLQ Handler
# =============================================================================

class DLQHandler:
    """
    Production-ready dead letter queue handler.
    
    Features:
    - Error classification and routing
    - Exponential backoff
    - Poison pill detection
    - Metrics and alerting
    - Replay support
    """
    
    def __init__(
        self,
        config: DLQConfig,
        kafka_producer,
        metrics_client,
        consumer_id: str,
        consumer_version: str = "1.0.0"
    ):
        self.config = config
        self.producer = kafka_producer
        self.metrics = metrics_client
        self.consumer_id = consumer_id
        self.consumer_version = consumer_version
        
        # Track retry attempts per message
        self._retry_counts: Dict[str, int] = {}
        self._first_failures: Dict[str, datetime] = {}
        
        # Metrics
        self.messages_retried = 0
        self.messages_dlqd = 0
        self.poison_pills_detected = 0
    
    def get_message_id(self, topic: str, partition: int, offset: int) -> str:
        """Generate unique ID for a message."""
        content = f"{topic}:{partition}:{offset}"
        return hashlib.md5(content.encode()).hexdigest()[:16]
    
    async def handle_failure(
        self,
        topic: str,
        partition: int,
        offset: int,
        key: Optional[bytes],
        value: bytes,
        headers: Dict[str, str],
        error: Exception
    ) -> None:
        """
        Handle a message processing failure.
        
        Decides whether to retry or send to DLQ based on error type
        and retry count.
        """
        msg_id = self.get_message_id(topic, partition, offset)
        
        # Classify error
        category = ErrorClassifier.classify(error)
        
        # Track first failure time
        if msg_id not in self._first_failures:
            self._first_failures[msg_id] = datetime.now(timezone.utc)
        
        # Get current retry count
        retry_count = self._retry_counts.get(msg_id, 0)
        
        # Determine action based on category
        if category == ErrorCategory.POISON:
            # Never retry poison pills
            logger.error(f"Poison pill detected: {error}")
            self.poison_pills_detected += 1
            await self._send_to_dlq(
                topic, partition, offset, key, value, headers,
                error, category, retry_count,
                dlq_suffix=self.config.poison_dlq_suffix
            )
            self._cleanup_tracking(msg_id)
            
        elif category == ErrorCategory.VALIDATION:
            # Don't retry validation errors
            logger.warning(f"Validation error, sending to DLQ: {error}")
            await self._send_to_dlq(
                topic, partition, offset, key, value, headers,
                error, category, retry_count,
                dlq_suffix=self.config.validation_dlq_suffix
            )
            self._cleanup_tracking(msg_id)
            
        elif retry_count < self.config.max_retries:
            # Retry with backoff
            backoff = self._calculate_backoff(retry_count)
            logger.info(
                f"Retrying message (attempt {retry_count + 1}/{self.config.max_retries}), "
                f"backoff: {backoff:.1f}s"
            )
            
            self._retry_counts[msg_id] = retry_count + 1
            self.messages_retried += 1
            
            # Wait for backoff
            await asyncio.sleep(backoff)
            
            # Raise to trigger reprocessing
            raise RetryableError(f"Retry attempt {retry_count + 1}")
            
        else:
            # Max retries exceeded
            logger.warning(f"Max retries exceeded, sending to DLQ: {error}")
            await self._send_to_dlq(
                topic, partition, offset, key, value, headers,
                error, category, retry_count
            )
            self._cleanup_tracking(msg_id)
    
    def _calculate_backoff(self, retry_count: int) -> float:
        """Calculate exponential backoff duration."""
        backoff = self.config.base_backoff_seconds * (
            self.config.backoff_multiplier ** retry_count
        )
        return min(backoff, self.config.max_backoff_seconds)
    
    async def _send_to_dlq(
        self,
        topic: str,
        partition: int,
        offset: int,
        key: Optional[bytes],
        value: bytes,
        headers: Dict[str, str],
        error: Exception,
        category: ErrorCategory,
        retry_count: int,
        dlq_suffix: Optional[str] = None
    ) -> None:
        """Send a failed message to the dead letter queue."""
        
        # Determine DLQ topic
        suffix = dlq_suffix or self.config.dlq_topic_suffix
        dlq_topic = f"{topic}{suffix}"
        
        # Parse original value
        try:
            original_value = json.loads(value.decode('utf-8'))
        except:
            original_value = {"raw": value.decode('utf-8', errors='replace')}
        
        # Build dead letter
        msg_id = self.get_message_id(topic, partition, offset)
        first_failed = self._first_failures.get(msg_id, datetime.now(timezone.utc))
        
        dead_letter = DeadLetter(
            id=msg_id,
            original_topic=topic,
            original_partition=partition,
            original_offset=offset,
            original_key=key.decode('utf-8') if key else None,
            original_value=original_value,
            original_headers=headers,
            error_category=category.value,
            error_type=type(error).__name__,
            error_message=str(error)[:1000],  # Truncate long messages
            stack_trace=traceback.format_exc()[:5000],  # Truncate
            first_failed_at=first_failed.isoformat(),
            last_failed_at=datetime.now(timezone.utc).isoformat(),
            retry_count=retry_count,
            consumer_id=self.consumer_id,
            consumer_version=self.consumer_version
        )
        
        # Send to DLQ
        await self.producer.send(
            dlq_topic,
            value=json.dumps(asdict(dead_letter)).encode('utf-8'),
            key=key
        )
        
        # Update metrics
        self.messages_dlqd += 1
        self.metrics.increment('dlq.messages.total', tags={'topic': topic, 'category': category.value})
        
        logger.info(f"Message sent to DLQ: {dlq_topic}")
    
    def _cleanup_tracking(self, msg_id: str) -> None:
        """Clean up tracking state for a message."""
        self._retry_counts.pop(msg_id, None)
        self._first_failures.pop(msg_id, None)


class RetryableError(Exception):
    """Raised to trigger message reprocessing."""
    pass


# =============================================================================
# Consumer with DLQ Support
# =============================================================================

class DLQConsumer:
    """
    Kafka consumer with integrated dead letter queue support.
    """
    
    def __init__(
        self,
        kafka_consumer,
        kafka_producer,
        dlq_handler: DLQHandler,
        process_fn: Callable[[Dict[str, Any]], Any],
        config: DLQConfig
    ):
        self.consumer = kafka_consumer
        self.producer = kafka_producer
        self.dlq_handler = dlq_handler
        self.process_fn = process_fn
        self.config = config
        
        self._running = False
    
    async def start(self):
        """Start consuming messages."""
        self._running = True
        logger.info("DLQ consumer started")
        
        async for msg in self.consumer:
            if not self._running:
                break
            
            await self._process_message(msg)
    
    async def stop(self):
        """Stop the consumer."""
        self._running = False
    
    async def _process_message(self, msg) -> None:
        """Process a single message with DLQ handling."""
        headers = {k: v.decode('utf-8', 'replace') for k, v in (msg.headers or [])}
        
        try:
            # Parse message
            value = json.loads(msg.value.decode('utf-8'))
            
            # Detect poison pills with timeout
            if self.config.poison_detection_enabled:
                try:
                    await asyncio.wait_for(
                        self._process_with_timeout(value),
                        timeout=self.config.poison_timeout_seconds
                    )
                except asyncio.TimeoutError:
                    raise PoisonPillError(
                        f"Processing timed out after {self.config.poison_timeout_seconds}s"
                    )
            else:
                await self.process_fn(value)
            
            # Success - commit
            await self.consumer.commit()
            
        except RetryableError:
            # Not committing is NOT enough: the consumer's position has already
            # moved past this message. Seek back so it is fetched again.
            self.consumer.seek(TopicPartition(msg.topic, msg.partition), msg.offset)
            
        except Exception as e:
            # Handle failure
            try:
                await self.dlq_handler.handle_failure(
                    topic=msg.topic,
                    partition=msg.partition,
                    offset=msg.offset,
                    key=msg.key,
                    value=msg.value,
                    headers=headers,
                    error=e
                )
                # Commit to move past this message
                await self.consumer.commit()
            except RetryableError:
                # Seek back so the same message is fetched and retried
                self.consumer.seek(TopicPartition(msg.topic, msg.partition), msg.offset)
    
    async def _process_with_timeout(self, value: Dict[str, Any]) -> None:
        """Process with timeout for poison pill detection."""
        await self.process_fn(value)


class PoisonPillError(Exception):
    """Indicates a poison pill message."""
    pass


# =============================================================================
# DLQ Replay Tool
# =============================================================================

class DLQReplayTool:
    """
    Tool for investigating and replaying DLQ messages.
    """
    
    def __init__(self, kafka_consumer, kafka_producer, db_client=None):
        self.consumer = kafka_consumer
        self.producer = kafka_producer
        self.db = db_client  # Optional: for tracking replays
    
    async def list_messages(
        self,
        dlq_topic: str,
        limit: int = 100,
        error_category: Optional[str] = None
    ) -> List[DeadLetter]:
        """List messages in the DLQ."""
        messages = []
        
        async for msg in self.consumer:
            if len(messages) >= limit:
                break
            
            dead_letter = DeadLetter(**json.loads(msg.value))
            
            if error_category and dead_letter.error_category != error_category:
                continue
            
            messages.append(dead_letter)
        
        return messages
    
    async def replay_message(
        self,
        dead_letter: DeadLetter,
        fix_fn: Optional[Callable[[Dict], Dict]] = None
    ) -> bool:
        """
        Replay a single message from the DLQ.
        
        Args:
            dead_letter: The dead letter to replay
            fix_fn: Optional function to fix the message before replay
        """
        value = dead_letter.original_value
        
        # Apply fix if provided
        if fix_fn:
            try:
                value = fix_fn(value)
            except Exception as e:
                logger.error(f"Fix function failed: {e}")
                return False
        
        # Republish to original topic
        try:
            await self.producer.send(
                dead_letter.original_topic,
                value=json.dumps(value).encode('utf-8'),
                key=dead_letter.original_key.encode() if dead_letter.original_key else None
            )
            
            logger.info(f"Replayed message {dead_letter.id} to {dead_letter.original_topic}")
            return True
            
        except Exception as e:
            logger.error(f"Replay failed: {e}")
            return False
    
    async def replay_batch(
        self,
        dead_letters: List[DeadLetter],
        fix_fn: Optional[Callable[[Dict], Dict]] = None,
        dry_run: bool = False
    ) -> Dict[str, int]:
        """Replay multiple messages."""
        results = {"success": 0, "failed": 0, "skipped": 0}
        
        for dl in dead_letters:
            if dry_run:
                logger.info(f"[DRY RUN] Would replay: {dl.id}")
                results["skipped"] += 1
                continue
            
            success = await self.replay_message(dl, fix_fn)
            if success:
                results["success"] += 1
            else:
                results["failed"] += 1
        
        return results

# DLQ Monitoring

class DLQMonitor:
    """
    Monitors dead letter queue health and triggers alerts.
    """
    
    def __init__(
        self,
        metrics_client,
        alerting_client,
        dlq_topics: List[str],
        config: DLQConfig
    ):
        self.metrics = metrics_client
        self.alerting = alerting_client
        self.dlq_topics = dlq_topics
        self.config = config
    
    async def check_health(self) -> Dict[str, Any]:
        """Check DLQ health across all topics."""
        results = {}
        
        for topic in self.dlq_topics:
            depth = await self.metrics.get_queue_depth(topic)
            growth_rate = await self.metrics.get_growth_rate(topic)
            
            results[topic] = {
                "depth": depth,
                "growth_rate": growth_rate,
                "status": self._determine_status(depth)
            }
            
            # Trigger alerts
            if depth >= self.config.dlq_critical_threshold:
                await self.alerting.send_alert(
                    severity="critical",
                    title=f"DLQ Critical: {topic}",
                    message=f"DLQ depth is {depth:,}, threshold is {self.config.dlq_critical_threshold:,}",
                    tags={"topic": topic}
                )
            elif depth >= self.config.dlq_alert_threshold:
                await self.alerting.send_alert(
                    severity="warning",
                    title=f"DLQ Warning: {topic}",
                    message=f"DLQ depth is {depth:,}, threshold is {self.config.dlq_alert_threshold:,}",
                    tags={"topic": topic}
                )
        
        return results
    
    def _determine_status(self, depth: int) -> str:
        if depth >= self.config.dlq_critical_threshold:
            return "critical"
        elif depth >= self.config.dlq_alert_threshold:
            return "warning"
        return "healthy"