10 - Dead Letter Queue and Error Handling in Kafka

A message that will not let itself be processed blocks the whole partition

In a queue like RabbitMQ a message that fails can go back to the queue and end up after the others. In Kafka there is no queue from which to remove a single message: there is an ordered log and an offset that advances. If your consumer cannot process the record at offset 2 and does not move on, records 3, 4 and all the following ones of that partition wait. If instead it moves on, ignoring the error, record 2 is lost. The message that blocks you forever, for example a corrupted JSON or a negative amount, is called a poison pill.

The answer is the dead letter queue (DLQ): a second topic where you put the records you cannot process, with the information needed to understand why, and in the meantime you move on. It looks simple and yet it is easy to get wrong; the most instructive example is the "classic" code that circulates, where a single misplaced commitSync() loses data. Here I reproduce it with the real MockConsumer and the real MockProducer of the Kafka 4.4.0 client (Maven Central, SHA-1 verified), which run the real API without a broker. The parts that require Spring Kafka are instead a template to adapt, read from the Spring Kafka documentation (consulted on 2026-11-12).

The bug that passes the tests and loses the records

The loop found in the examples has this shape: for each record of the batch you try to process it; if it succeeds, commitSync(); if the error is transient, you do not commit and you "retry"; if it is permanent, you send it to the DLQ and commit. The flaw is in the semantics of commitSync() without arguments: it saves the consumer's position, that is the end of everything that poll() returned, not the offset of the record just processed. In a batch of five records, the commit after the first record already saves 5. If the process then dies, the rest of the batch will never be read again. And the "retry" retries nothing: the loop simply moves on to the next record.

The listing below compares the naive version and a correct one on the same batch of five records. Offset 2 is invalid data (a permanent error); offset 3 fails twice and succeeds on the third attempt (a transient error). The correct version retries on the record, sends to the DLQ what does not heal and saves the record's explicit offset plus one only after deciding its fate.

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.serialization.StringSerializer;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.*;

public class Dlq {
    static class Permanent extends RuntimeException { Permanent(String m) { super(m); } }
    static class Transient extends RuntimeException { Transient(String m) { super(m); } }

    static final TopicPartition TP = new TopicPartition("orders", 0);
    static final Map<Long, Integer> tentativiFalliti = new HashMap<>();

    // Offset 2: invalid data, it will never heal. Offset 3: the downstream service only answers from the third attempt.
    static void elabora(ConsumerRecord<String, String> r) {
        if (r.offset() == 2) throw new Permanent("negative amount");
        if (r.offset() == 3 && tentativiFalliti.merge(3L, 1, Integer::sum) <= 2) throw new Transient("downstream service not responding");
    }

    static MockConsumer<String, String> nuovoConsumer() {
        MockConsumer<String, String> c = new MockConsumer<>("earliest");
        c.assign(List.of(TP));
        c.updateBeginningOffsets(Map.of(TP, 0L));
        for (long i = 0; i < 5; i++) c.addRecord(new ConsumerRecord<>("orders", 0, i, "order-" + i, "payload-" + i));
        return c;
    }

    static void versioneIngenua() {
        MockConsumer<String, String> consumer = nuovoConsumer();
        tentativiFalliti.clear();
        List<Long> elaborati = new ArrayList<>();
        for (ConsumerRecord<String, String> r : consumer.poll(Duration.ofMillis(10))) {
            try {
                elabora(r);
                elaborati.add(r.offset());
                consumer.commitSync();                       // saves the POSITION, that is the end of the whole batch
                if (r.offset() == 0) System.out.println("naive: after offset 0 alone the committed offset is "
                    + consumer.committed(Set.of(TP)).get(TP).offset());
            } catch (Transient e) {
                // "retry": no commit, but the loop moves on to the next record
            } catch (Permanent e) {
                consumer.commitSync();
            }
        }
        System.out.println("naive: processed " + elaborati + ", final committed "
            + consumer.committed(Set.of(TP)).get(TP).offset() + " -> offset 3 was never processed nor sent to the DLQ");
    }

    static void versioneCorretta() {
        MockConsumer<String, String> consumer = nuovoConsumer();
        MockProducer<String, String> dlq = new MockProducer<>(true, null, new StringSerializer(), new StringSerializer());
        tentativiFalliti.clear();
        List<Long> elaborati = new ArrayList<>();
        final int MAX_TENTATIVI = 3;
        for (ConsumerRecord<String, String> r : consumer.poll(Duration.ofMillis(10))) {
            Exception ultimo = null;
            int tentativi = 0;
            while (tentativi < MAX_TENTATIVI) {              // retries happen ON the record, before moving on to the next one
                tentativi++;
                try { elabora(r); ultimo = null; break; }
                catch (Permanent e) { ultimo = e; break; }  // no point retrying
                catch (Transient e) { ultimo = e; }        // the backoff wait would go here
            }
            if (ultimo == null) elaborati.add(r.offset());
            else {
                ProducerRecord<String, String> p = new ProducerRecord<>("orders.DLT", r.key(), r.value());
                p.headers().add("dlq-original-topic", r.topic().getBytes(StandardCharsets.UTF_8));
                p.headers().add("dlq-original-offset", String.valueOf(r.offset()).getBytes(StandardCharsets.UTF_8));
                p.headers().add("dlq-error-class", ultimo.getClass().getSimpleName().getBytes(StandardCharsets.UTF_8));
                p.headers().add("dlq-attempts", String.valueOf(tentativi).getBytes(StandardCharsets.UTF_8));
                dlq.send(p);
            }
            consumer.commitSync(Map.of(TP, new OffsetAndMetadata(r.offset() + 1)));   // only after deciding its fate
        }
        System.out.println("correct: processed " + elaborati + ", final committed "
            + consumer.committed(Set.of(TP)).get(TP).offset());
        for (ProducerRecord<String, String> p : dlq.history()) {
            StringBuilder h = new StringBuilder();
            for (Header x : p.headers()) h.append(x.key()).append('=').append(new String(x.value(), StandardCharsets.UTF_8)).append(' ');
            System.out.println("correct: in " + p.topic() + " " + p.key() + " | " + h.toString().trim());
        }
    }

    public static void main(String[] args) {
        versioneIngenua();
        versioneCorretta();
    }
}

// Output:
// naive: after offset 0 alone the committed offset is 5
// naive: processed [0, 1, 4], final committed 5 -> offset 3 was never processed nor sent to the DLQ
// correct: processed [0, 1, 3, 4], final committed 5
// correct: in orders.DLT order-2 | dlq-original-topic=orders dlq-original-offset=2 dlq-error-class=Permanent dlq-attempts=1

In the naive version the first line says it all: after processing offset 0 alone the committed offset is already 5. The final result is [0, 1, 4] processed: offset 2 is not in the DLQ (omitted in the model) and offset 3 has vanished, because its "retry" never happened. In the correct version the processed ones are [0, 1, 3, 4] and offset 2 appears in orders.DLT with the information that is needed: origin topic and offset, error class, number of attempts. The commit goes through commitSync(Map) with offset + 1 after each record. An alternative that costs fewer commits is to save once per batch, but only after deciding the fate of all the records. The MockProducer accepts the DLQ like any producer: in a real cluster you must choose acks=all and check the outcome of the send, because a DLQ that fails silently is worse than no DLQ.

Transient and permanent errors: they are not treated the same way

The distinction that holds everything up is between errors that can heal (a downstream service that does not respond, a timeout) and errors that will never heal (a malformed message, a violated business rule). Retrying a permanent error is wasted time and delays the partition; sending a transient error straight to the DLQ loses the chance to recover. In code the distinction is expressed with two exception types or an enumeration, and it decides the path: the permanent one goes to the DLQ at the first hit, the transient one retries with a limit.

For retries the wait must not be fixed. Exponential backoff with jitter doubles the cap at every attempt and picks a random value under that cap, to prevent all consumers from retrying at the same instant. It is the same principle as the Kafka clients themselves: in the producer documentation, retry.backoff.ms (100) is the initial value and retry.backoff.max.ms (1000) the maximum, with a random jitter added so as not to synchronize the clients. The listing computes the cap and a random wait with a fixed seed, so the output is repeatable.

import java.util.Random;

public class Backoff {
    public static void main(String[] args) {
        long base = 1000, massimo = 30000;
        Random caso = new Random(42);                       // fixed seed: the output is repeatable
        System.out.println("attempt   cap(ms)  wait with full jitter(ms)");
        for (int t = 1; t <= 6; t++) {
            long tetto = Math.min(base * (1L << (t - 1)), massimo);   // 1000, 2000, 4000, ... up to 30000
            long attesa = (long) (caso.nextDouble() * tetto);         // a random value between 0 and the cap
            System.out.printf("%9d  %9d  %26d%n", t, tetto, attesa);
        }
    }
}

// Output:
// attempt   cap(ms)  wait with full jitter(ms)
//         1       1000                         727
//         2       2000                        1366
//         3       4000                        1234
//         4       8000                        2216
//         5      16000                       10648
//         6      30000                       27101

The cap doubles at every attempt (1000, 2000, 4000 ... up to the limit of 30000 ms) and the wait is a number between 0 and that cap: with another seed the same six attempts would have other values. The practical consequence is in the upper part of the table: with a maximum of three attempts a consumer waits at most a few seconds, but during that time the partition does not advance. This is the cost of the blocking retry; if the downstream service can stay down for minutes, waiting in the loop makes the lag of the whole partition grow.

Non-blocking retries: the retry topics

To avoid stopping the partition you move the record to a retry topic and carry on: a dedicated consumer picks it up after the planned delay, and if it fails again it moves it to the next topic or to the DLQ. Spring Kafka offers this since version 2.7 with the @RetryableTopic annotation; the documentation specifies two limits to know: non-blocking retries are not supported with batch listeners and cannot be combined with the container's transactions. The conceptual price is that the re-read record is no longer in its position: the order relative to the other records with the same key is no longer guaranteed.

The basic behavior, with blocking retries and sending to the DLQ, is instead a standard component: the DefaultErrorHandler with a BackOff and a DeadLetterPublishingRecoverer. The default destination topic is <original topic>-dlt, in the same partition as the original record, so the DLQ must have at least as many partitions as the origin topic; the published record keeps the original headers and adds others for the exception (DLT_EXCEPTION_MESSAGE, DLT_EXCEPTION_STACKTRACE and so on).

// Template to adapt (not compiled: it needs Spring Kafka in the project). The real code of the Spring project decides the rest.
@Bean
CommonErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
    // After the failed attempts the record is published to <topic>-dlt (the default name), in the same partition as the original record
    DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
    // FixedBackOff(interval, retries): FixedBackOff(1000L, 3L) = 1 delivery + 3 retries, 1 second apart
    return new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 3L));
}

The line to read carefully is FixedBackOff(1000L, 3L): the second argument is the number of retries, not of attempts. The documentation shows it with an example: FixedBackOff(0L, 2L) has "recover after 3 failures" in the example's comment. With 3L the record is delivered four times in all before ending up in the DLQ. For deserialization errors there is a second component, the ErrorHandlingDeserializer: the documentation explains that a deserialization error happens before poll() returns, so the framework could not handle it; the deserializer delegates to the real one, and if that fails it returns a null value and puts in the DeserializationException header the exception with the raw bytes. In this way the poison pill reaches the error handler, which sends it to the DLQ without blocking the consumer.

If you do not use Spring, two paths are documented for Kafka itself. Kafka Streams has, in the 4.4 documentation, the property errors.dead.letter.queue.topic.name, which makes the default handler build a DLQ record towards the indicated topic (see the article on Kafka Streams). Kafka Connect has errors.deadletterqueue.topic.name, valid for sink connectors (see the article on Connect).

The DLQ is not a trash can

A quarantine topic that nobody looks at is an elegant way of losing data. Three habits prevent it. First: the alarm on its growth. Any message that reaches the DLQ is an event that someone must know about; the technique is that of the lag described in the article on monitoring, applied to the -dlt topic. Second: the headers. A record in the DLQ without origin topic, partition and offset, error class and message, number of attempts is a riddle; the model above writes them all. Third: the retention. The topic must keep the records long enough for a person to inspect them, and how much is "enough" is decided by your reaction time: the default for topics is 7 days (retention.ms 604800000, topic documentation), and with shifts and weekends it may not be enough.

Reprocessing is the other half. After resolving the cause, you reread the DLQ records and send them back to the origin topic, using the header with the original topic. Two warnings: the resent record ends up at the end of the topic, so it loses its order relative to records with the same key that were already processed (if order matters, the consumer must be idempotent or compare versions), and it must be marked with a header to tell it apart from new messages. Consumer idempotency is described, at the pattern level, in the article on idempotency, while the general DLQ pattern in asynchronous systems is in the article of the Event-Driven Architecture series.

What you can do now

You know why commitSync() without arguments saves the position and not the last processed record, and how "retry without commit" loses data. You can write a loop that retries on the record, tells permanent errors from transient ones, sends to the DLQ with the right headers and saves offset + 1 only after the decision. You can compute an exponential backoff with jitter, and you know that with FixedBackOff(1000L, 3L) the attempts are four. You know when it is worth moving the retry to a separate topic and what it costs in terms of order. And you know that a DLQ is useful only if someone sees it grow. To size the topics, retention and replicas of a cluster that also hosts DLQs, move on to the article on Kafka in production. If you come from another point of the series, the links are: advanced producers and consumers.