Implementing Dead Letter Queues in Kafka Consumers

When operating Apache Kafka at scale, platform and SRE teams inevitably encounter poison pill messages: records that cannot be processed despite repeated attempts. These failures stem from schema mismatches, corrupted payloads, transient downstream outages, or business-logic exceptions. Without a deliberate failure-handling strategy, a single malformed record can stall an entire consumer group, because the consumer cannot commit past an offset it never successfully processed. Implementing a Dead Letter Queue (DLQ) pattern gives you a systematic way to isolate problematic records, preserve them for forensic analysis, and let the primary flow continue.

This article covers the architectural decisions, implementation patterns, and operational concerns for building DLQs in Kafka consumer applications. We focus on manual consumer implementations using the Java client, where you retain full control over offset management and error routing: a critical distinction from managed frameworks that abstract these concerns away. The techniques here complement the broader delivery guarantees covered in Producers, Consumers & Delivery Semantics.

Understanding the Dead Letter Queue Pattern in Kafka

A Dead Letter Queue in Kafka is not a built-in broker feature; it is an application-level pattern. It consists of a separate topic (or set of topics) to which the consumer publishes records it cannot process, along with metadata describing why they failed. The consumer then commits the source-topic offset and moves on, so a single bad record never blocks the partition.

This is the fail-open middle ground: acknowledge the offset on the source topic, move the poison record to a side channel, and continue processing. It is the philosophy favoured in high-throughput environments where tail latency and pipeline stall are unacceptable. The alternative, fail-closed (halt on first unprocessable record), is correct only when strict ordering or zero-loss-without-isolation is a hard requirement, and even then the DLQ is usually the better building block.

When to Route to a DLQ

Not every processing failure warrants DLQ routing. Transient errors (network timeouts, temporary database unavailability, broker leader elections) should trigger retries with backoff. The DLQ becomes appropriate when:

  • Schema violations prevent deserialisation entirely, making retries futile.
  • Business-rule violations indicate the message is invalid regardless of delivery timing.
  • Idempotency violations occur despite correct producer configuration, suggesting a duplicate or out-of-sequence record that cannot be safely applied.
  • Exhausted retry budgets: after a configured maximum number of attempts, further retries would only delay processing of healthy records.

The decision to route to a DLQ should be explicit in code, not a catch-all for unhandled exceptions. Every catch block must distinguish between recoverable and terminal failures. This discipline prevents the DLQ from becoming a dumping ground for transient blips that self-resolve within seconds.

flowchart TD A["Process record"] --> B{"Failure?"} B -->|No| C["Commit source offset, continue"] B -->|"Terminal: schema/business/idempotency"| D["Route to DLQ"] B -->|Transient| E{"Retry budget exhausted?"} E -->|No| F["Retry with backoff"] F --> A E -->|Yes| D D --> G["After DLQ write acked: commit source offset"]

DLQ Topic Design Considerations

A dedicated DLQ topic should mirror the partitioning strategy of the source topic where ordering matters for analysis, but operators often choose a lower partition count for the DLQ to reduce overhead, since failed messages are usually a small fraction of total throughput. The replication factor must match your durability requirements: failed messages are often the most valuable for debugging, so losing them is unacceptable. Use replication-factor=3 with min.insync.replicas=2 to match the producer’s acks=all.

DLQ retention should be generous. Unlike source topics that age out according to business SLAs, DLQ messages may need to survive for weeks while root causes are investigated. Set retention.ms accordingly. Keep cleanup.policy=delete rather than compact for most DLQs: compaction retains only the latest record per key, which silently discards the first occurrence of a failure, usually the one you most want to inspect.

Implementing a DLQ Producer in the Consumer Application

The core implementation embeds a Kafka producer inside the consumer application, dedicated solely to publishing failed records. This producer must be configured for maximum reliability, because losing a DLQ message defeats the entire pattern.

Producer Configuration for DLQ Reliability

The DLQ producer operates under different constraints than a typical application producer; it must prioritise durability over throughput. The following snippet shows the critical settings:

Properties dlqProps = new Properties();
dlqProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092");
dlqProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
    "org.apache.kafka.common.serialization.ByteArraySerializer");
dlqProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
    "org.apache.kafka.common.serialization.ByteArraySerializer");
dlqProps.put(ProducerConfig.ACKS_CONFIG, "all");
dlqProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
dlqProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 30000);
dlqProps.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none");

KafkaProducer<byte[], byte[]> dlqProducer = new KafkaProducer<>(dlqProps);

Setting acks=all ensures the DLQ record is replicated to all in-sync replicas before the producer treats it as committed. Enabling idempotence (enable.idempotence=true) prevents duplicate DLQ entries when the producer retries internally, and it also sets safe defaults for you: acks=all, retries=Integer.MAX_VALUE, and max.in.flight.requests.per.connection=5. Note that with idempotence enabled, Kafka preserves per-partition ordering even with up to five in-flight requests, so there is no need to force max.in.flight.requests.per.connection=1 to keep failures from one source partition in order. Do not set retries=0 on this producer: that would disable the retries that protect against transient broker hiccups. For a deeper exploration of idempotence guarantees, refer to Idempotent Producers and Exactly-Once Semantics.

Constructing the DLQ Message

The DLQ message must carry enough context for operators to diagnose the failure without the original consumer logs. At minimum, include the original key, value, topic, partition, and offset. Augment this with failure metadata in the record headers:

public void sendToDLQ(ConsumerRecord<byte[], byte[]> original, Exception cause, int retryCount) {
    Headers headers = new RecordHeaders();
    headers.add("original-topic", original.topic().getBytes(StandardCharsets.UTF_8));
    headers.add("original-partition",
        String.valueOf(original.partition()).getBytes(StandardCharsets.UTF_8));
    headers.add("original-offset",
        String.valueOf(original.offset()).getBytes(StandardCharsets.UTF_8));
    headers.add("failure-timestamp",
        String.valueOf(System.currentTimeMillis()).getBytes(StandardCharsets.UTF_8));
    headers.add("exception-class", cause.getClass().getName().getBytes(StandardCharsets.UTF_8));
    headers.add("exception-message",
        (cause.getMessage() != null ? cause.getMessage() : "").getBytes(StandardCharsets.UTF_8));
    headers.add("retry-count", String.valueOf(retryCount).getBytes(StandardCharsets.UTF_8));

    ProducerRecord<byte[], byte[]> dlqRecord = new ProducerRecord<>(
        "orders-dlq", null, original.key(), original.value(), headers);

    dlqProducer.send(dlqRecord, (metadata, exception) -> {
        if (exception != null) {
            log.error("Failed to publish to DLQ. Original offset: {}",
                original.offset(), exception);
        }
    });
}

Note the null guard on cause.getMessage(): Throwable.getMessage() returns null for exceptions constructed without a message (for example, new NullPointerException()), and calling .getBytes() on that null would throw inside the DLQ path, defeating the whole point. Using ByteArraySerializer for both key and value avoids introducing a second serialisation format that could itself fail; the original bytes are preserved exactly as received, enabling bit-for-bit replay once the root cause is resolved.

Retry Logic and Backoff Strategies

Before routing to the DLQ, the consumer should attempt bounded retries with exponential backoff. The bound is essential: unbounded retries on a poison record block the partition indefinitely.

Implementing Bounded Retries with Backoff

The following example demonstrates an in-poll retry loop using a per-partition retry counter:

private final Map<TopicPartition, Integer> retryCounts = new ConcurrentHashMap<>();
private static final int MAX_RETRIES = 5;
private static final long BASE_BACKOFF_MS = 1000;

while (true) {
    ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<byte[], byte[]> record : records) {
        TopicPartition tp = new TopicPartition(record.topic(), record.partition());
        try {
            processRecord(record);
            retryCounts.remove(tp);
        } catch (TransientException e) {
            int attempts = retryCounts.merge(tp, 1, Integer::sum);
            if (attempts <= MAX_RETRIES) {
                long backoff = BASE_BACKOFF_MS * (long) Math.pow(2, attempts - 1);
                consumer.seek(tp, record.offset()); // re-read this record next poll
                Thread.sleep(backoff);
            } else {
                sendToDLQ(record, e, attempts);
                retryCounts.remove(tp);
            }
        } catch (TerminalException e) {
            sendToDLQ(record, e, 0);
            retryCounts.remove(tp);
        }
    }
    consumer.commitSync();
}

On a transient error this seeks back to the failing offset so the next poll() re-reads that record, effectively retrying the same record until it succeeds or the budget is exhausted. This is simpler than pause/resume, but it has two real caveats. First, seek only takes effect on the next poll(); the current for loop keeps iterating over the records already returned, so you typically break out of the loop after seeking rather than continue processing records you are about to re-read. Second, Thread.sleep blocks the poll loop for the whole consumer, not just the affected partition, which delays every other partition this consumer owns. For short backoffs that is acceptable; for long ones, prefer the pause/resume approach below.

Avoiding Consumer Group Rebalancing During Retries

Long backoffs risk triggering a rebalance if max.poll.interval.ms (default 300000 ms, i.e. 5 minutes) is exceeded between poll() calls. The consumer must return to poll() within that interval to remain in the group. When cumulative backoff approaches the threshold, reduce MAX_RETRIES, raise max.poll.interval.ms, or move retries off the poll thread entirely. The mechanics of group membership and rebalance protocols are detailed in Consumer Group Rebalancing: Protocols and Tuning.

A safer pattern for long backoffs pauses only the affected partition and resumes it after the backoff elapses, keeping the consumer polling (and thus alive in the group):

consumer.pause(Collections.singleton(tp));
scheduler.schedule(() -> consumer.resume(Collections.singleton(tp)),
    backoff, TimeUnit.MILLISECONDS);

Because pause, resume, seek, and poll must all be called from the same thread that owns the consumer, the scheduled resume here only sets a flag that is observed; in production, signal the poll thread to call resume rather than invoking it from the scheduler thread directly. The KafkaConsumer is not thread-safe.

Offset Management with DLQ Routing

Offset management becomes the crux of correctness once messages are routed to a DLQ. You must commit the source offset for a DLQ-routed message so it is not reprocessed on restart, but only after the DLQ write is durable, otherwise a failed DLQ write plus a committed offset equals silent data loss.

Commit Strategies for DLQ-Routed Messages

The safest approach commits the source offset only after the DLQ producer acknowledges the write. This gives at-least-once semantics for the DLQ: if the write fails, the offset is not committed and the record is reprocessed on restart. The following callback-based commit wires this together:

dlqProducer.send(dlqRecord, (metadata, exception) -> {
    if (exception == null) {
        consumer.commitSync(Collections.singletonMap(
            new TopicPartition(original.topic(), original.partition()),
            new OffsetAndMetadata(original.offset() + 1)));
    } else {
        log.error("DLQ write failed, offset not committed. Will retry on restart.");
    }
});

One important caveat: the send callback runs on the producer’s I/O thread, not the consumer’s poll thread, and KafkaConsumer is not thread-safe. Calling commitSync directly from the callback as shown can race with the poll loop. In practice, have the callback enqueue the offset into a thread-safe structure and let the poll thread commit it on its next iteration. The code above is the conceptual contract (commit only on success); the threading is the part you must get right in production.

Preventing Offset Gaps and Data Loss

If the consumer commits past a DLQ-routed message but the DLQ write fails silently, the original record is lost. Guard against transient failures with acks=all, idempotence, and a non-zero delivery.timeout.ms on the DLQ producer. But no amount of producer retries survives a permanent broker outage. In that case the consumer should halt and alert rather than keep advancing offsets and risk loss. The operational stance is explicit: a stalled pipeline is recoverable; silently dropped records are not.

Operationalising DLQ Topics

Once the pattern is in place, operational concerns shift to monitoring, alerting, and remediation. A DLQ with no consumers is a data sink where messages accumulate until retention expires. Decide up front whether to alert on any DLQ message or only past a threshold.

Monitoring and Alerting on DLQ Depth

The broker exposes per-topic ingress via the JMX MBean kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec, which has a topic tag. The Prometheus metric name you alert on depends entirely on your exporter: the Prometheus JMX Exporter, for example, lowercases and flattens the MBean into something like kafka_server_brokertopicmetrics_messagesinpersec (the exact name depends on your jmx_exporter rules). Confirm the name in your own /metrics endpoint before writing the rule below; the example assumes a metric exposed as a counter with a topic label.

groups:
  - name: kafka_dlq_alerts
    rules:
      - alert: HighDLQIngressRate
        expr: sum by (topic) (rate(kafka_server_brokertopicmetrics_messagesinpersec{topic=~".*-dlq"}[5m])) > 10
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "High DLQ ingress rate on {{ $labels.topic }}"
          description: "DLQ topic {{ $labels.topic }} is receiving {{ $value }} messages/sec."

Also monitor consumer lag on the DLQ topic. If a remediation consumer is replaying DLQ records, growing lag means the replay is falling behind.

Replaying DLQ Messages

Replay is the primary remediation workflow: inspect the DLQ records, fix the downstream system or schema, then republish the original payloads to the source topic. A simple replay utility reads from the DLQ and produces to the source topic, preserving the original key and value:

Properties consumerProps = new Properties();
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092");
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "dlq-replay-tool");
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
    "org.apache.kafka.common.serialization.ByteArrayDeserializer");
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
    "org.apache.kafka.common.serialization.ByteArrayDeserializer");
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

Properties producerProps = new Properties();
producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092");
producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
    "org.apache.kafka.common.serialization.ByteArraySerializer");
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
    "org.apache.kafka.common.serialization.ByteArraySerializer");
producerProps.put(ProducerConfig.ACKS_CONFIG, "all");
producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

try (KafkaConsumer<byte[], byte[]> replayConsumer = new KafkaConsumer<>(consumerProps);
     KafkaProducer<byte[], byte[]> replayProducer = new KafkaProducer<>(producerProps)) {

    replayConsumer.subscribe(Collections.singleton("orders-dlq"));

    while (true) {
        ConsumerRecords<byte[], byte[]> records = replayConsumer.poll(Duration.ofMillis(500));
        for (ConsumerRecord<byte[], byte[]> record : records) {
            String originalTopic = new String(
                record.headers().lastHeader("original-topic").value(), StandardCharsets.UTF_8);
            ProducerRecord<byte[], byte[]> replayRecord =
                new ProducerRecord<>(originalTopic, record.key(), record.value());
            replayProducer.send(replayRecord).get(); // block until acked before committing
        }
        replayProducer.flush();
        replayConsumer.commitSync();
    }
}

The critical detail is that Producer.send() is asynchronous: it returns a Future immediately and the broker ack arrives later. If you commit the DLQ consumer’s offset right after send() returns (without waiting), a subsequent producer failure would lose the record, because the offset was already advanced. Here send(...).get() blocks until each republish is acknowledged, and flush() before commitSync() is a belt-and-braces guarantee that the whole batch is durable before the DLQ offset is committed. For higher throughput, collect the per-record Futures and join them all before flush()/commit rather than blocking on each one. For large DLQ backlogs, consider archiving DLQ contents to object storage first; see Building a Dead Letter Queue with Kafka Connect and S3 Sink.

DLQ Topic Lifecycle Management

Create DLQ topics explicitly rather than relying on broker auto-creation, so partition count, replication factor, and retention are deliberate:

kafka-topics.sh --bootstrap-server broker1:9092 \
  --create --topic orders-dlq \
  --partitions 12 \
  --replication-factor 3 \
  --config retention.ms=2592000000 \
  --config cleanup.policy=delete

A 30-day retention (2592000000 ms) gives ample time for forensic analysis. Avoid cleanup.policy=compact unless the DLQ specifically needs deduplication, since compaction can drop the first occurrence of a failure, which is usually the most valuable for root-cause analysis.

Testing DLQ Behaviour Under Failure Conditions

A DLQ implementation is only as good as its behaviour under real failure modes. Testing must cover the failure of the DLQ producer itself, broker unavailability, and serialisation edge cases in the DLQ path, not just the happy path.

Simulating Broker Failures

To exercise the consumer’s behaviour when a DLQ write cannot reach enough replicas, stop one broker and observe that the consumer does not advance offsets for records it failed to DLQ.

A common pitfall here: kafka-server-stop.sh takes no broker-id argument in classic (single-host-per-broker) deployments. The script simply finds and signals the Kafka process(es) running on the host where you run it, so you stop a specific broker by running it on that broker’s host (or by kill-ing that broker’s PID). Only in KRaft combined-mode deployments, where a controller and broker share a host, did KIP-979 (Apache Kafka 3.7.0) add --process-role=<broker|controller> and --node-id=<id> flags to disambiguate which co-located process to stop. There is no --broker <broker-id> flag.

# On the host running the broker you want to take down (classic deployment):
kafka-server-stop.sh
# Or target its PID directly:
#   kill $(jps -l | grep kafka.Kafka | awk '{print $1}')

# In KRaft combined-mode, stop only the co-located broker process (Kafka 3.7+):
#   kafka-server-stop.sh --process-role broker

# Produce a poison message via the surviving brokers:
echo "corrupt-payload" | kafka-console-producer.sh \
  --bootstrap-server broker2:9092,broker3:9092 \
  --topic orders

# Observe consumer logs: ERROR on DLQ write, offset NOT committed, consumer keeps polling.

After restarting the broker, the consumer should retry the DLQ write on a later poll cycle and succeed, validating that the DLQ path degrades gracefully under partial broker availability.

Validating Serialisation Robustness

The DLQ path must never throw an unhandled exception, or it would crash the consumer thread. Test with data that exercises edge cases such as null keys and null exception messages:

@Test
public void testDLQHandlesNullExceptionMessage() {
    ConsumerRecord<byte[], byte[]> record = new ConsumerRecord<>(
        "orders", 0, 0L, null, "corrupt-value".getBytes(StandardCharsets.UTF_8));
    Exception cause = new NullPointerException(); // getMessage() == null
    assertDoesNotThrow(() -> dlqHandler.sendToDLQ(record, cause, 3));
}

ByteArraySerializer is inherently safe, but header values built from strings must be null-checked before getBytes(), which is exactly what the cause.getMessage() guard in sendToDLQ handles.

Integrating DLQs with Schema Registry

When using Confluent Schema Registry, deserialisation failures are a primary source of DLQ traffic. The Avro/Protobuf/JSON deserialiser throws a SerializationException when the payload does not conform to the registered schema. To accelerate diagnosis, capture the schema ID from the message and store it in the DLQ headers.

Capturing Schema Metadata on Failure

Confluent’s wire format prepends a 5-byte header to the payload: one magic byte (0x00), followed by a 4-byte big-endian schema ID, then the serialised payload. The following wrapper intercepts the failure and extracts that ID:

public byte[] deserializeWithDLQ(String topic, byte[] data) {
    try {
        return avroDeserializer.deserialize(topic, data);
    } catch (SerializationException e) {
        if (data != null && data.length >= 5 && data[0] == 0x0) {
            ByteBuffer buffer = ByteBuffer.wrap(data);
            byte magicByte = buffer.get();        // expected 0x00
            int schemaId = buffer.getInt();       // 4-byte big-endian schema ID

            Headers dlqHeaders = new RecordHeaders();
            dlqHeaders.add("schema-id",
                String.valueOf(schemaId).getBytes(StandardCharsets.UTF_8));
            dlqHeaders.add("magic-byte",
                String.valueOf(magicByte).getBytes(StandardCharsets.UTF_8));
        }
        sendToDLQ(new ConsumerRecord<>(topic, 0, 0L, null, data), e, 0);
        throw e; // re-throw to prevent processing the unreadable record
    }
}

ByteBuffer is big-endian by default, which matches the wire format, so no byte-order configuration is needed. The captured schema ID lets operators query the registry for the expected schema and diff it against the failing payload, which is the fastest path to diagnosing schema-evolution breakage. The Confluent Schema Registry API exposes endpoints to retrieve schemas by ID for automated DLQ inspection tooling.

Performance Implications and Tuning

The DLQ producer adds latency and resource consumption inside the consumer application. DLQ traffic is usually low volume, but the producer still consumes JVM heap, a network I/O thread, and contributes to GC pressure.

Resource Sizing for DLQ Producers

The DLQ producer maintains its own buffer (buffer.memory, default 32 MB) and a background I/O thread. In a consumer handling hundreds of thousands of records per second, even a 0.1% failure rate produces meaningful DLQ volume, so size the buffer to absorb bursts:

# DLQ producer tuning for a high-throughput consumer
buffer.memory=67108864          # 64 MB
batch.size=16384                # 16 KB (default)
linger.ms=5                     # brief linger to batch bursts
compression.type=none           # skip CPU overhead for low-volume DLQ

Watch the producer metrics buffer-available-bytes and waiting-threads (both under kafka.producer:type=producer-metrics,client-id={clientId}) to detect backpressure: waiting-threads rising above zero means application threads are blocking on buffer.memory. If the DLQ producer cannot keep up and your sendToDLQ blocks, that backpressure propagates into the main poll loop. In extreme cases, offload DLQ writes to a separate thread with a bounded queue, accepting that an overflowing queue may drop DLQ records: a trade-off that prioritises primary-pipeline liveness over forensic completeness.

Avoiding Head-of-Line Blocking

When one partition mixes healthy and poison records, retry-and-DLQ logic can block subsequent healthy records on that partition. The seek-based retry above makes this worse, since it re-reads from the failing offset. To mitigate, give each partition its own processing queue so healthy records are not stuck behind a slow one:

// Conceptual per-partition processing queue
Map<TopicPartition, BlockingQueue<ConsumerRecord<byte[], byte[]>>> partitionQueues;

// Main poll loop distributes records to per-partition queues
for (ConsumerRecord<byte[], byte[]> record : records) {
    TopicPartition tp = new TopicPartition(record.topic(), record.partition());
    partitionQueues.get(tp).offer(record);
}

// Per-partition worker threads handle retries and DLQ routing independently.

This decouples polling from processing so one partition’s DLQ routing does not delay another’s. It complicates offset management, though: within a partition, offsets must still be committed in order, so a stalled record blocks committing later offsets for that partition. You also must pause partitions whose queues fill up to avoid unbounded memory growth. The trade-off is worthwhile when cross-partition throughput matters more than minimising in-partition latency.

Conclusion

Implementing Dead Letter Queues in Kafka consumers is a foundational practice for production-grade stream processing. The pattern decouples error handling from the primary path, preserving both pipeline liveness and data integrity. The decisions that determine resilience are concrete: bounded retries with exponential backoff, a durable DLQ producer (acks=all, idempotence, non-zero delivery.timeout.ms), offset commits that happen only after the DLQ write is acknowledged, and monitoring of DLQ ingress and lag.

These techniques assume a manual consumer, giving you full control over the error-handling lifecycle. Managed frameworks like Kafka Streams and Kafka Connect offer built-in DLQ support, but understanding the underlying mechanics is what lets you debug and tune those abstractions. Test under realistic failure conditions, monitor continuously, and treat the DLQ as a first-class operational concern rather than an afterthought.

In this section

1 guide in this area.

Back to Producers Consumers Delivery Semantics