#Fallback Routing for Failed Migration Jobs

Geospatial migration pipelines fail in ways that plain ETL frameworks rarely anticipate. When converting legacy vector datasets — Shapefiles, FileGDB, KML — or raster archives into the columnar GeoParquet layout or FlatGeobuf, three distinct failure categories emerge: transient cloud throttling, deterministic data errors such as invalid geometries or missing CRS definitions, and systemic failures caused by memory exhaustion or schema drift. Without a structured routing layer, any of these collapses the entire batch run or, worse, silently writes corrupt chunks to the target location.

This guide walks through designing a production fallback router that intercepts every failure, classifies it deterministically, and directs execution to the correct recovery path — all without cascading halts or lost data lineage. It is part of the broader Data Conversion & Migration Pipelines framework.

#Prerequisites

Confirm the following before wiring up fallback routing:

• Python 3.10+ with a strict virtual environment (venv or conda).
• Core libraries: geopandas>=1.0, pyarrow>=14.0, shapely>=2.0, pyogrio>=0.7 — pin these versions in requirements.txt to avoid silent ABI breaks.
• Retry library: tenacity>=8.2 for bounded exponential backoff with jitter.
• Cloud SDKs: boto3 (AWS S3), gcsfs (GCP), or adlfs (Azure Data Lake) matching your target storage layer.
• Observability: structured JSON logging, OpenTelemetry tracing, and a metrics sink (Prometheus or CloudWatch) to track fallback rates and DLQ depth.
• Dataset profile: know your chunk size (recommended: 64 MB–256 MB per partition), geometry type distribution, and CRS metadata before designing routing thresholds.
• State store: a lightweight key-value service (DynamoDB, Cloud Firestore, or Redis) keyed by dataset_id:chunk_hash for deduplication and lineage tracking.

#Architectural Foundations

Fallback routing works as a decision layer inserted between the conversion executor and its output target. The central insight is that failure types have fundamentally different resolution strategies — retrying a geometry validation error wastes compute and delays escalation, while sending a throttled cloud request straight to a dead-letter queue wastes a recovery opportunity.

The diagram below shows how a single failed chunk propagates through the three-tier classification and routing system:

The routing layer integrates with building batch conversion pipelines with Python without introducing blocking I/O or thread contention — the classifier runs synchronously in the worker, while dispatch to recovery queues is non-blocking.

#Step-by-Step Workflow

#1. Intercept and Classify the Failure

Wrap the core conversion function in a structured exception handler that captures the exception type, stack trace, chunk identifier, and cloud request metadata. Classify failures into three tiers:

• Transient: network timeouts, HTTP 5xx responses, temporary lock contention, rate-limited API calls (AWS Throttling, GCP RESOURCE_EXHAUSTED).
• Data-specific: invalid geometries, missing CRS definitions, unsupported coordinate dimensions, Arrow schema violations, or corrupted binary blobs.
• Systemic: MemoryError, schema drift beyond tolerance thresholds, persistent storage unavailability, or disk exhaustion on the worker.

Classification drives routing. Transient errors trigger exponential backoff via retry logic for cloud migration pipelines. Data-specific and systemic failures require deterministic branching to specialist handlers. Correct schema mapping for legacy to modern formats upstream reduces the data-specific failure rate by catching type coercion issues, nullability violations, and precision loss before execution begins.

#2. Route to Recovery Pathways

Once classified, route the job to the appropriate handler. A robust routing engine uses a state machine or message queue (AWS SQS, GCP Pub/Sub, Kafka) to decouple failure handling from the primary conversion thread.

• Transient path: re-enqueue with an incremented retry counter. Apply jittered exponential backoff using tenacity. Cap retries at four attempts to avoid masking persistent infrastructure degradation.
• Data-specific path: isolate the problematic chunk, log the exact geometry or attribute failure, and route to a remediation queue. For malformed polygons or mixed geometry collections, apply shapely.make_valid, cast to MultiPolygon, or quarantine records for manual review.
• Systemic path: halt the worker, emit a critical alert, and route the job to a dead-letter queue. Safely quarantine affected partitions before triggering forensic analysis to prevent partial writes from polluting the target lakehouse.

#3. Execute Recovery and Validate Output

After routing, the recovery handler must guarantee idempotency. Use atomic write patterns — staging directories with final mv, or cloud-native CopyObject operations — to ensure partial conversions are never visible to downstream consumers. Validate recovered chunks against schema contracts and spatial integrity rules before committing. When a recovered chunk is re-written, keep its row group sizing and ZSTD compression level identical to the rest of the partition so a single remediated file never becomes an outlier that breaks predicate push-down on the target table.

Use the chunk_hash stored in your state store to detect duplicate messages. If a message is redelivered after a worker restart, the idempotency check prevents double-processing without requiring distributed locks.

#4. Instrument and Tune Routing Thresholds

Emit fallback-rate metrics, DLQ depth, and recovery latency as structured log events. Track the ratio of routed jobs to total jobs per dataset type. A spike in DataSpecificError routing usually indicates upstream ingestion corruption or schema drift — investigate the source before adjusting thresholds.

#Production-Ready Implementation

The following snippet demonstrates exception classification, routing dispatch, and idempotent recovery. All signatures are typed; library version requirements are noted inline.

# Requires: geopandas>=1.0, pyarrow>=14.0, tenacity>=8.2, boto3>=1.34
from __future__ import annotations

import logging
import os
from pathlib import Path
from typing import Literal

import geopandas as gpd
from botocore.exceptions import BotoCoreError, ClientError
from tenacity import (
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential,
)

logger = logging.getLogger(__name__)

RoutingTier = Literal["transient", "data_specific", "systemic", "unknown"]


# ---------------------------------------------------------------------------
# Custom exception taxonomy
# ---------------------------------------------------------------------------

class TransientError(Exception):
    """Recoverable infrastructure error — eligible for retry with backoff."""


class DataSpecificError(Exception):
    """Deterministic data error — route to remediation queue, do not retry."""


class SystemicError(Exception):
    """Unrecoverable worker condition — halt worker and escalate to DLQ."""


# ---------------------------------------------------------------------------
# Classification
# ---------------------------------------------------------------------------

def classify_exception(exc: Exception) -> RoutingTier:
    """Map a raw exception to a routing tier.

    Returns one of 'transient', 'data_specific', 'systemic', or 'unknown'.
    Extend the mapping as new exception patterns are observed in production.
    """
    if isinstance(exc, (BotoCoreError, ClientError)):
        code: str = (
            getattr(exc, "response", {})
            .get("Error", {})
            .get("Code", "")
        )
        if code in {"Throttling", "RequestLimitExceeded", "SlowDown", "503"}:
            return "transient"
        # Permanent S3 errors (NoSuchBucket, AccessDenied) are systemic
        return "systemic"

    if isinstance(exc, MemoryError):
        return "systemic"

    msg = str(exc).lower()
    if any(kw in msg for kw in ("invalid geometry", "crs", "wkb", "schema mismatch")):
        return "data_specific"

    if isinstance(exc, (TimeoutError, ConnectionError)):
        return "transient"

    return "unknown"


# ---------------------------------------------------------------------------
# Retry-decorated transient handler
# ---------------------------------------------------------------------------

@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    retry=retry_if_exception_type(TransientError),
    reraise=True,
)
def _attempt_conversion(chunk_path: str, staging_path: str) -> None:
    """Read a spatial chunk and write to a staging Parquet path.

    Raises TransientError to trigger tenacity retry.
    Raises DataSpecificError or SystemicError for non-retriable failures.
    """
    try:
        gdf = gpd.read_file(chunk_path)
    except Exception as exc:
        tier = classify_exception(exc)
        if tier == "transient":
            raise TransientError("Read failed — transient condition") from exc
        if tier == "data_specific":
            raise DataSpecificError(f"Cannot read chunk: {exc}") from exc
        raise SystemicError(f"Worker-level failure reading chunk: {exc}") from exc

    # Validate geometry before committing any I/O
    invalid_mask = ~gdf.geometry.is_valid
    if invalid_mask.any():
        raise DataSpecificError(
            f"{invalid_mask.sum()} invalid geometries in {chunk_path}"
        )

    gdf.to_parquet(staging_path, index=False, compression="zstd")


# ---------------------------------------------------------------------------
# Top-level router
# ---------------------------------------------------------------------------

def convert_and_route_chunk(
    chunk_path: str,
    target_path: str,
    *,
    remediation_queue: str = "geo-remediation",
    dlq_name: str = "geo-dlq",
) -> bool:
    """Convert a spatial chunk to GeoParquet with full fallback routing.

    Returns True on success, False if the chunk was routed to a queue.
    Guarantees idempotent writes via staging-then-promote pattern.
    """
    staging = f"{target_path}.staging"

    try:
        _attempt_conversion(chunk_path, staging)

        # Atomic promotion: rename staging -> final target
        Path(staging).replace(target_path)
        logger.info(
            "chunk_converted",
            extra={"chunk": chunk_path, "target": target_path},
        )
        return True

    except TransientError as exc:
        # tenacity exhausted all retries — treat as systemic now
        logger.error(
            "chunk_exhausted_retries",
            extra={"chunk": chunk_path, "error": str(exc)},
        )
        _send_to_dlq(chunk_path, reason=str(exc), queue=dlq_name)
        return False

    except DataSpecificError as exc:
        logger.warning(
            "chunk_routed_to_remediation",
            extra={"chunk": chunk_path, "error": str(exc)},
        )
        _send_to_remediation(chunk_path, reason=str(exc), queue=remediation_queue)
        return False

    except SystemicError as exc:
        logger.critical(
            "chunk_routed_to_dlq_systemic",
            extra={"chunk": chunk_path, "error": str(exc)},
        )
        _send_to_dlq(chunk_path, reason=str(exc), queue=dlq_name)
        raise  # Re-raise so the orchestrator halts this worker

    finally:
        # Always clean up staging artifact to prevent orphaned partial files
        if os.path.exists(staging):
            os.remove(staging)


# ---------------------------------------------------------------------------
# Queue dispatch stubs — replace with your message broker client
# ---------------------------------------------------------------------------

def _send_to_remediation(chunk_path: str, reason: str, queue: str) -> None:
    """Dispatch chunk metadata to the remediation queue."""
    logger.info("dispatch_remediation", extra={"queue": queue, "chunk": chunk_path, "reason": reason})
    # Example: sqs_client.send_message(QueueUrl=queue, MessageBody=json.dumps({...}))


def _send_to_dlq(chunk_path: str, reason: str, queue: str) -> None:
    """Dispatch chunk metadata to the dead-letter queue for forensic analysis."""
    logger.info("dispatch_dlq", extra={"queue": queue, "chunk": chunk_path, "reason": reason})
    # Example: sqs_client.send_message(QueueUrl=queue, MessageBody=json.dumps({...}))

#Reference Matrix: Failure Tier vs Routing Outcome

#State Machine Design

A production-grade fallback router relies on explicit state transitions rather than implicit exception propagation. Each chunk progresses through a finite set of states:

PENDING → PROCESSING → COMPLETED

or on failure:

PROCESSING → RETRYING → (success) COMPLETED | (exhausted) QUARANTINED

PROCESSING → REMEDIAL (data-specific)

PROCESSING → QUARANTINED (systemic)

Maintain the state store keyed by dataset_id:chunk_hash. This enables three capabilities:

• Deduplication: prevents duplicate processing when workers restart or messages are redelivered from the queue.
• Auditability: provides a complete lineage trail from ingestion through retry attempts to final commit or quarantine — critical for regulatory pipelines.
• Dynamic thresholds: allows platform teams to adjust routing rules (for example, lowering the transient retry ceiling during a known outage) without redeploying conversion code.

#Failure Modes and Gotchas

Bare except: blocks absorb systemic signals. Catching BaseException or using an untyped except: clause swallows MemoryError, KeyboardInterrupt, and SystemExit. Map exceptions explicitly to routing tiers and let anything uncategorised escalate.

Unbounded retries mask infrastructure degradation. Without a hard ceiling, a worker can spin in the transient retry path for hours while the underlying cloud region is impaired. Cap at four attempts and escalate to the DLQ — the DLQ replay mechanism is the correct channel for resuming after an outage.

Staging artifacts accumulate on worker crashes. The finally block in convert_and_route_chunk removes the staging file even when an exception escapes, but a hard worker kill (SIGKILL, OOM killer) skips finally. Add a startup sweep that scans for orphaned .staging files older than the maximum expected conversion time and either removes or re-queues them.

DLQ messages expire before remediation runs. Default SQS retention is 4 days; a long weekend outage can lose messages before the on-call team processes them. Set retention to 14 days and monitor DLQ age as a P2 alert.

CRS mismatch surfaces as a silent data error rather than an exception. If the source dataset has no CRS definition (common in legacy Shapefiles), geopandas reads it without error but pyproj will fail later during reprojection. Add an explicit CRS presence check — if gdf.crs is None: raise DataSpecificError(...) — immediately after reading the file.

Mixed geometry types cause Arrow schema conflicts. A chunk containing both Polygon and MultiPolygon geometries will fail when writing to a strict-schema Parquet partition. Normalise geometry type to the most general type in the chunk (gdf.geometry = gdf.geometry.apply(lambda g: g if g.geom_type == "MultiPolygon" else g.__class__([g]))) before the write step, or route to remediation for explicit type promotion.

#Observability and Continuous Tuning

Fallback routing is only as effective as its telemetry. Implement these monitoring practices as first-class pipeline concerns, not afterthoughts:

• Fallback rate by tier: track fallback_rate = routed_chunks / total_chunks broken down by tier. A sustained DataSpecificError rate above 5% signals upstream schema or geometry quality issues that will not resolve through routing alone.
• DLQ depth and age: alert when DLQ size exceeds a per-dataset baseline or when the oldest message exceeds 48 hours. Stale entries mean remediation automation is broken.
• Recovery latency: measure time from failure detection to successful reprocessing. Latency above 10 minutes for transient errors suggests retry backoff ceilings are too high for your workload’s SLA.
• Distributed tracing: propagate OpenTelemetry trace IDs from the initial ingestion request through retry attempts and DLQ archival. This makes it possible to correlate a quarantined chunk with the exact upstream batch that produced it.

Use these signals to tune routing thresholds dynamically. If the TransientError routing rate exceeds 15% during peak hours, automatically increase backoff ceilings or provision additional workers rather than lowering the retry ceiling.

#FAQ

How do I distinguish a transient network error from a permanent data error during migration?

Inspect the exception type and embedded status codes. Cloud SDK errors with codes like Throttling, SlowDown, or RequestLimitExceeded are transient. Geometry validation failures from Shapely or GDAL, CRS parse errors, and Arrow schema violations are data-specific and will not resolve on retry. Map each exception class to a routing tier in a centralised classify_exception function rather than scattering isinstance checks throughout your conversion code.

What is a safe maximum retry count before escalating to a dead-letter queue?

Four attempts with exponential backoff covering roughly 2–30 seconds per interval is a practical ceiling for transient errors in cloud object storage. Beyond four retries the failure pattern is almost always infrastructure degradation rather than a momentary blip, and continued retries mask the root cause. Escalate to a dead-letter queue after the final attempt so the chunk can be replayed when the underlying condition is resolved.

How do I prevent partial writes from poisoning a target lakehouse partition?

Always write to a staging path first (for example, target_key + '.staging'), then promote with an atomic CopyObject or rename operation only after the full chunk has been written and validated. Object storage rename is effectively atomic at the key level. If the worker dies mid-write, the staging object is never promoted, so downstream readers never observe a partial file.

Should fallback routing live in the same worker process as conversion, or in a separate service?

For small pipelines, embedding routing logic in the worker is fine. For production-scale jobs with thousands of chunks per run, decouple failure handling via a message queue (SQS, Pub/Sub, or Kafka). This lets you scale remediation workers independently, replay dead-letter messages without touching conversion code, and tune routing rules at runtime without redeployment.

#Related

• Building Batch Conversion Pipelines with Python — pipeline architecture this fallback router slots into
• Retry Logic for Cloud Migration Pipelines — deep dive into backoff configuration, jitter, and circuit breakers
• Schema Mapping for Legacy to Modern Formats — reduce data-specific failures upstream with robust type mapping
• Preserving Metadata During GeoParquet Conversion — maintain CRS, layer names, and attribute metadata through recovery paths

← Back to Data Conversion & Migration Pipelines