Why not retry a failing message forever?

Consumption is ordered per partition, so one message that always fails blocks every message behind it.

What should a dead-letter message contain?

The original key and value plus headers with the error type, message and source topic/partition/offset, so it can be diagnosed and replayed.

Which errors deserve retries?

Transient ones such as timeouts or unavailable dependencies; invalid data won't succeed on retry and should go to the DLQ.

Retries and Dead-Letter Topics — Kafka in Production

What you'll build

An order consumer where one message is malformed (a "poison pill") and another hits a temporary error. Transient failures are retried; the poison message goes to orders.dlq with headers describing the error, and processing of the partition continues.

The idea

A message that always fails would block its partition forever if you keep retrying.
Retry transient errors (timeouts, 503s) a few times with backoff; for longer delays, use retry topics.
Send messages that still fail, or can never succeed (invalid data), to a dead-letter topic (DLQ) with the original key, value and error details in headers, then commit and move on.
Monitor and replay the DLQ after fixing the cause.

The code

Retry transient errors, dead-letter poison messages

import json
import os
from confluent_kafka import Consumer, Producer
from confluent_kafka.admin import AdminClient, NewTopic

BOOTSTRAP = os.environ["KAFKA_BOOTSTRAP"]
admin = AdminClient({"bootstrap.servers": BOOTSTRAP})
for f in admin.create_topics([NewTopic(t, 1, 1) for t in ("orders", "orders.dlq")]).values():
    f.result()
producer = Producer({"bootstrap.servers": BOOTSTRAP, "enable.idempotence": True})
for value in ['{"id": 1, "total": 20}', '{"id": 2, "total": "abc"', '{"id": 3, "total": 35}', '{"id": 4, "total": 15}']:
    producer.produce("orders", key="shop-1", value=value)
producer.flush()

class TransientError(Exception):
    pass

attempts = {}
def process(order):
    attempts[order["id"]] = attempts.get(order["id"], 0) + 1
    if order["id"] == 3 and attempts[3] < 3:
        raise TransientError("inventory service timeout")
    return f"order {order['id']} reserved"

consumer = Consumer({"bootstrap.servers": BOOTSTRAP, "group.id": "fulfilment", "auto.offset.reset": "earliest", "enable.auto.commit": False})
consumer.subscribe(["orders"])
handled = 0
while handled < 4:
    msg = consumer.poll(5.0)
    if msg is None:
        break
    for attempt in range(1, 4):
        try:
            print(process(json.loads(msg.value())), f"(attempt {attempt})")
            break
        except TransientError as e:
            print(f"retrying after: {e}")
        except (ValueError, KeyError) as e:
            producer.produce("orders.dlq", key=msg.key(), value=msg.value(), headers={
                "error": type(e).__name__, "source": f"orders/{msg.partition()}/{msg.offset()}"})
            producer.flush()
            print(f"dead-lettered offset {msg.offset()}: {type(e).__name__}")
            break
    consumer.commit(msg, asynchronous=False)
    handled += 1
consumer.close()

dlq = Consumer({"bootstrap.servers": BOOTSTRAP, "group.id": "dlq-inspector", "auto.offset.reset": "earliest"})
dlq.subscribe(["orders.dlq"])
msg = dlq.poll(10.0)
print("DLQ message:", msg.value().decode(), dict((k, v.decode()) for k, v in msg.headers()))
dlq.close()
for f in admin.delete_topics(["orders", "orders.dlq"]).values():
    f.result()
for f in admin.delete_consumer_groups(["fulfilment", "dlq-inspector"]).values():
    f.result()

import json
import os
from confluent_kafka import Consumer, Producer
from confluent_kafka.admin import AdminClient, NewTopic

BOOTSTRAP = os.environ["KAFKA_BOOTSTRAP"]
admin = AdminClient({"bootstrap.servers": BOOTSTRAP})
for f in admin.create_topics([NewTopic(t, 1, 1) for t in ("orders", "orders.dlq")]).values():
    f.result()
producer = Producer({"bootstrap.servers": BOOTSTRAP, "enable.idempotence": True})
for value in ['{"id": 1, "total": 20}', '{"id": 2, "total": "abc"', '{"id": 3, "total": 35}', '{"id": 4, "total": 15}']:
    producer.produce("orders", key="shop-1", value=value)
producer.flush()

class TransientError(Exception):
    pass

attempts = {}
def process(order):
    attempts[order["id"]] = attempts.get(order["id"], 0) + 1
    if order["id"] == 3 and attempts[3] < 3:
        raise TransientError("inventory service timeout")
    return f"order {order['id']} reserved"

consumer = Consumer({"bootstrap.servers": BOOTSTRAP, "group.id": "fulfilment", "auto.offset.reset": "earliest", "enable.auto.commit": False})
consumer.subscribe(["orders"])
handled = 0
while handled < 4:
    msg = consumer.poll(5.0)
    if msg is None:
        break
    for attempt in range(1, 4):
        try:
            print(process(json.loads(msg.value())), f"(attempt {attempt})")
            break
        except TransientError as e:
            print(f"retrying after: {e}")
        except (ValueError, KeyError) as e:
            producer.produce("orders.dlq", key=msg.key(), value=msg.value(), headers={
                "error": type(e).__name__, "source": f"orders/{msg.partition()}/{msg.offset()}"})
            producer.flush()
            print(f"dead-lettered offset {msg.offset()}: {type(e).__name__}")
            break
    consumer.commit(msg, asynchronous=False)
    handled += 1
consumer.close()

dlq = Consumer({"bootstrap.servers": BOOTSTRAP, "group.id": "dlq-inspector", "auto.offset.reset": "earliest"})
dlq.subscribe(["orders.dlq"])
msg = dlq.poll(10.0)
print("DLQ message:", msg.value().decode(), dict((k, v.decode()) for k, v in msg.headers()))
dlq.close()
for f in admin.delete_topics(["orders", "orders.dlq"]).values():
    f.result()
for f in admin.delete_consumer_groups(["fulfilment", "dlq-inspector"]).values():
    f.result()

Output

order 1 reserved (attempt 1)
dead-lettered offset 1: JSONDecodeError
retrying after: inventory service timeout
retrying after: inventory service timeout
order 3 reserved (attempt 3)
order 4 reserved (attempt 1)
DLQ message: {"id": 2, "total": "abc" {'error': 'JSONDecodeError', 'source': 'orders/0/1'}

Run this on your computer — see “Get Started” for setup.

Common mistakes

Retrying a poison message forever, blocking the whole partition (and every key behind it).
Dead-lettering without context (no error, no source offset), making replays guesswork.
Nobody watching the DLQ; alert on its size.
Retrying non-idempotent side effects without deduplication.

Key points

Retry transient errors a bounded number of times.
Dead-letter unprocessable messages with error headers, then move on.
Monitor and replay the DLQ after fixing the cause.

Further reading: Kafka Connect: dead letter queues · Message headers (KIP-82)

What you'll build

The idea

The code

Common mistakes

Key points

Practice questions