Error Handling, Retries, and Dead Letter Queues

Error Handling and Resilience Patterns for Data Pipelines

Production data pipelines fail. The question is not if your pipeline will fail, but how it handles failure.

The Error Landscape in Data Pipelines

Error Categories:

Handling Strategies:

1. Retries with exponential backoff — for transient errors
2. Dead letter queues — for poison messages
3. Circuit breakers — for cascading failure prevention
4. Fallback mechanisms — for graceful degradation

Key Insight: Each error category requires a different handling strategy.

Circuit Breaker Pattern

Exponential Backoff Timeline

Architecture Diagram

Key Concepts

Production Code

Retry Handler with Exponential Backoff

import time
import random
import logging
from typing import Callable, Any, Optional
from functools import wraps
from dataclasses import dataclass
from datetime import datetime
import traceback

logger = logging.getLogger(__name__)


@dataclass
class RetryConfig:
    """
    Configuration for retry behavior.
    
    Parameters:
        max_retries (int): Maximum number of retry attempts before raising exception
        base_delay (float): Initial delay in seconds before first retry
        max_delay (float): Maximum delay cap in seconds (prevents unbounded growth)
        exponential_base (float): Base for exponential growth (2.0 = doubling each retry)
        jitter (bool): Add random jitter to prevent thundering herd problem
        retryable_exceptions (tuple): Exception types that should trigger retry
        
    Example with backoff timing:
        max_retries=5, base_delay=1.0, exponential_base=2.0
        Attempt 1: wait 1.0s
        Attempt 2: wait 2.0s
        Attempt 3: wait 4.0s
        Attempt 4: wait 8.0s
        Attempt 5: wait 16.0s
        Total wait: 31.0s
    """
    max_retries: int = 5
    base_delay: float = 1.0
    max_delay: float = 60.0
    exponential_base: float = 2.0
    jitter: bool = True
    retryable_exceptions: tuple = (Exception,)


class RetryHandler:
    """Production retry handler with exponential backoff and jitter."""

    def __init__(self, config: RetryConfig = None):
        self.config = config or RetryConfig()

    def __call__(self, func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            last_exception = None
            for attempt in range(self.config.max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except self.config.retryable_exceptions as e:
                    last_exception = e
                    if attempt == self.config.max_retries:
                        logger.error(
                            f"Max retries ({self.config.max_retries}) exceeded "
                            f"for {func.__name__}: {e}"
                        )
                        raise

                    # Calculate delay with exponential backoff
                    delay = min(
                        self.config.base_delay * (self.config.exponential_base ** attempt),
                        self.config.max_delay,
                    )

                    # Add jitter
                    if self.config.jitter:
                        delay = delay * random.uniform(0, 1)

                    logger.warning(
                        f"Attempt {attempt + 1}/{self.config.max_retries} failed "
                        f"for {func.__name__}: {e}. Retrying in {delay:.2f}s..."
                    )
                    time.sleep(delay)

            raise last_exception

        return wrapper


# Usage example
@RetryHandler(config=RetryConfig(
    max_retries=5,
    base_delay=1.0,
    max_delay=30.0,
    jitter=True,
    retryable_exceptions=(ConnectionError, TimeoutError),
))
def fetch_from_api(url: str, timeout: int = 30) -> dict:
    """Fetch data from external API with automatic retries."""
    import requests
    response = requests.get(url, timeout=timeout)
    response.raise_for_status()
    return response.json()


# Custom retry handler for specific error patterns
class SmartRetryHandler:
    """Retry handler that adapts behavior based on error type."""

    def __init__(self):
        self.error_counts = {}
        self.circuit_open = False
        self.circuit_open_until = None

    def execute_with_retry(
        self,
        func: Callable,
        *args,
        max_retries: int = 3,
        **kwargs,
    ) -> Any:
        """Execute function with smart retry logic."""
        # Check circuit breaker
        if self.circuit_open:
            if datetime.utcnow() < self.circuit_open_until:
                raise CircuitBreakerOpenError(
                    f"Circuit breaker open until {self.circuit_open_until}"
                )
            else:
                self.circuit_open = False
                logger.info("Circuit breaker entering half-open state")

        for attempt in range(max_retries + 1):
            try:
                result = func(*args, **kwargs)
                # Success: reset error count
                self.error_counts.pop(func.__name__, None)
                return result

            except RateLimitError as e:
                # Rate limit: wait longer before retry
                delay = min(60 * (2 ** attempt), 300)
                logger.warning(f"Rate limited. Waiting {delay}s before retry.")
                time.sleep(delay)

            except TemporaryError as e:
                # Transient error: standard exponential backoff
                delay = min(2 ** attempt, 30)
                logger.warning(f"Transient error. Retrying in {delay}s: {e}")
                time.sleep(delay)

            except PermanentError as e:
                # Permanent error: don't retry, send to DLQ
                logger.error(f"Permanent error, not retrying: {e}")
                raise

            except Exception as e:
                # Unknown error: track and potentially open circuit
                self.error_counts[func.__name__] = (
                    self.error_counts.get(func.__name__, 0) + 1
                )
                if self.error_counts[func.__name__] >= 10:
                    self.circuit_open = True
                    self.circuit_open_until = datetime.utcnow() + timedelta(seconds=60)
                    logger.error(
                        f"Circuit breaker OPEN for {func.__name__} "
                        f"after {self.error_counts[func.__name__]} consecutive errors"
                    )
                raise

        raise Exception(f"Max retries ({max_retries}) exceeded for {func.__name__}")


class CircuitBreakerOpenError(Exception):
    pass

Dead Letter Queue Implementation

import json
import logging
from datetime import datetime
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
from kafka import KafkaProducer, KafkaConsumer

logger = logging.getLogger(__name__)


@dataclass
class DLQMessage:
    """Message stored in dead letter queue with full context."""
    original_topic: str
    original_partition: int
    original_offset: int
    original_key: Optional[str]
    original_value: str
    error_type: str
    error_message: str
    error_traceback: str
    retry_count: int
    first_failure_time: str
    last_failure_time: str
    pipeline_name: str
    stage_name: str


class DeadLetterQueueManager:
    """Manage dead letter queues for failed pipeline records."""

    def __init__(
        self,
        kafka_bootstrap: str,
        dlq_topic: str = "pipeline.dlq",
        max_retries: int = 3,
    ):
        self.dlq_topic = dlq_topic
        self.max_retries = max_retries
        self.producer = KafkaProducer(
            bootstrap_servers=kafka_bootstrap,
            acks="all",
            enable_idempotence=True,
            value_serializer=lambda v: json.dumps(v, default=str).encode("utf-8"),
        )
        self.retry_counts: Dict[str, int] = {}

    def send_to_dlq(
        self,
        original_message,
        error: Exception,
        pipeline_name: str,
        stage_name: str,
    ):
        """Send a failed message to the dead letter queue."""
        dlq_msg = DLQMessage(
            original_topic=original_message.topic,
            original_partition=original_message.partition,
            original_offset=original_message.offset,
            original_key=original_message.key.decode("utf-8") if original_message.key else None,
            original_value=original_message.value.decode("utf-8"),
            error_type=type(error).__name__,
            error_message=str(error),
            error_traceback=traceback.format_exc(),
            retry_count=self.retry_counts.get(
                f"{original_message.topic}:{original_message.offset}", 0
            ),
            first_failure_time=datetime.utcnow().isoformat(),
            last_failure_time=datetime.utcnow().isoformat(),
            pipeline_name=pipeline_name,
            stage_name=stage_name,
        )

        self.producer.send(
            topic=self.dlq_topic,
            key=original_message.key,
            value=asdict(dlq_msg),
        )

        logger.warning(
            f"Message sent to DLQ: topic={original_message.topic}, "
            f"offset={original_message.offset}, error={type(error).__name__}"
        )

    def should_retry(self, message_key: str) -> bool:
        """Check if a message should be retried or sent to DLQ."""
        count = self.retry_counts.get(message_key, 0)
        if count >= self.max_retries:
            return False
        self.retry_counts[message_key] = count + 1
        return True

    def replay_dlq(
        self,
        target_processor: callable,
        filter_fn: callable = None,
        max_messages: int = None,
    ) -> Dict:
        """Replay messages from DLQ through a processor."""
        consumer = KafkaConsumer(
            self.dlq_topic,
            bootstrap_servers=self.producer.config["bootstrap_servers"],
            auto_offset_reset="earliest",
            value_deserializer=lambda m: json.loads(m.decode("utf-8")),
        )

        replayed = 0
        failed = 0
        skipped = 0

        for message in consumer:
            if max_messages and replayed >= max_messages:
                break

            dlq_msg = message.value

            # Apply filter if provided
            if filter_fn and not filter_fn(dlq_msg):
                skipped += 1
                continue

            try:
                # Reconstruct original message for processing
                original = {
                    "topic": dlq_msg["original_topic"],
                    "key": dlq_msg["original_key"],
                    "value": dlq_msg["original_value"],
                }
                target_processor(original)
                replayed += 1
                logger.info(f"Successfully replayed message: {dlq_msg['original_offset']}")
            except Exception as e:
                failed += 1
                logger.error(f"Replay failed for offset {dlq_msg['original_offset']}: {e}")

        consumer.close()
        return {"replayed": replayed, "failed": failed, "skipped": skipped}

    def get_dlq_stats(self) -> Dict:
        """Get DLQ statistics for monitoring."""
        consumer = KafkaConsumer(
            self.dlq_topic,
            bootstrap_servers=self.producer.config["bootstrap_servers"],
            auto_offset_reset="earliest",
        )

        total_messages = 0
        error_types = {}
        pipelines = {}

        for partition in consumer.assignment():
            total_messages += consumer.end_offsets([partition])[partition]

        consumer.close()

        return {
            "total_messages": total_messages,
            "error_types": error_types,
            "pipelines": pipelines,
        }


# Example: Pipeline with DLQ integration
def process_with_dlq(kafka_bootstrap: str, input_topic: str):
    """Process messages with dead letter queue for failures."""
    dlq = DeadLetterQueueManager(kafka_bootstrap, "pipeline.dlq")
    consumer = KafkaConsumer(
        input_topic,
        bootstrap_servers=kafka_bootstrap,
        group_id="pipeline-processor",
        auto_offset_reset="earliest",
        enable_auto_commit=False,
    )

    for message in consumer:
        message_key = f"{message.topic}:{message.offset}"

        if not dlq.should_retry(message_key):
            dlq.send_to_dlq(
                message,
                Exception("Max retries exceeded"),
                "daily_orders_etl",
                "transform",
            )
            consumer.commit()
            continue

        try:
            # Process message
            data = json.loads(message.value.decode("utf-8"))
            # ... transformation logic ...
            consumer.commit()

        except Exception as e:
            dlq.send_to_dlq(message, e, "daily_orders_etl", "transform")

    consumer.close()

Best Practices

1. Classify errors before retrying. Only retry transient errors (timeouts, rate limits, locks). Route permanent errors to DLQ.
2. Use exponential backoff with jitter to prevent thundering herd. Base delay: 1s, max delay: 60s, jitter: random(0,1).
3. Set max retries based on error tolerance. For critical pipelines: 5 retries. For non-critical: 3 retries.
4. Implement DLQs for every pipeline. Store original message, error details, retry count, and timestamps.
5. Ensure idempotency for all retried operations. Use deduplication keys, UPSERT operations, and transactional writes.
6. Deploy circuit breakers before external service calls. Configure: error rate threshold (50%), minimum requests (10), timeout (30s).
7. Implement fallbacks for graceful degradation. Use cached data, default values, or partial results when primary operations fail.
8. Set timeouts on all external calls. Prevent hung requests from blocking resources. Typical timeout: 30s.
9. Monitor error rates per pipeline stage, error type, and error category. Alert on anomalies.
10. Build DLQ tooling: create tools to inspect, debug, and reprocess DLQ messages. Automate common remediations.

Error Handling Pattern Selection

Retry Configuration Guide

See Also

• 027 - Pipeline Monitoring and Observability - Monitoring error rates and alerts
• 025 - Data Quality: Validation Frameworks - Quality error handling
• 019 - Apache Kafka: Topics, Producers, and Consumers - DLQ implementation with Kafka
• 026 - Data Pipeline Testing - Testing error handling logic
• 028 - Error Handling, Retries, and Dead Letter Queues - This lesson