Kafka Dead Letter Topics: Handling Failed Messages

A Dead Letter Topic (DLT) is a dedicated Kafka topic where messages that cannot be processed successfully are routed after exhausting all retry attempts. Instead of blocking the consumer or silently dropping records, the DLT captures failed messages along with diagnostic headers so they can be inspected, monitored, and replayed later.

Share

Contents

What Are Dead Letter Topics

In any event-driven system, some messages will inevitably fail to process. A consumer might encounter malformed JSON, a missing required field, a downstream service outage, or a schema mismatch. Without a strategy for handling these failures, the consumer either blocks indefinitely on the bad record or skips it with no trace.

A Dead Letter Topic (DLT) solves this problem by providing a safe destination for messages that have exhausted all retry attempts. The original message is published to the DLT along with metadata headers that capture the failure context: the original topic, partition, offset, exception class, and stack trace.

Common reasons a message ends up in a DLT:

  • Poison pill records: messages that can never be deserialized or processed regardless of how many times they are retried (e.g., corrupted bytes, incompatible schema changes).
  • Validation failures: the payload is syntactically valid but violates business rules (missing required fields, out-of-range values).
  • Transient downstream failures: a database or external API is temporarily unavailable and retries are exhausted.
  • Deserialization errors: the consumer cannot convert the raw bytes into the expected object type.
A DLT does not replace retries: it complements them. Retries handle transient failures; the DLT captures messages that fail permanently or outlast the retry budget.

DLT Naming Conventions

Spring Kafka follows a standard naming convention where the DLT name is derived from the original topic name by appending a .DLT suffix. For example, if your source topic is orders, the dead letter topic will be orders.DLT.

Original TopicDead Letter Topic
ordersorders.DLT
paymentspayments.DLT
user-eventsuser-events.DLT

When a message is forwarded to the DLT, Spring Kafka attaches several headers to preserve context about the original failure:

  • kafka_dlt-original-topic: the topic where the message was originally consumed.
  • kafka_dlt-original-partition: the partition number in the source topic.
  • kafka_dlt-original-offset: the offset of the failed record.
  • kafka_dlt-original-timestamp: the original record timestamp.
  • kafka_dlt-exception-fqcn: the fully qualified class name of the exception.
  • kafka_dlt-exception-message: the exception message text.
  • kafka_dlt-exception-stacktrace: the full stack trace of the failure.
Make sure the DLT topic is pre-created in production environments with the appropriate number of partitions and replication factor. Relying on auto-topic-creation can lead to topics with undesirable defaults.

Spring Kafka DefaultErrorHandler

The DefaultErrorHandler is the primary error handling mechanism in Spring Kafka. It replaces the older SeekToCurrentErrorHandler and provides built-in support for back-off retries and dead letter publishing through DeadLetterPublishingRecoverer.

Java
@Configuration
public class KafkaConsumerConfig {

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> pf) {
        return new KafkaTemplate<>(pf);
    }

    @Bean
    public DeadLetterPublishingRecoverer deadLetterRecoverer(KafkaTemplate<String, String> template) {
        // Routes failed records to {originalTopic}.DLT
        return new DeadLetterPublishingRecoverer(template,
            (record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition()));
    }

    @Bean
    public DefaultErrorHandler errorHandler(DeadLetterPublishingRecoverer recoverer) {
        // Retry 3 times with 1-second back-off, then send to DLT
        BackOff backOff = new FixedBackOff(1000L, 3);
        DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, backOff);
        return handler;
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
            ConsumerFactory<String, String> cf, DefaultErrorHandler errorHandler) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(cf);
        factory.setCommonErrorHandler(errorHandler);
        return factory;
    }
}

With this configuration, when a listener throws an exception, the DefaultErrorHandler retries the record up to 3 times with a 1-second delay between attempts. If the record still fails after all retries, the DeadLetterPublishingRecoverer publishes it to the DLT with full diagnostic headers.

You can also use an exponential back-off strategy for retries:

Java
@Bean
public DefaultErrorHandler errorHandler(DeadLetterPublishingRecoverer recoverer) {
    // Exponential back-off: initial 1s, multiplier 2, max 10s, max retries 5
    ExponentialBackOff backOff = new ExponentialBackOff(1000L, 2.0);
    backOff.setMaxElapsedTime(30000L); // give up after 30 seconds total
    return new DefaultErrorHandler(recoverer, backOff);
}

Retry Topics

Spring Kafka provides a powerful non-blocking retry mechanism through the @RetryableTopic annotation. Instead of blocking the consumer thread during retries, failed messages are published to intermediate retry topics before finally landing in the DLT.

The retry chain creates a series of topics: orders-retry-0, orders-retry-1, orders-retry-2, and finally orders-dlt. Each retry topic has its own consumer that waits for the configured back-off period before attempting reprocessing.

Java
@Component
public class OrderConsumer {

    @RetryableTopic(
        attempts = "4",
        backoff = @Backoff(delay = 2000, multiplier = 2, maxDelay = 16000),
        autoCreateTopics = "true",
        topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE,
        dltStrategy = DltStrategy.FAIL_ON_ERROR
    )
    @KafkaListener(topics = "orders", groupId = "order-processor")
    public void consume(ConsumerRecord<String, String> record) {
        Order order = parseOrder(record.value());
        orderService.process(order); // may throw on failure
    }

    @DltHandler
    public void handleDlt(ConsumerRecord<String, String> record) {
        log.error("Message exhausted all retries. Topic: {}, Key: {}, Value: {}",
            record.topic(), record.key(), record.value());
        alertService.notifyDltMessage(record);
    }
}

The retry flow with the above configuration proceeds as follows:

StepTopicDelay
1st attemptordersImmediate
2nd attemptorders-retry-02 seconds
3rd attemptorders-retry-14 seconds
4th attemptorders-retry-28 seconds
Dead letterorders-dltN/A
Non-blocking retries through @RetryableTopic allow the consumer to continue processing other messages while failed records wait in retry topics. This is much more efficient than blocking retries for transient failures.

Custom Dead Letter Processing

Beyond the default @DltHandler, you can set up a dedicated consumer that reads from the DLT and applies custom processing logic, logging to an external system, sending alerts, or attempting reprocessing with modified parameters.

Java
@Component
public class DltConsumer {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private final AlertService alertService;

    public DltConsumer(KafkaTemplate<String, String> kafkaTemplate, AlertService alertService) {
        this.kafkaTemplate = kafkaTemplate;
        this.alertService = alertService;
    }

    @KafkaListener(topics = "orders.DLT", groupId = "dlt-processor")
    public void processDlt(ConsumerRecord<String, String> record) {
        // Extract original failure context from headers
        String originalTopic = headerValue(record, KafkaHeaders.DLT_ORIGINAL_TOPIC);
        String exceptionClass = headerValue(record, KafkaHeaders.DLT_EXCEPTION_FQCN);
        String exceptionMessage = headerValue(record, KafkaHeaders.DLT_EXCEPTION_MESSAGE);

        log.error("DLT record — original topic: {}, exception: {} - {}",
            originalTopic, exceptionClass, exceptionMessage);

        // Persist to a dead letter database for manual review
        deadLetterRepository.save(DeadLetterRecord.builder()
            .originalTopic(originalTopic)
            .key(record.key())
            .value(record.value())
            .exceptionClass(exceptionClass)
            .exceptionMessage(exceptionMessage)
            .receivedAt(Instant.now())
            .build());

        // Send alert for critical topics
        if ("payments".equals(originalTopic)) {
            alertService.sendPagerDutyAlert("Payment DLT message: " + record.key());
        }
    }

    private String headerValue(ConsumerRecord<String, String> record, String headerKey) {
        Header header = record.headers().lastHeader(headerKey);
        return header != null ? new String(header.value(), StandardCharsets.UTF_8) : "unknown";
    }
}

DLT Headers

When Spring Kafka publishes a record to the DLT, it attaches a rich set of headers that capture the full failure context. These headers are essential for diagnosing issues and building automated reprocessing logic.

HeaderConstantDescription
kafka_dlt-original-topic KafkaHeaders.DLT_ORIGINAL_TOPIC Source topic name where the message was originally consumed
kafka_dlt-original-partition KafkaHeaders.DLT_ORIGINAL_PARTITION Partition number in the source topic
kafka_dlt-original-offset KafkaHeaders.DLT_ORIGINAL_OFFSET Offset of the failed record in the source partition
kafka_dlt-original-timestamp KafkaHeaders.DLT_ORIGINAL_TIMESTAMP Timestamp of the original record
kafka_dlt-original-timestamp-type KafkaHeaders.DLT_ORIGINAL_TIMESTAMP_TYPE Timestamp type (CreateTime or LogAppendTime)
kafka_dlt-exception-fqcn KafkaHeaders.DLT_EXCEPTION_FQCN Fully qualified class name of the exception
kafka_dlt-exception-message KafkaHeaders.DLT_EXCEPTION_MESSAGE The exception message text
kafka_dlt-exception-stacktrace KafkaHeaders.DLT_EXCEPTION_STACKTRACE Full stack trace of the causing exception
Java
// Reading all DLT headers from a record
private void logDltHeaders(ConsumerRecord<String, String> record) {
    for (Header header : record.headers()) {
        if (header.key().startsWith("kafka_dlt")) {
            log.info("DLT Header — {}: {}",
                header.key(),
                new String(header.value(), StandardCharsets.UTF_8));
        }
    }
}

Non-Retryable Exceptions

Not all exceptions warrant retries. A deserialization error or a validation failure will produce the same result on every attempt, retrying is wasteful. Spring Kafka lets you classify certain exceptions as non-retryable so that they are sent directly to the DLT without any retry attempts.

Java
@Bean
public DefaultErrorHandler errorHandler(DeadLetterPublishingRecoverer recoverer) {
    BackOff backOff = new FixedBackOff(1000L, 3);
    DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, backOff);

    // These exceptions skip retries and go straight to DLT
    handler.addNotRetryableExceptions(
        DeserializationException.class,
        MessageConversionException.class,
        ConversionException.class,
        MethodArgumentNotValidException.class,
        NullPointerException.class
    );

    return handler;
}

For @RetryableTopic, you can specify non-retryable exceptions directly in the annotation:

Java
@RetryableTopic(
    attempts = "4",
    backoff = @Backoff(delay = 2000, multiplier = 2),
    exclude = {
        DeserializationException.class,
        IllegalArgumentException.class
    }
)
@KafkaListener(topics = "orders", groupId = "order-processor")
public void consume(ConsumerRecord<String, String> record) {
    orderService.process(record.value());
}

You can also mark exceptions as non-retryable programmatically from within your listener:

Java
@KafkaListener(topics = "orders", groupId = "order-processor")
public void consume(ConsumerRecord<String, String> record) {
    try {
        Order order = objectMapper.readValue(record.value(), Order.class);
        if (order.getAmount() == null) {
            // This will never succeed on retry — send straight to DLT
            throw new NotRetryableException("Order amount is null for key: " + record.key());
        }
        orderService.process(order);
    } catch (JsonProcessingException e) {
        throw new NotRetryableException("Invalid JSON payload", e);
    }
}
Be deliberate about which exceptions you mark as non-retryable. A NullPointerException caused by a missing field in the payload is genuinely non-retryable, but one caused by a race condition in your code might succeed on retry.

Monitoring Dead Letter Topics

A DLT that silently accumulates messages without anyone noticing is as bad as having no DLT at all. Monitoring and alerting on DLT activity is essential for maintaining system health.

  • Consumer lag on DLT topics, if a DLT consumer is not keeping up, messages pile up and you lose visibility into failures.
  • DLT message rate: a sudden spike in DLT messages often indicates a deployment issue (bad schema, broken downstream service).
  • Alert on any DLT message for critical topics: payment or order topics should trigger immediate alerts when a message reaches the DLT.
  • Track DLT age: messages sitting in the DLT for days may indicate that no one is reviewing or reprocessing them.
Java
@Component
public class DltMetricsCollector {

    private final MeterRegistry meterRegistry;
    private final Counter dltMessageCounter;

    public DltMetricsCollector(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
        this.dltMessageCounter = Counter.builder("kafka.dlt.messages.received")
            .description("Number of messages received in DLT topics")
            .register(meterRegistry);
    }

    @KafkaListener(topicPattern = ".*\\.DLT", groupId = "dlt-monitor")
    public void monitorDlt(ConsumerRecord<String, String> record) {
        String originalTopic = headerValue(record, KafkaHeaders.DLT_ORIGINAL_TOPIC);

        // Increment counter with topic tag for per-topic metrics
        dltMessageCounter.increment();
        meterRegistry.counter("kafka.dlt.messages.by.topic",
            "original_topic", originalTopic).increment();

        // Gauge for time-in-DLT tracking
        meterRegistry.timer("kafka.dlt.message.age",
            "original_topic", originalTopic)
            .record(Duration.between(
                Instant.ofEpochMilli(record.timestamp()), Instant.now()));
    }

    private String headerValue(ConsumerRecord<String, String> record, String key) {
        Header header = record.headers().lastHeader(key);
        return header != null ? new String(header.value(), StandardCharsets.UTF_8) : "unknown";
    }
}
Use a wildcard topic pattern like .*\.DLT to monitor all DLT topics from a single consumer. Combine this with Prometheus and Grafana dashboards for real-time visibility into failure rates across all topics.

Reprocessing Strategy

Once a root cause is fixed: a bug is patched, a downstream service is restored, or a schema is corrected: you need a strategy for replaying DLT messages back to the original topic.

The simplest approach is a dedicated reprocessing service that reads from the DLT, optionally transforms the message, and publishes it back to the source topic.

Java
@Service
public class DltReprocessingService {

    private final KafkaTemplate<String, String> kafkaTemplate;

    public DltReprocessingService(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    /**
     * Replay a single DLT record back to its original topic.
     * Call this from an admin API endpoint or a scheduled job.
     */
    public void replayRecord(ConsumerRecord<String, String> dltRecord) {
        String originalTopic = headerValue(dltRecord, KafkaHeaders.DLT_ORIGINAL_TOPIC);
        if (originalTopic == null) {
            log.error("Cannot replay — missing original topic header for key: {}", dltRecord.key());
            return;
        }

        // Build a new ProducerRecord with the original key and value
        ProducerRecord<String, String> replayRecord =
            new ProducerRecord<>(originalTopic, dltRecord.key(), dltRecord.value());

        // Add a header to track that this is a replayed message
        replayRecord.headers().add("x-replay-source", "dlt-reprocessor".getBytes(StandardCharsets.UTF_8));
        replayRecord.headers().add("x-replay-timestamp",
            Instant.now().toString().getBytes(StandardCharsets.UTF_8));

        kafkaTemplate.send(replayRecord).whenComplete((result, ex) -> {
            if (ex != null) {
                log.error("Failed to replay DLT record to {}: {}", originalTopic, ex.getMessage());
            } else {
                log.info("Replayed DLT record to {} partition {} offset {}",
                    result.getRecordMetadata().topic(),
                    result.getRecordMetadata().partition(),
                    result.getRecordMetadata().offset());
            }
        });
    }

    private String headerValue(ConsumerRecord<String, String> record, String key) {
        Header header = record.headers().lastHeader(key);
        return header != null ? new String(header.value(), StandardCharsets.UTF_8) : null;
    }
}

Expose a REST endpoint to trigger reprocessing on demand:

Java
@RestController
@RequestMapping("/api/dlt")
public class DltReprocessController {

    private final Consumer<String, String> consumer;
    private final DltReprocessingService reprocessingService;

    @PostMapping("/replay/{topic}")
    public ResponseEntity<String> replayDlt(
            @PathVariable String topic,
            @RequestParam(defaultValue = "100") int maxRecords) {

        String dltTopic = topic + ".DLT";
        TopicPartition tp = new TopicPartition(dltTopic, 0);
        consumer.assign(List.of(tp));
        consumer.seekToBeginning(List.of(tp));

        int replayed = 0;
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(5));
        for (ConsumerRecord<String, String> record : records) {
            if (replayed >= maxRecords) break;
            reprocessingService.replayRecord(record);
            replayed++;
        }

        return ResponseEntity.ok("Replayed " + replayed + " records from " + dltTopic);
    }
}
  • Add replay headers: tag replayed messages with a header so consumers can distinguish them from fresh records and avoid infinite DLT loops.
  • Set a replay limit: track how many times a message has been replayed and stop after a threshold to prevent endless cycles.
  • Replay in batches: if the DLT has thousands of records, replay in controlled batches to avoid overwhelming the target topic consumers.
  • Validate before replay: if the root cause was a data issue, consider filtering or transforming DLT records before replaying them.
Always test your reprocessing logic in a staging environment first. Replaying a large volume of DLT messages into a production topic without throttling can cause cascading failures downstream.

Related articles

All Kafka articles →
Kafka - Topics and PartitionsKafka Topics and Partitions: replication factor, leader/follower, log segments, retention, compaction.Apache Kafka - IntroductionApache Kafka Introduction: architecture, producers, consumers, brokers, topics, partitions and use cases.Kafka Connect: Custom Source & Sink ConnectorsKafka Connect custom connectors: build a custom SourceConnector to ingest data into Kafka and a SinkConnector to write Kafka data to an external system.Kafka Monitoring: Metrics & Consumer LagKafka Monitoring: track broker health, consumer lag, producer metrics, and JMX-based observability.