Definition
- Dead-letter queue (DLQ) -- a separate topic where a consumer sends messages it cannot process after exhausting retries, so a single bad message does not block or get dropped.
- Poison pill -- a record that always fails, like malformed data or a business rule it can never satisfy; the DLQ pattern moves it aside safely.
- Retry classification -- transient failures (timeouts, network) are retried with backoff; permanent failures (deserialization, validation) go straight to the DLQ.
- Failure metadata -- the DLQ record preserves the original key and value and adds headers with the exception, original topic, partition, offset, and timestamp.
- Commit ordering -- commit the source offset only after the record is either processed successfully or confirmed written to the DLQ.
Imagine a settlement clerk at National Bank of India who hits a transfer slip that is torn and unreadable. If the clerk keeps retrying forever, the whole tray stalls. Instead, after a couple of attempts, the clerk places the bad slip in a clearly labelled problem tray with a note explaining what went wrong, then moves on. Nothing is lost and nothing is blocked, and a supervisor can later examine the problem tray.
Understanding the Concepts
What problem does a dead-letter queue solve?
The core problem is the poison pill: a single record that always fails to process. Without a DLQ, a naive consumer that retries forever blocks the partition, and one that skips it loses the data silently. The DLQ gives a safe third option: move the bad record aside and continue.
package com.bank.kafka;
public class PoisonPillProblem {
public static void main(String[] args) {
System.out.println("Without DLQ:");
System.out.println(" Retry forever -> partition blocked, lag grows");
System.out.println(" Skip silently -> data lost, no visibility");
System.out.println();
System.out.println("With DLQ:");
System.out.println(" Move bad record to DLQ -> partition advances, data preserved");
System.out.println(" Supervisor inspects DLQ -> fix and redrive");
}
}
Expected output:
Without DLQ:
Retry forever -> partition blocked, lag grows
Skip silently -> data lost, no visibility
With DLQ:
Move bad record to DLQ -> partition advances, data preserved
Supervisor inspects DLQ -> fix and redrive
How do you implement a DLQ consumer?
The consumer disables auto-commit, processes each record, retries transient failures, and on permanent failure or retry exhaustion sends to the DLQ topic, then commits.
package com.bank.kafka;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
public class DlqConsumer {
private static final String SOURCE_TOPIC = "transfers";
private static final String DLQ_TOPIC = "transfers.DLT";
private static final int MAX_RETRIES = 3;
public static void main(String[] args) {
// Consumer config
Properties consumerProps = new Properties();
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092");
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "settlement-service");
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "50");
// DLQ producer config
Properties producerProps = new Properties();
producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092");
producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
producerProps.put(ProducerConfig.ACKS_CONFIG, "all");
producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
KafkaProducer<String, String> dlqProducer = new KafkaProducer<>(producerProps)) {
consumer.subscribe(Collections.singletonList(SOURCE_TOPIC));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
for (ConsumerRecord<String, String> record : records) {
try {
processTransfer(record);
} catch (PermanentFailureException e) {
sendToDlq(dlqProducer, record, e, 0);
} catch (Exception e) {
handleWithRetry(dlqProducer, record, e);
}
}
if (!records.isEmpty()) {
consumer.commitSync();
}
}
}
}
private static void processTransfer(ConsumerRecord<String, String> record) {
// Parse and validate the transfer
System.out.println("Processing: " + record.key() + " -> " + record.value());
if (record.value() == null || record.value().isEmpty()) {
throw new PermanentFailureException("Empty transfer payload");
}
// Apply to ledger...
}
private static void handleWithRetry(
KafkaProducer<String, String> dlqProducer,
ConsumerRecord<String, String> record, Exception e) {
for (int attempt = 1; attempt <= MAX_RETRIES; attempt++) {
try {
Thread.sleep(attempt * 1000L); // Backoff
processTransfer(record);
return; // Retry succeeded
} catch (Exception retryEx) {
System.out.println("Retry " + attempt + "/" + MAX_RETRIES + " failed");
}
}
sendToDlq(dlqProducer, record, e, MAX_RETRIES);
}
private static void sendToDlq(
KafkaProducer<String, String> dlqProducer,
ConsumerRecord<String, String> record, Exception e, int attempts) {
ProducerRecord<String, String> dlqRecord =
new ProducerRecord<>(DLQ_TOPIC, record.key(), record.value());
// Add failure metadata as headers
dlqRecord.headers().add(new RecordHeader("original-topic",
SOURCE_TOPIC.getBytes(StandardCharsets.UTF_8)));
dlqRecord.headers().add(new RecordHeader("original-partition",
String.valueOf(record.partition()).getBytes(StandardCharsets.UTF_8)));
dlqRecord.headers().add(new RecordHeader("original-offset",
String.valueOf(record.offset()).getBytes(StandardCharsets.UTF_8)));
dlqRecord.headers().add(new RecordHeader("exception-class",
e.getClass().getName().getBytes(StandardCharsets.UTF_8)));
dlqRecord.headers().add(new RecordHeader("exception-message",
e.getMessage().getBytes(StandardCharsets.UTF_8)));
dlqRecord.headers().add(new RecordHeader("retry-attempts",
String.valueOf(attempts).getBytes(StandardCharsets.UTF_8)));
try {
dlqProducer.send(dlqRecord).get(); // Wait for ack
System.out.println("Sent to DLQ: " + record.key() + " reason: " + e.getMessage());
} catch (Exception dlqEx) {
// If DLQ write fails, do NOT commit -- record will be retried
throw new RuntimeException("DLQ write failed", dlqEx);
}
}
}
class PermanentFailureException extends RuntimeException {
public PermanentFailureException(String message) { super(message); }
}
Expected output:
Processing: SAV2001 -> DEBIT 50000
Processing: SAV3001 -> (empty payload)
Sent to DLQ: SAV3001 reason: Empty transfer payload
How do you classify retryable vs permanent failures?
A transient failure like a downstream timeout should be retried because it may succeed. A permanent failure like deserialization or validation should go straight to the DLQ because retrying will never help.
| Exception Type | Classification | Action |
|---|---|---|
| DeserializationException | Permanent | DLQ immediately |
| ValidationException | Permanent | DLQ immediately |
| IOException / timeout | Transient | Retry with backoff |
| Database connection error | Transient | Retry with backoff |
| Retries exhausted | N/A | DLQ after max attempts |
What metadata should the DLQ record contain?
Preserve the original key and value so the message can be reprocessed later, and add headers with diagnostic context.
original-topic-- where the record came fromoriginal-partition-- which partitionoriginal-offset-- the exact offsetexception-class-- what went wrongexception-message-- error detailsretry-attempts-- how many retries were triedtimestamp-- when the failure happened
How do you handle DLQ write failures?
If the DLQ write itself fails, do not commit the source offset. The record will be retried on the next poll, keeping the guarantee that every record is either processed or dead-lettered.
package com.bank.kafka;
public class DlqWriteFailure {
public static void main(String[] args) {
System.out.println("DLQ write ordering:");
System.out.println(" 1. Try process record");
System.out.println(" 2. On failure: send to DLQ, wait for ack");
System.out.println(" 3. DLQ ack received: commit source offset");
System.out.println(" 4. DLQ write FAILS: do NOT commit -> retry on next poll");
System.out.println();
System.out.println("Guarantee: every record is either processed OR dead-lettered");
}
}
Expected output:
DLQ write ordering:
1. Try process record
2. On failure: send to DLQ, wait for ack
3. DLQ ack received: commit source offset
4. DLQ write FAILS: do NOT commit -> retry on next poll
Guarantee: every record is either processed OR dead-lettered
How does Spring Kafka handle DLQ declaratively?
In Spring Kafka the whole pattern is available with DefaultErrorHandler and DeadLetterPublishingRecoverer, so you rarely hand-write it.
package com.bank.kafka;
// Spring Kafka declarative DLQ (pseudo-code)
public class SpringDlqConfig {
public static void main(String[] args) {
System.out.println("Spring Kafka DLQ:");
System.out.println(" DefaultErrorHandler errorHandler = new DefaultErrorHandler(");
System.out.println(" new DeadLetterPublishingRecoverer(kafkaTemplate),");
System.out.println(" new FixedBackOff(1000L, 3L) // 3 retries, 1s backoff");
System.out.println(" );");
System.out.println(" // Automatically retries, then publishes to DLT topic");
}
}
Expected output:
Spring Kafka DLQ:
DefaultErrorHandler errorHandler = new DefaultErrorHandler(
new DeadLetterPublishingRecoverer(kafkaTemplate),
new FixedBackOff(1000L, 3L) // 3 retries, 1s backoff
);
// Automatically retries, then publishes to DLT topic
What are common mistakes with DLQ implementations?
The most dangerous mistake is committing the source offset before confirming the DLQ write, which loses the record on DLQ failure. Another common error is not classifying exceptions, so transient timeouts go straight to the DLQ instead of being retried. A third mistake is treating the DLQ as a dumping ground without monitoring or alerts, letting it grow silently.
package com.bank.kafka;
public class DlqMistakes {
public static void main(String[] args) {
System.out.println("WRONG: commit before DLQ ack");
System.out.println(" consumer.commitSync();");
System.out.println(" dlqProducer.send(dlqRecord); // If this fails, record lost!");
System.out.println();
System.out.println("CORRECT: DLQ ack first, then commit");
System.out.println(" dlqProducer.send(dlqRecord).get(); // Wait for ack");
System.out.println(" consumer.commitSync(); // Safe: DLQ confirmed");
System.out.println();
System.out.println("WRONG: all exceptions to DLQ (no retry for transient)");
System.out.println(" catch (Exception e) { sendToDlq(record, e); }");
System.out.println();
System.out.println("CORRECT: classify first");
System.out.println(" if (isTransient(e)) retry(record);");
System.out.println(" else sendToDlq(record, e);");
}
}
Expected output:
WRONG: commit before DLQ ack
consumer.commitSync();
dlqProducer.send(dlqRecord); // If this fails, record lost!
CORRECT: DLQ ack first, then commit
dlqProducer.send(dlqRecord).get(); // Wait for ack
consumer.commitSync(); // Safe: DLQ confirmed
WRONG: all exceptions to DLQ (no retry for transient)
catch (Exception e) { sendToDlq(record, e); }
CORRECT: classify first
if (isTransient(e)) retry(record);
else sendToDlq(record, e);
What should a senior engineer know?
Monitor and alert on DLQ depth, because a growing DLQ is an incident signal, not a dumping ground. Add a tiered retry with separate retry topics and increasing backoff before the final DLQ for transient errors. Build a redrive job that replays fixed messages from the DLQ back to the source topic.
Visual Overview
DLQ Processing Flow
====================
┌──────────────────┐
│ consumer.poll() │
└────────┬─────────┘
│
▼
┌──────────────────┐
│ Process record │
└────────┬─────────┘
│
┌────┴────────────────┐
│ │
▼ ▼
┌──────────┐ ┌────────────────┐
│ SUCCESS │ │ FAILURE │
└────┬─────┘ └───────┬────────┘
│ │
│ ┌──────┴──────┐
│ │ │
│ ▼ ▼
│ ┌──────────┐ ┌────────────┐
│ │Transient │ │ Permanent │
│ │(timeout) │ │ (bad data) │
│ └────┬─────┘ └─────┬──────┘
│ │ │
│ ▼ │
│ ┌──────────┐ │
│ │ Retry │ │
│ │ (backoff) │ │
│ └────┬─────┘ │
│ │ │
│ ┌────┴────┐ │
│ │ │ │
│ ▼ ▼ │
│ Success Exhausted │
│ │ │ │
│ │ ┌────┘ │
│ │ │ │
│ ▼ ▼ ▼
│ │ ┌────────────────────┐
│ │ │ Send to DLQ topic │
│ │ │ + failure headers │
│ │ └─────────┬──────────┘
│ │ │
▼ ▼ ▼
┌────────────────────────────────┐
│ consumer.commitSync() │
│ (only after process or DLQ ack)│
└────────────────────────────────┘
Key Takeaway
A dead-letter queue is a separate topic where a consumer sends messages it cannot process, so a single poison message never blocks the partition and is never silently dropped. I classify failures as transient or permanent, retry transient ones with backoff, and route permanent ones or exhausted retries to the DLQ with full failure metadata.
The critical ordering is that I commit the source offset only after the record is either processed successfully or confirmed written to the DLQ. If the DLQ write fails, I do not commit, so the record is retried rather than lost.
In my National Bank of India settlement service, occasional malformed transfers used to block the partition. We added a DLQ with failure headers, classified errors, and committed only after each record was handled. We also alerted on DLQ depth and built a redrive job for fixed messages.
Remember one sentence: a DLQ consumer routes poison messages aside with failure metadata so the partition advances without losing data, and commits only after process or DLQ-write succeeds.
---