What a poison message is and how it causes a consumer crash loop

A poison message is one that fails the same way on every delivery attempt. The cause is in the message itself, not in the environment. Typical examples are a payload that no longer matches the deserializer's schema, a field that triggers an unhandled null or parse error, or a record that violates an invariant the business logic assumes. Retrying it changes nothing because nothing about it changes between attempts.

The crash loop works like this. The consumer fetches the message, and processing throws an exception that escapes the handler or kills the process. Because the consumer never committed the offset (on Kafka) or never acknowledged the message (on a queue broker), the position does not move forward. The consumer restarts or rejoins its group. Kafka resumes from the last committed offset, and a queue broker redelivers the unacknowledged message. The consumer reads the same bytes and fails again.

On Kafka, every record behind the bad one in that partition stops moving, and consumer lag climbs steadily even though the consumer looks busy. Queue brokers show the same symptom in any queue or ordering group where redelivered messages block the ones behind them.

Adding more retries does not fix this. A retry policy cannot tell a poison message from a temporary database timeout, so it either gives up too early on recoverable errors or keeps spinning on unrecoverable ones. The real work is classifying each failure. This problem belongs to backend and platform engineers who own consumers on Kafka or queue brokers and who get paged when lag alarms fire.

 Kafka partition / broker queue (ordered)
 +-------+-------+-------+-------+-------+
 |  m41  |  m42  |  m43  |  m44  |  m45  |<-- producers keep appending
 +-------+-------+-------+-------+-------+
   done      |   |<------- stuck ------->|
             |     lag grows for everything behind m42
             |
             | (1) fetch / redeliver m42 (poison message);
             |     committed position still points here
             v
 +-----------------------------+
 | Consumer                    |<-------------------+
 | deserialize -> handle       |                    |
 +-----------------------------+                    |
             |                                      |
             | (2) exception escapes handler        |
             v                                      |
 +-----------------------------+                    |
 | Process crashes / restarts  |                    |
 | or rejoins consumer group   |                    |
 +-----------------------------+                    |
             |                                      |
             | (3) offset never committed /         |
             |     message never acked              |
             v                                      |
 +-----------------------------+  (4) same bytes,   |
 | Resume at committed offset  |      same failure  |
 | or redeliver unacked msg    |--------------------+
 +-----------------------------+

A poison message (m42) fails on every delivery, and because its offset is never committed or acknowledged, the consumer loops through fetch, failure, restart and redelivery while every record behind it in the partition or queue waits and lag grows.

Classifying failures: deterministic vs transient

The useful question is not "did processing fail?" but "would processing this exact message succeed if we tried again later, unchanged?" If the answer is no, retrying only burns time while the partition stalls behind it.

Failure class Typical signal Deterministic? Routing
Deserialization Malformed bytes, unknown magic byte, truncated payload Yes Quarantine immediately, no retry
Schema / validation Missing required field, unknown enum, constraint violation Yes Quarantine immediately; alert the producer's owner
Business logic Referenced entity absent, invariant violated, unhandled code path Usually Quarantine after zero or one retry; may become valid after a fix and replay
Transient infrastructure Timeouts, connection refused, 503, throttling, lock contention No Retry with backoff; pause consumption if the dependency stays down

A single retry policy treats every row the same, and that fails in two directions.

Transient classified as poison. A downstream database goes unavailable for ten minutes. Every message in that window fails, and each one lands in the quarantine topic. The dead-letter queue (DLQ) fills with valid messages. Someone then has to bulk-replay them, often out of order relative to traffic that arrived after the outage. The correct response to a dependency-wide failure is backpressure: pause, back off, and let consumer lag absorb the outage.

Poison classified as transient. A payload that always throws a null-reference error gets "retry with backoff, unlimited attempts." The consumer never commits past that offset. Lag grows, health checks may restart the pod, and the cycle repeats.

Classify by exception type and error code at the point of failure, not by attempt count alone. Attempt limits are a backstop for cases your classifier misses.

import json
from enum import Enum

class Route(Enum):
    QUARANTINE = "quarantine"        # deterministic: retrying the same bytes cannot succeed
    RETRY_ONCE = "retry_once"        # business logic: one retry, then quarantine for replay after a fix
    BACKOFF_AND_PAUSE = "backoff"    # transient: keep the offset, slow down, let lag absorb it

# App-defined exceptions raised by validation and domain code.
class SchemaValidationError(Exception): ...
class BusinessRuleError(Exception): ...
class DependencyUnavailable(Exception):
    def __init__(self, status=None):
        super().__init__(f"dependency unavailable status={status}")
        self.status = status

TRANSIENT_HTTP = {429, 502, 503, 504}

def classify(exc: BaseException) -> Route:
    # Decide by exception type at the point of failure, not by attempt count.
    if isinstance(exc, (json.JSONDecodeError, UnicodeDecodeError)):
        return Route.QUARANTINE          # malformed or truncated payload
    if isinstance(exc, SchemaValidationError):
        return Route.QUARANTINE          # also alert the producer's owner
    if isinstance(exc, BusinessRuleError):
        return Route.RETRY_ONCE
    if isinstance(exc, (TimeoutError, ConnectionError)):
        return Route.BACKOFF_AND_PAUSE   # never send these to the DLQ
    if isinstance(exc, DependencyUnavailable) and exc.status in TRANSIENT_HTTP:
        return Route.BACKOFF_AND_PAUSE
    # Unknown errors (e.g. a TypeError from a bad payload) are likely code bugs.
    # Treat as poison after a bounded retry so they cannot stall the partition.
    return Route.RETRY_ONCE

def should_quarantine(route: Route, attempts: int) -> bool:
    if route is Route.QUARANTINE:
        return True
    if route is Route.RETRY_ONCE:
        return attempts >= 2             # original attempt plus one retry
    return False                         # transient: pause consumption, do not dead-letter

A failure classifier that routes deserialization and schema errors straight to quarantine, gives business-logic and unknown errors one bounded retry, and keeps transient dependency errors out of the DLQ so the consumer backs off instead.

Detecting poison messages before they stall a partition

To enforce that backstop, the consumer needs to know how many times it has seen a message. Some brokers track this for you. SQS exposes a receive count, and RabbitMQ quorum queues track deliveries. Kafka does not, because consumption is an offset position, not a per-message lease. On Kafka, you either republish to a retry topic with an incremented attempt header, or you keep counts keyed by topic, partition, and offset in an external store.

Attempt counts are not the only signal. These others are often visible earlier:

  • Repeated exceptions on the same key or offset. If your error logs carry partition and offset, three identical stack traces at one offset point to a deterministic failure. That is stronger evidence than a high count alone.
  • Restarts correlated with a fixed committed offset. If the pod restarts and the committed offset does not move between restarts, you have a crash loop, not a slow consumer.
  • Skewed lag. Lag that grows on one partition while the others drain points to a single blocking message. Lag that grows evenly across partitions points to throughput or a downstream outage.
Signal Where it lives Trade-off
Broker delivery count Broker metadata Unavailable on Kafka
Attempt header Message headers Lost when a crash replays the original record from the log
External counter store Database or cache New dependency and a write on every attempt
Restart and offset correlation Metrics and alerts Detects the problem but does not isolate the message
Per-partition lag skew Lag monitoring Cheap to collect, but needs a threshold that tolerates hot keys

How to implement it: per-message boundaries, Kafka poison pills, and bounded retries

Put a boundary around every message

Wrap each record's handling in its own try/catch, with the poll loop outside it. An exception from one record should become a routing decision. It should never terminate the process. Without this boundary, the orchestrator restarts the consumer, the consumer resumes from the last committed offset, and the same record crashes it again.

Catch Kafka poison pills at the deserializer

A Kafka poison pill often fails before your handler runs. If the configured deserializer throws, the error comes out of the poll call and the consumer cannot get past that offset. Use a wrapping deserializer that catches the failure and returns an error record containing the raw bytes, topic, partition, offset, and exception. Your per-message boundary can then quarantine it like any other failure.

Commit only after a terminal outcome

Commit or ack an offset only once its message has been processed or quarantined. A crash before that point means redelivery, so handlers must be idempotent. Use dedupe keys, upserts, or conditional writes so that a replayed record produces the same result.

Bound retries in three tiers

  1. In-process backoff. Retry briefly with jitter, and only for errors classified as transient.
  2. Retry topics or delayed queues. Move longer waits off the main partition so healthy traffic keeps flowing.
  3. Attempt ceiling. After a fixed number of attempts, send the message to quarantine. Do not loop indefinitely.

Catch-all handling has a cost. A bad deploy can quietly quarantine every message while the consumer looks healthy. To catch this, alert on the quarantine rate and automatically pause consumption when that rate crosses a threshold.

// Wrapper: deserialization failures become data instead of exceptions thrown from poll().
record Parsed<T>(T value, byte[] raw, Exception error) {}

class SafeDeserializer<T> implements Deserializer<Parsed<T>> {
  private final Deserializer<T> inner;
  SafeDeserializer(Deserializer<T> inner) { this.inner = inner; }

  @Override public Parsed<T> deserialize(String topic, byte[] data) {
    try { return new Parsed<>(inner.deserialize(topic, data), data, null); }
    catch (Exception e) { return new Parsed<>(null, data, e); } // keep raw bytes for quarantine
  }
}

// Poll loop with enable.auto.commit=false. The loop stays outside the per-record boundary.
while (running) {
  ConsumerRecords<String, Parsed<Order>> batch = consumer.poll(Duration.ofMillis(500));
  for (ConsumerRecord<String, Parsed<Order>> rec : batch) {
    try {
      if (rec.value().error() != null) {
        quarantine(rec, rec.value().error());   // topic, partition, offset, raw bytes, cause
        continue;
      }
      handleWithBackoff(rec.value().value());   // tier 1: short jittered retries, transient errors only
    } catch (TransientException e) {
      int attempts = attemptsHeader(rec) + 1;   // attempt count travels in a record header
      if (attempts < MAX_ATTEMPTS) sendToRetryTopic(rec, attempts);  // tier 2: off the main partition
      else quarantine(rec, e);                                       // tier 3: hard ceiling
    } catch (Exception e) {
      quarantine(rec, e);                       // permanent failure: no retries
    }
    quarantineRate.check();                     // alerts and pauses consumption if the rate spikes
  }
  consumer.commitSync();                        // every record above is now processed or quarantined
}

A wrapping Kafka deserializer turns poison pills into error records, and each record has its own try/catch that retries transient errors, sends longer waits to a retry topic, enforces an attempt ceiling, and commits only after every record reaches a terminal outcome.

Designing message quarantine that supports reproduction and replay

A quarantine record is only useful if someone can reproduce the failure from it later. Store the original payload bytes exactly as received, without deserializing and re-serializing them. Re-encoding can normalize away the malformed field, wrong schema ID, or bad character encoding that caused the crash. Keep the original headers and key as well, then add metadata alongside them:

  • source topic or queue, partition, and offset
  • consumer group
  • exception class and a short stack trace summary
  • attempt count
  • quarantine timestamp

The offset and partition let an engineer locate neighboring messages. The exception class lets tooling group thousands of records into a handful of distinct bugs.

Operations. Alert on inflow rate, not total size. A queue holding 50,000 old records that nobody has touched is a backlog problem, while 200 new records in five minutes means something just broke. Give the quarantine an owning team and a retention policy. Without an owner, it fills up with records nobody reviews. Without retention, it grows indefinitely.

Replay. Send fixed or reprocessed messages back through the same validation and processing path as live traffic, never through a side door that skips checks. Replay should be idempotent and rate-limited so a bulk replay can't overwhelm downstream systems.

Sensitive data. Full payloads can include personal or regulated data that now sits in a second, often less-guarded store. Apply the same encryption, access controls, and deletion obligations as the source system. Restrict read access to the owning team and log who inspects records.

import time
import traceback
from confluent_kafka import Producer  # assumes the confluent-kafka client

DLQ_TOPIC = "orders.dlq"  # placeholder; owned by the orders team, with its own retention

def quarantine(producer: Producer, msg, exc: Exception, attempts: int, group_id: str) -> None:
    """Copy a failed message to the DLQ without touching its bytes."""
    # Keep the original headers so schema IDs and tracing context survive.
    headers = list(msg.headers() or [])

    # Short summary only: the exception type and the innermost frame, not a full dump
    # that could echo payload contents into logs and dashboards.
    tb = traceback.extract_tb(exc.__traceback__)
    where = f"{tb[-1].filename}:{tb[-1].lineno}" if tb else "unknown"

    headers += [
        ("dlq.source.topic", msg.topic().encode()),
        ("dlq.source.partition", str(msg.partition()).encode()),
        ("dlq.source.offset", str(msg.offset()).encode()),
        ("dlq.consumer.group", group_id.encode()),
        ("dlq.exception.class", type(exc).__qualname__.encode()),  # grouping key for triage
        ("dlq.stack.summary", where.encode()),
        ("dlq.attempts", str(attempts).encode()),
        ("dlq.quarantined.at.ms", str(int(time.time() * 1000)).encode()),
    ]

    # Pass msg.value() and msg.key() through as raw bytes. Never deserialize and
    # re-serialize them, because that can erase the defect you need to reproduce.
    producer.produce(DLQ_TOPIC, value=msg.value(), key=msg.key(), headers=headers)

    # Block until the DLQ write is acknowledged. Commit the source offset only after
    # flush() returns 0; otherwise the poison message could be lost entirely.
    if producer.flush(10) != 0:
        raise RuntimeError("DLQ write not confirmed; do not commit source offset")

A Python quarantine helper that forwards the original key and value bytes to a dead-letter topic unchanged, keeps the original headers, and adds headers for source topic, partition, offset, consumer group, exception class, stack summary, attempt count and timestamp.

When to avoid skipping: ordering-aware isolation and alternatives

Dead-lettering assumes each message stands alone. That breaks when consumers apply state transitions per key. Suppose an account's credit event is quarantined and the following debit is processed. The downstream balance is now wrong, and replaying the quarantined record later applies it out of order. Entity lifecycles such as created, updated, deleted are especially fragile. A skipped created turns every later event for that key into a second failure, or worse, into a silent upsert with partial data.

When order matters per key, the question is how much of the stream should wait for one bad record.

Strategy Availability Correctness Main risk
Skip to DLQ High: partition keeps flowing Broken for the affected key Downstream state silently diverges; replay applies events out of order
Park later messages for the same key High for other keys Preserved per key Parked backlog grows without bound; needs its own storage, ordering, and drain logic
Halt partition and alert Zero for every key on the partition Fully preserved One bad record becomes an outage for every key it shares a partition with
Keep retrying Same as halting, but quieter Preserved Looks like a slow consumer rather than a stuck one; lag alerts fire late

Parking by key is usually the right default for ordered domains. The consumer keeps a set of blocked keys. Any later message for a blocked key is diverted to the parked store in arrival order, and the whole sequence is released together once the head record is fixed. Halting fits only when partitions map to few keys, or when wrong output costs more than delayed output.

Impact on development cost and engineering time

The full design is not free. Retry topics multiply the topic count you must provision, monitor, and set retention on. Each retry tier needs its own consumer group, and lag alerts must be tuned per group. A DLQ is only useful with tooling to inspect, filter, and replay it. It also needs a named owner, or it becomes a silent graveyard. Expect the build cost to land up front and the operating cost to recur as on-call attention.

The return is avoided incident work. Without isolation, every poison message becomes a page, a diagnosis, and a manual offset change made under pressure. That last step is risky, because skipping the wrong offset loses valid data.

Illustrative estimate (assumed numbers, not a benchmark)

Annual hours saved = incidents per year × hours per incident × engineers involved

Assume 12 crash-loop incidents a year, 3 hours each, and 2 engineers per incident. That gives 12 × 3 × 2 = 72 hours. If the full design takes 80 hours to build and 20 hours a year to maintain, it pays back in a little over a year. The minimal setup, at about 16 hours, pays back within the first quarter.

Choosing the scope

Start minimal with a per-message catch, an attempt ceiling, and a DLQ. This is enough when traffic is low, ordering does not matter, and most failures are deterministic. Move to tiered retries, parked stores, and replay paths in these cases:

  • Transient failures are common.
  • Per-key ordering matters.
  • DLQ volume is high enough that manual triage becomes a recurring task of its own.

How OneEnv can help

Disclosure: OneEnv is the product behind this blog.

OneEnv (oneenv.io) is a workspace-based governance platform for the things services share — environment variables, database structure, API routes and validation rules. A change to a shared value is staged on a branch, reviewed once, and then every affected service explicitly approves or rejects it before it goes live, so you can see who a change touches before it ships instead of finding out in production.

Applied to poison message handling, this is where it helps:

Saves time

  • Every service that inherits a shared value is listed in one place, so nobody searches repos and chats to find out who is affected.
  • One review per branch replaces coordinating each team separately; each service still keeps its own approve/reject.
  • The dependency graph answers "what does this touch?" in seconds instead of a manual audit.

Saves cost

  • Fewer production incidents and rollbacks caused by a shared value, route or column changing under a consumer unannounced.
  • Less duplicated configuration to maintain: one shared definition inherited by many services.
  • You can size a change's blast radius before merging, avoiding surprise cross-team firefighting.
  • OneEnv has a free plan (1 workspace, unlimited services and members, full governance), so trying it costs nothing.

These are qualitative benefits, not measured results — how much you save depends on how many services share the value.

Want to try it? Start at oneenv.io.