10 - Dead Letter Queue e Error Handling in Kafka

Un messaggio che non si lascia elaborare blocca tutta la partizione

In una coda come RabbitMQ un messaggio che fallisce può tornare in coda e finire dopo gli altri. In Kafka non esiste una coda dalla quale togliere un singolo messaggio: c'è un log ordinato e un offset che avanza. Se il tuo consumatore non riesce a elaborare il record all'offset 2 e non va avanti, i record 3, 4 e tutti i successivi di quella partizione aspettano. Se invece va avanti ignorando l'errore, il record 2 è perso. Il messaggio che ti blocca per sempre, per esempio un JSON corrotto o un importo negativo, si chiama poison pill.

La risposta è la dead letter queue (DLQ): un secondo topic dove metti i record che non riesci a elaborare, con le informazioni per capire perché, e intanto vai avanti. Sembra semplice e invece è facile sbagliarlo; l'esempio più istruttivo è il codice «classico» che circola, in cui un solo commitSync() mal piazzato fa perdere dati. Qui lo riproduco con il vero MockConsumer e il vero MockProducer del client Kafka 4.4.0 (Maven Central, SHA-1 verificato), che eseguono l'API reale senza broker. Le parti che richiedono Spring Kafka sono invece uno schema da adattare, letto dalla documentazione di Spring Kafka (consultata il 2026-11-12).

Il bug che passa i test e perde i record

Il ciclo che si trova negli esempi ha questa forma: per ogni record del batch si prova a elaborare; se riesce, commitSync(); se l'errore è transitorio, non si fa commit e si «ritenta»; se è permanente, si manda in DLQ e si fa commit. Il difetto è nella semantica di commitSync() senza argomenti: salva la posizione del consumatore, cioè la fine di tutto ciò che poll() ha restituito, non l'offset del record appena elaborato. In un batch da cinque record, il commit dopo il primo record salva già 5. Se poi il processo cade, il resto del batch non verrà mai riletto. E il «ritenta» non ritenta nulla: il ciclo passa semplicemente al record successivo.

Il listato sotto mette a confronto la versione ingenua e una corretta sullo stesso batch di cinque record. L'offset 2 è un dato non valido (errore permanente); l'offset 3 fallisce due volte e riesce al terzo tentativo (errore transitorio). La versione corretta fa ritentare sul record, manda in DLQ ciò che non guarisce e salva l'offset esplicito del record più uno solo dopo aver deciso il suo destino.

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 Permanente extends RuntimeException { Permanente(String m) { super(m); } }
    static class Transitorio extends RuntimeException { Transitorio(String m) { super(m); } }

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

    // Offset 2: dato non valido, non guarirà mai. Offset 3: il servizio a valle risponde solo dal terzo tentativo.
    static void elabora(ConsumerRecord<String, String> r) {
        if (r.offset() == 2) throw new Permanente("importo negativo");
        if (r.offset() == 3 && tentativiFalliti.merge(3L, 1, Integer::sum) <= 2) throw new Transitorio("servizio a valle non risponde");
    }

    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<>("ordini", 0, i, "ordine-" + 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();                       // salva la POSIZIONE, cioè la fine dell'intero batch
                if (r.offset() == 0) System.out.println("ingenua: dopo il solo offset 0 il committed è "
                    + consumer.committed(Set.of(TP)).get(TP).offset());
            } catch (Transitorio e) {
                // "ritenta": nessun commit, ma il ciclo prosegue col record successivo
            } catch (Permanente e) {
                consumer.commitSync();
            }
        }
        System.out.println("ingenua: elaborati " + elaborati + ", committed finale "
            + consumer.committed(Set.of(TP)).get(TP).offset() + " -> l'offset 3 non è mai stato elaborato né mandato in 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) {              // i retry avvengono SUL record, prima di passare al successivo
                tentativi++;
                try { elabora(r); ultimo = null; break; }
                catch (Permanente e) { ultimo = e; break; }  // inutile riprovare
                catch (Transitorio e) { ultimo = e; }        // qui andrebbe l'attesa con backoff
            }
            if (ultimo == null) elaborati.add(r.offset());
            else {
                ProducerRecord<String, String> p = new ProducerRecord<>("ordini.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)));   // solo dopo aver deciso il destino
        }
        System.out.println("corretta: elaborati " + elaborati + ", committed finale "
            + 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("corretta: in " + p.topic() + " " + p.key() + " | " + h.toString().trim());
        }
    }

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

// Output:
// ingenua: dopo il solo offset 0 il committed è 5
// ingenua: elaborati [0, 1, 4], committed finale 5 -> l'offset 3 non è mai stato elaborato né mandato in DLQ
// corretta: elaborati [0, 1, 3, 4], committed finale 5
// corretta: in ordini.DLT ordine-2 | dlq-original-topic=ordini dlq-original-offset=2 dlq-error-class=Permanente dlq-attempts=1

Nella versione ingenua la prima riga dice tutto: dopo aver elaborato il solo offset 0 il committed vale già 5. Il risultato finale è [0, 1, 4] elaborati: l'offset 2 non è in DLQ (nel modello omessa) e l'offset 3 è sparito, perché il suo «ritentativo» non è mai avvenuto. Nella versione corretta gli elaborati sono [0, 1, 3, 4] e l'offset 2 compare in ordini.DLT con le informazioni che servono: topic e offset di origine, classe dell'errore, numero di tentativi. Il commit passa da commitSync(Map) con offset + 1 dopo ogni record. Un'alternativa che costa meno commit è salvare una volta per batch, ma solo dopo aver deciso il destino di tutti i record. Il MockProducer accetta la DLQ come qualsiasi producer: in un cluster vero devi scegliere acks=all e controllare l'esito dell'invio, perché una DLQ che fallisce in silenzio è peggio di nessuna DLQ.

Errori transitori e permanenti: non si trattano allo stesso modo

La distinzione che regge tutto è tra errori che possono guarire (un servizio a valle che non risponde, un timeout) e errori che non guariranno mai (un messaggio malformato, una regola di business violata). Ritentare un errore permanente è tempo perso e ritarda la partizione; mandare subito in DLQ un errore transitorio perde la possibilità di recuperare. Nel codice la distinzione si esprime con due tipi di eccezione o un'enumerazione, e decide la strada: il permanente va in DLQ al primo colpo, il transitorio ritenta con un limite.

Per i ritentativi l'attesa non deve essere fissa. Il backoff esponenziale con jitter raddoppia il tetto a ogni tentativo e sceglie un valore casuale sotto quel tetto, per evitare che tutti i consumatori ritentino nello stesso istante. È lo stesso principio dei client Kafka stessi: nella documentazione del producer, retry.backoff.ms (100) è il valore iniziale e retry.backoff.max.ms (1000) il massimo, con un jitter casuale aggiunto per non sincronizzare i client. Il listato calcola il tetto e un'attesa casuale con un seme fisso, così l'uscita è ripetibile.

import java.util.Random;

public class Backoff {
    public static void main(String[] args) {
        long base = 1000, massimo = 30000;
        Random caso = new Random(42);                       // seme fisso: l'uscita è ripetibile
        System.out.println("tentativo  tetto(ms)  attesa con full jitter(ms)");
        for (int t = 1; t <= 6; t++) {
            long tetto = Math.min(base * (1L << (t - 1)), massimo);   // 1000, 2000, 4000, ... fino a 30000
            long attesa = (long) (caso.nextDouble() * tetto);         // un valore casuale tra 0 e il tetto
            System.out.printf("%9d  %9d  %26d%n", t, tetto, attesa);
        }
    }
}

// Output:
// tentativo  tetto(ms)  attesa con full jitter(ms)
//         1       1000                         727
//         2       2000                        1366
//         3       4000                        1234
//         4       8000                        2216
//         5      16000                       10648
//         6      30000                       27101

Il tetto raddoppia a ogni tentativo (1000, 2000, 4000 ... fino al limite di 30000 ms) e l'attesa è un numero tra 0 e quel tetto: con un altro seme gli stessi sei tentativi avrebbero altri valori. La conseguenza pratica è nella parte alta della tabella: con un massimo di tre tentativi un consumatore attende al più qualche secondo, ma durante quel tempo la partizione non avanza. È il costo del retry bloccante; se il servizio a valle può restare giù per minuti, aspettare nel ciclo fa crescere il lag di tutta la partizione.

Retry non bloccanti: i topic di ritentativo

Per non fermare la partizione si sposta il record in un topic di ritentativo e si va avanti: un consumatore dedicato lo riprende dopo il ritardo previsto, e se fallisce ancora lo sposta nel topic successivo o nella DLQ. Spring Kafka lo offre dalla versione 2.7 con l'annotazione @RetryableTopic; la documentazione precisa due limiti da conoscere: i ritentativi non bloccanti non sono supportati con i listener a batch e non si combinano con le transazioni del container. Il prezzo concettuale è che il record riletto non è più nella sua posizione: l'ordine rispetto agli altri record della stessa chiave non è più garantito.

Il comportamento di base, con ritentativi bloccanti e invio alla DLQ, è invece un componente standard: il DefaultErrorHandler con un BackOff e un DeadLetterPublishingRecoverer. Il valore predefinito del topic di destinazione è <topic originale>-dlt, nella stessa partizione del record originale, quindi la DLQ deve avere almeno altrettante partizioni del topic di origine; il record pubblicato conserva gli header originali e ne aggiunge altri per l'eccezione (DLT_EXCEPTION_MESSAGE, DLT_EXCEPTION_STACKTRACE e così via).

// Schema da adattare (non compilato: richiede Spring Kafka nel progetto). Il codice reale del progetto Spring decide il resto.
@Bean
CommonErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
    // Dopo i tentativi falliti il record viene pubblicato su <topic>-dlt (il nome predefinito), nella stessa partizione del record originale
    DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
    // FixedBackOff(intervallo, ritentativi): FixedBackOff(1000L, 3L) = 1 consegna + 3 ritentativi, a 1 secondo l'uno dall'altro
    return new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 3L));
}

La riga da leggere con attenzione è FixedBackOff(1000L, 3L): il secondo argomento è il numero di ritentativi, non di tentativi. La documentazione lo dimostra con un esempio: FixedBackOff(0L, 2L) ha nel commento dell'esempio «recover after 3 failures». Con 3L il record viene consegnato quattro volte in tutto prima di finire in DLQ. Per gli errori di deserializzazione c'è un secondo componente, l'ErrorHandlingDeserializer: la documentazione spiega che un errore di deserializzazione avviene prima che poll() restituisca, quindi il framework non potrebbe gestirlo; il deserializer delega a quello vero, e se questo fallisce restituisce un valore nullo e mette nell'header DeserializationException l'eccezione con i byte grezzi. Così il poison pill arriva fino al gestore d'errore, che lo manda in DLQ senza bloccare il consumatore.

Se non usi Spring, due strade sono documentate per Kafka stesso. Kafka Streams ha, nella documentazione 4.4, la proprietà errors.dead.letter.queue.topic.name, che fa costruire al gestore predefinito un record di DLQ verso il topic indicato (vedi l'articolo su Kafka Streams). Kafka Connect ha errors.deadletterqueue.topic.name, valida per i connector sink (vedi l'articolo su Connect).

La DLQ non è un cestino

Un topic di quarantena che nessuno guarda è un modo elegante di perdere dati. Tre abitudini lo evitano. Primo: l'allarme sulla sua crescita. Qualunque messaggio che arriva in DLQ è un evento che qualcuno deve conoscere; la tecnica è quella del lag descritto nell'articolo sul monitoraggio, applicata al topic -dlt. Secondo: gli header. Un record in DLQ senza topic, partizione e offset di origine, classe e messaggio dell'errore, numero di tentativi è un enigma; il modello qui sopra li scrive tutti. Terzo: la retention. Il topic deve conservare i record abbastanza a lungo perché una persona li ispezioni, e quanto sia «abbastanza» lo decide il tuo tempo di reazione: il default dei topic è 7 giorni (retention.ms 604800000, documentazione dei topic), e con turni e weekend può non bastare.

Il reprocessing è l'altra metà. Dopo aver risolto la causa, si rileggono i record della DLQ e si rimandano al topic di origine, usando l'header con il topic originale. Due avvertenze: il record rimandato finisce in fondo al topic, quindi perde l'ordine rispetto ai record con la stessa chiave già elaborati (se l'ordine conta, il consumatore deve essere idempotente o comparare versioni), e va marcato con un header per distinguerlo dai messaggi nuovi. L'idempotenza del consumatore è descritta, a livello di pattern, nell'articolo sull'idempotenza, mentre il pattern generale della DLQ nei sistemi asincroni è nell'articolo della serie Event-Driven Architecture.

Che cosa sai fare adesso

Sai perché commitSync() senza argomenti salva la posizione e non l'ultimo record elaborato, e come il «ritenta senza commit» perde dati. Sai scrivere un ciclo che ritenta sul record, distingue errori permanenti da transitori, manda in DLQ con gli header giusti e salva offset + 1 solo dopo la decisione. Sai calcolare un backoff esponenziale con jitter, e che con FixedBackOff(1000L, 3L) i tentativi sono quattro. Sai quando conviene spostare il ritentativo su un topic separato e che costo ha in termini di ordine. E sai che una DLQ è utile solo se qualcuno ne vede la crescita. Per dimensionare topic, retention e repliche di un cluster che ospita anche le DLQ, passa all'articolo su Kafka in produzione. Se arrivi da un altro punto della serie, i collegamenti sono: i producer e consumer avanzati.