Dead-Letter Topics & Poison-Message Handling for Trigger Streams

A single malformed geofence trigger can halt an entire partition. In a Kafka-style ordered log, a consumer that cannot process the message at offset 4,182,006 — because a NaN coordinate slipped past ingress validation, because a schema field the deserializer requires vanished in a rolling deploy, or because the downstream webhook returns a persistent HTTP 500 — has exactly two honest choices: block on that offset forever, or advance past it. Blocking converts one bad message into unbounded consumer lag that starves every well-formed ENTER/EXIT trigger queued behind it; naive skipping silently drops triggers that a compliance auditor will later demand. The dead-letter topic is the third option: quarantine the undeliverable message to a side channel, commit the offset, and keep the partition flowing. This page expands the routing and failure-isolation model introduced in event routing and backpressure, and the failure it addresses is head-of-line blocking under poison load — the traffic shape where one unprocessable record threatens the liveness of the whole stream.

The reader here is a backend or platform engineer draining a geofence-trigger stream through a Python consumer, deciding when to retry a failed trigger in place, when to bounce it through a delayed-retry topic, and when to give up and quarantine it. The distinction is not cosmetic: retrying a transient downstream timeout is correct, and retrying a structurally poisoned message is an infinite loop that pins a CPU and never commits an offset. Getting the classification right — transient versus terminal — is the entire discipline, and it is enforced by a delivery-attempt counter carried in the message headers and a retry ceiling that converts “retry forever” into “retry three times, then quarantine.”

Failure Taxonomy: Transient, Terminal, and the Poison Pill #

The first decision is classification, because the response is opposite for each class and a misclassification is expensive in both directions. A transient failure is one where retrying the identical bytes at a later time can succeed: a downstream consumer returning HTTP 503, a connection reset, a Redis dedup store momentarily unreachable, a producer-side RequestTimedOut. A terminal failure is one where retrying the identical bytes can never succeed: a payload that fails schema validation, a NaN or out-of-range coordinate that survived ingress, a trigger referencing a fence_id that no longer exists, an unparseable timestamp. Retrying a terminal failure is the canonical poison pill — the message is toxic to the consumer, and because a naive consumer treats every exception identically, it re-reads the same offset, throws the same exception, and never commits, pinning the partition at 100% CPU while lag climbs without bound.

The concrete numbers that anchor this design: a healthy geofence trigger stream should hold a DLQ rate below 0.01% of processed messages (one quarantine per 10,000 triggers) and a retry ceiling of 3 attempts before quarantine. A DLQ rate above 0.1% is a code-red incident — it means either an upstream producer is emitting structurally invalid triggers or a downstream dependency is hard-down and the retry classifier is mislabeling its outage as terminal. Both figures are budgets you alert on, not aspirations. The classifier that separates the two classes is the highest-leverage code in the consumer, and it must default to terminal for unrecognized exceptions rather than transient: an unknown error retried three times and quarantined costs three wasted attempts, but an unknown structural error misclassified as transient and retried forever costs the partition.

Terminal classification also carries an idempotency subtlety. A trigger that failed after partially executing its side effect — the webhook was delivered but the acknowledgement was lost — must not be treated as a clean terminal failure, because quarantining and later replaying it double-fires the trigger. This is why dead-lettering is inseparable from idempotent trigger emission semantics: the deterministic idempotency key is what makes a later DLQ replay safe, and without it the DLQ becomes a duplicate-generation machine.

Algorithmic Divergence: Retry-in-Place vs Side-Topic Retry vs Immediate DLQ #

Three strategies compete for a failed message, and they differ in blocking behavior, ordering guarantees, and how they interact with the consumer’s poll loop. The cost model is about who waits and for how long.

Retry-in-place re-invokes the handler synchronously within the poll loop, sleeping a backoff interval between attempts. Its latency for a message that ultimately succeeds on attempt $n$ is ${\sum }_{i=0}^{n-1}b\cdot 2^i=b(2^n-1)$ for base backoff $b$ — an exponential wall. With $b=200\text{ms}$ and a ceiling of 3, a message that fails twice then succeeds blocks its partition for $200+400=600\text{ms}$, during which no other message on that partition is processed. Retry-in-place preserves strict ordering but converts every retried message into head-of-line latency for its neighbors, and it must be bounded tightly or it recreates the poison-pill stall it was meant to avoid.

Side-topic retry commits the source offset immediately and republishes the failed message to a dedicated delayed-retry topic (often tiered: retry-5s, retry-30s, retry-5m), where a separate consumer with a matching consume delay re-attempts it out of band. This unblocks the source partition instantly — the tradeoff is that it sacrifices ordering for the retried minority, because a retried ENTER can now land after a later EXIT for the same device. That reordering is only safe because the downstream consumer is idempotent and transition-aware.

Immediate DLQ skips retry entirely and quarantines on first failure. It is correct exactly when the failure is provably terminal — the classifier is certain no retry can help — and wrong for anything transient, since it turns a 200ms blip into a permanent quarantine.

Production geofence pipelines almost always run a hybrid: bounded retry-in-place for fast transients (a single 100–200ms backoff to ride out a connection blip), a tiered side-topic for slow transients (a downstream that is degraded for minutes), and immediate DLQ for anything the classifier calls terminal. The retry ceiling of 3 spans in-place plus side-topic attempts combined, tracked by a delivery-attempt header so the count survives the republish.

Implementation Reference #

The critical path is the consumer’s per-message resolution: classify the failure, consult the attempt counter in the headers, and route to commit, retry, or DLQ. The code below is the production shape — the attempt count travels in a Kafka header so it is preserved across the source topic and the retry topic, and the DLQ record is enriched with the failure reason, the original topic/partition/offset, and a stack digest so a later operator can triage without re-deriving context.

from __future__ import annotations

import asyncio
import json
import logging
import math
import time
from dataclasses import dataclass
from enum import Enum

from aiokafka import AIOKafkaConsumer, AIOKafkaProducer
from aiokafka.structs import ConsumerRecord

logger = logging.getLogger(__name__)

RETRY_CEILING: int = 3          # attempts across in-place + side-topic before DLQ
BASE_BACKOFF_S: float = 0.2     # first in-place retry delay; doubles per attempt
DLQ_TOPIC: str = "geofence.triggers.dlq"
RETRY_TOPIC: str = "geofence.triggers.retry"


class Disposition(Enum):
    COMMIT = "commit"       # succeeded or provably terminal -> advance offset
    RETRY = "retry"         # transient, attempts remain
    DEAD_LETTER = "dlq"     # exhausted or poison -> quarantine


class TerminalError(Exception):
    """Structurally unprocessable: NaN coords, schema drift, unknown fence."""


class TransientError(Exception):
    """Retry of identical bytes may succeed: downstream 5xx, timeout, reset."""


@dataclass(slots=True)
class Trigger:
    device_id: str
    fence_id: str
    transition: str          # ENTER | EXIT | DWELL
    event_time_ns: int
    idempotency_key: str     # deterministic; makes a later DLQ replay safe


def _attempt_count(record: ConsumerRecord) -> int:
    # The counter rides in a header so it survives republish to the retry topic.
    for key, value in record.headers or ():
        if key == "x-delivery-attempt":
            return int(value.decode())
    return 0


def _decode_or_terminal(record: ConsumerRecord) -> Trigger:
    # A decode/validate failure is terminal by construction: identical bytes
    # will never parse. Raising TerminalError routes straight to the DLQ.
    try:
        raw = json.loads(record.value)
        lat, lon = float(raw["lat"]), float(raw["lon"])
        if math.isnan(lat) or math.isnan(lon) or not (-90 <= lat <= 90):
            raise TerminalError(f"bad coords lat={lat} lon={lon}")
        return Trigger(
            device_id=str(raw["device_id"]),
            fence_id=str(raw["fence_id"]),
            transition=str(raw["transition"]),
            event_time_ns=int(raw["event_time_ns"]),
            idempotency_key=str(raw["idempotency_key"]),
        )
    except (KeyError, ValueError, TypeError) as exc:
        raise TerminalError(f"schema drift: {exc}") from exc


class DeadLetterConsumer:
    def __init__(self, producer: AIOKafkaProducer) -> None:
        self._producer = producer
        self._metrics: dict[str, float] = {"processed": 0.0, "retried": 0.0, "dlq": 0.0}

    async def _deliver(self, trigger: Trigger) -> None:
        # Placeholder for the real downstream call (webhook / gRPC / DB write).
        # A 5xx or timeout here must raise TransientError; a 4xx that means
        # "this trigger is invalid" must raise TerminalError.
        raise NotImplementedError

    async def _handle(self, record: ConsumerRecord) -> Disposition:
        attempt = _attempt_count(record)
        try:
            trigger = _decode_or_terminal(record)   # terminal path short-circuits
            await self._deliver(trigger)
            self._metrics["processed"] += 1
            return Disposition.COMMIT
        except TerminalError as exc:
            logger.error("poison message at %s-%s@%s: %s",
                         record.topic, record.partition, record.offset, exc)
            await self._to_dlq(record, reason=f"terminal:{exc}")
            return Disposition.COMMIT              # commit source; message is quarantined
        except TransientError as exc:
            if attempt + 1 >= RETRY_CEILING:
                logger.warning("retry ceiling hit at offset %s: %s", record.offset, exc)
                await self._to_dlq(record, reason=f"exhausted:{exc}")
                return Disposition.COMMIT
            await self._to_retry(record, attempt + 1)
            return Disposition.COMMIT              # offset committed; retry is out-of-band
        except Exception as exc:                    # noqa: BLE001 default to terminal
            # Unknown errors are quarantined, NOT retried forever. Three wasted
            # attempts is cheap; an infinite loop on a poison pill is not.
            logger.exception("unclassified failure at offset %s", record.offset)
            await self._to_dlq(record, reason=f"unclassified:{exc}")
            return Disposition.COMMIT

    async def _to_retry(self, record: ConsumerRecord, next_attempt: int) -> None:
        await self._producer.send_and_wait(
            RETRY_TOPIC, value=record.value, key=record.key,
            headers=[("x-delivery-attempt", str(next_attempt).encode())],
        )
        self._metrics["retried"] += 1

    async def _to_dlq(self, record: ConsumerRecord, *, reason: str) -> None:
        # Enrich so triage needs no archaeology: origin, reason, and a timestamp.
        envelope = {
            "origin_topic": record.topic,
            "origin_partition": record.partition,
            "origin_offset": record.offset,
            "reason": reason,
            "quarantined_at_ns": time.time_ns(),
            "payload_b64": record.value.decode("latin1"),
        }
        await self._producer.send_and_wait(
            DLQ_TOPIC, value=json.dumps(envelope).encode(), key=record.key,
        )
        self._metrics["dlq"] += 1

Two decisions carry the weight. First, every branch returns Disposition.COMMIT for the source offset — success, retry, and quarantine all advance the source partition, because the retry and DLQ records are durably persisted to their own topics before the source offset moves. The offset never blocks. Second, the bare except Exception defaults to the DLQ, not to retry: an engineer who forgets to classify a new error type gets a quarantined message and a DLQ-rate alert, not a wedged partition. The delayed-retry consumer (not shown) is a near-mirror that consumes RETRY_TOPIC with a consume delay and reuses _handle, so the attempt ceiling is enforced uniformly.

Memory Footprint & Streaming Churn #

The DLQ path is deliberately allocation-light on the hot path — the common case is Disposition.COMMIT on first success, which allocates only the decoded Trigger (a slots=True dataclass at ~120 bytes) and no envelope. The envelope construction, base64 of the payload, and JSON re-serialization only run on the failure path, so their cost is gated by the DLQ rate: at the 0.01% budget across 25k triggers/sec, the DLQ producer sees ~2.5 records/sec, and its serialization churn is invisible against the main stream. This asymmetry is the point — you pay for quarantine metadata only when you quarantine.

The dangerous churn is in the retry topic under a downstream outage. If a dependency goes hard-down and the classifier correctly labels every delivery TransientError, the retry topic fills at nearly the full ingest rate, and the tiered retry consumers hold those messages in flight through their backoff delays. A retry-5m tier draining 25k/sec of failed triggers accumulates 7.5M in-flight records over its five-minute delay window — enough to matter for broker disk and for the retry consumer’s own memory. The mitigation is the retry ceiling doing its job: with a ceiling of 3, a persistently failing message reaches the DLQ within three tiers rather than cycling the retry topic forever, and the DLQ itself is size- and time-bounded by a topic retention of 7–14 days so quarantined triggers are triaged or expired, never accumulated indefinitely. Pair this with the backpressure and flow-control strategies that shed load upstream when the retry topic backlog signals a downstream that cannot keep up.

Async Mutation Boundaries & Queue Semantics #

The consumer must never let the DLQ producer become its own blocking dependency. The send_and_wait calls in _to_dlq and _to_retry are awaited, which means a slow or unavailable DLQ broker can itself stall the poll loop — a quarantine path that blocks is a poison pill with extra steps. Two guards keep the boundary safe. First, the producer runs with bounded in-flight batching (max.in.flight.requests) and its own timeout, so a DLQ send that cannot complete fails fast rather than hanging. Second, the resolution is idempotent at the DLQ level: because the DLQ record carries the origin offset and the trigger’s idempotency key, a duplicate quarantine caused by a retried _handle after a crash is deduplicated downstream rather than double-counted against the DLQ-rate alert.

Offset commit ordering is the correctness invariant. The source offset must commit only after the retry or DLQ record is acknowledged by the broker — commit-then-produce would drop a message on a crash between the two, and produce-then-commit at-least-once may duplicate the DLQ record, which the idempotency key absorbs. This is a deliberate at-least-once boundary; the tradeoffs of pushing it to exactly-once are covered in exactly-once vs at-least-once trigger delivery. When the DLQ producer is itself degraded, trip a circuit breaker that pauses the source consumer (consumer.pause()) rather than dropping triggers — pausing preserves at-least-once, whereas continuing without a working DLQ silently loses the poison messages you were trying to preserve.

Operational Runbook & Failure Mitigation #

When the DLQ rate alert fires, the cause is one of a small set of failure modes. Diagnose with py-spy dump on the live consumer for a wall-clock stack, the DLQ envelope’s reason field aggregated by prefix, and consumer-lag metrics per partition.

The standing diagnostic loop:

1. Confirm the rate. Pull dlq_rate = dlq / processed from Prometheus. Below 0.01% is nominal; between 0.01% and 0.1% is a warning to triage the top reason; above 0.1% is a page.
2. Attribute by reason. Group the last hour of DLQ envelopes by the reason prefix (terminal: / exhausted: / unclassified:). A spike in terminal: points upstream; a spike in exhausted: points downstream; any unclassified: is a classifier gap to fix.
3. Localize the source. Group by origin_topic/origin_partition. A single producer or partition dominating the DLQ isolates the fault to one emitter rather than a systemic issue.
4. Verify no partition is stalled. Diff per-partition consumer lag. A poison-pill regression shows as one partition’s lag climbing while siblings stay flat — the signature that a message is being retried in place rather than dead-lettered.
5. Confirm replay-safety before draining. Never drain the DLQ back into the main stream without confirming each trigger’s idempotency key is intact, or you convert quarantined triggers into duplicates. The safe procedure is in replaying dead-letter triggers safely.
6. Quantify GC pressure on the retry path. Under a downstream outage the retry/DLQ serialization churn rises; read gc.get_stats() and confirm gen-2 pauses stay under 10ms, pooling the envelope buffers if serialization dominates.

Continuous monitoring should target a sustained DLQ rate under 0.01%, zero unclassified: reasons, and per-partition lag within 2× of the fleet median — a single partition drifting is the earliest sign of a pre-DLQ poison stall. Wire these into the broader production monitoring and observability surface so the DLQ rate sits beside queue depth and P99 latency on one board.

Architectural Guidance: When to Choose Which #

The routing decision is driven by two axes — whether the failure is provably terminal, and how strict the downstream ordering requirement is.

The invariant across every variant is that the source partition never blocks: a message resolves to committed, re-enqueued, or quarantined within a bounded number of attempts, and in all three the source offset advances. Dead-lettering is controlled loss with a paper trail, not silent drop — every quarantined trigger is durable, enriched, and replayable, so the choice to give up on a message is always reversible. Instrument the DLQ rate as a first-class SLO, keep the classifier’s default terminal, and hold the retry ceiling low, and one poison message costs one quarantine record instead of one stalled stream.

Frequently Asked Questions #

Related #

• Event Routing & Backpressure — parent overview of trigger routing, flow control, and failure isolation.
• Replaying Dead-Letter Geofence Triggers Safely — idempotent, rate-limited drain of the DLQ back into the stream.
• Idempotent Trigger Emission Semantics for Geofencing — the deterministic keys that make DLQ replay safe.
• Backpressure & Flow-Control Strategies — shedding upstream load when the retry backlog signals a slow downstream.
• Production Monitoring & Observability — surfacing DLQ rate beside queue depth and P99 latency.