03 - Kafka Producer e Consumer Avanzati: Acks, Idempotenza e Retry

Un ordine arriva una volta, due volte o mai? Dipende da sei proprietà

Il tuo servizio di checkout invia un ordine a Kafka. Il broker lo scrive, la risposta si perde per strada, il client scade e riprova. Risultato possibile: nel topic ci sono due ordini identici e il cliente viene addebitato due volte. Oppure il contrario: il leader conferma e muore prima che un follower copi il record, e l'ordine sparisce. Sono due guasti diversi e vengono evitati da due meccanismi diversi, acks con min.insync.replicas per la durabilità e l'idempotenza per i duplicati. Sul lato della lettura il rischio è speculare: un offset salvato troppo presto fa saltare record, uno salvato troppo tardi li fa rielaborare.

La notizia utile è che la documentazione di Kafka 4.4 (consultata il 2026-11-12) imposta già per default la combinazione sicura. Il listato qui sotto è codice reale: legge la configurazione del producer del client Java 4.4.0 (kafka-clients-4.4.0.jar da Maven Central, SHA-1 verificato) senza collegarsi a nessun broker, e stampa i valori che avresti se non scrivessi nulla.

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.ConfigException;
import java.util.*;

public class ConfigProducer {
    static Map<String, Object> base() {
        Map<String, Object> p = new HashMap<>();
        p.put("bootstrap.servers", "localhost:9092");   // non viene contattato: si legge solo la configurazione
        p.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        p.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        return p;
    }

    public static void main(String[] args) {
        ProducerConfig predefiniti = new ProducerConfig(base());
        for (String k : List.of("acks", "enable.idempotence", "retries", "delivery.timeout.ms",
                "request.timeout.ms", "linger.ms", "batch.size", "max.in.flight.requests.per.connection"))
            System.out.println(k + " = " + predefiniti.values().get(k));

        Map<String, Object> conflitto = base();
        conflitto.put("enable.idempotence", true);      // richiesta esplicita...
        conflitto.put("acks", "1");                     // ...incompatibile con acks=1
        try {
            new ProducerConfig(conflitto);
        } catch (ConfigException e) {
            System.out.println("ConfigException: " + e.getMessage());
        }
    }
}

// Output:
// acks = -1
// enable.idempotence = true
// retries = 2147483647
// delivery.timeout.ms = 120000
// request.timeout.ms = 30000
// linger.ms = 5
// batch.size = 16384
// max.in.flight.requests.per.connection = 5
// ConfigException: Must set acks to all in order to use the idempotent producer. Otherwise we cannot guarantee idempotence.

Le prime otto righe sono i default. acks compare come -1, che è la forma numerica di all; enable.idempotence è true; retries vale 2147483647 (praticamente «riprova finché non scade il tempo»), perché il limite vero è delivery.timeout.ms, 120000 ms; max.in.flight.requests.per.connection è 5, il massimo compatibile con l'idempotenza. Il linger.ms a 5 è una novità: le note di upgrade dicono che il default è passato da 0 a 5 in Kafka 4.0, perché lotti più grandi in genere non peggiorano la latenza. L'ultima riga è la parte istruttiva: se chiedi esplicitamente l'idempotenza e imposti acks=1, il client rifiuta la configurazione con un ConfigException. Se invece non nomini l'idempotenza e imposti acks=1, la documentazione dice che l'idempotenza viene semplicemente disattivata invece di fallire: è il modo più comune di perdere una garanzia credendo di averla.

Le tre garanzie di consegna, e che cosa costa ciascuna

La sezione «Message Delivery Semantics» della documentazione distingue at most once (un messaggio può perdersi ma non si duplica), at least once (non si perde ma può duplicarsi) e exactly once (una sola volta). Il predefinito di Kafka è at-least-once; per ottenere at-most-once devi disattivare i retry del producer e salvare l'offset prima di elaborare il batch. Exactly-once, per un flusso che legge da Kafka e scrive su Kafka, richiede le transazioni, di cui si occupa l'articolo successivo. Verso un sistema esterno (un database, un'API) la documentazione è netta: serve la cooperazione di quel sistema, per esempio una chiave di deduplicazione.

La scelta dipende dal danno di un errore. Un contatore di clic può perdere qualche evento (at-most-once è tollerabile); un addebito non può né perdersi né duplicarsi (serve la deduplicazione, e l'idempotenza del producer è solo il primo anello). Ricorda che l'idempotenza del producer protegge dai duplicati nel log; non dice nulla su ciò che il consumatore fa dopo aver letto.

acks e min.insync.replicas: quando un record è al sicuro

acks stabilisce quante conferme il producer attende. Con acks=0 non attende nulla: il record è considerato inviato quando entra nel buffer del socket, l'offset restituito è sempre -1 e i retry non hanno effetto, perché il client non sa se qualcosa è andato storto. Con acks=1 attende il leader, che può confermare e cadere prima che i follower copino. Con acks=all attende tutte le repliche dell'ISR, ma «tutte» può voler dire una sola, il leader, se l'ISR si è ridotta. Per questo serve il secondo parametro, min.insync.replicas (default 1 nella documentazione dei topic): con acks=all, se l'ISR ha meno membri di quel minimo, il producer riceve un errore (NotEnoughReplicas o NotEnoughReplicasAfterAppend). Con replication factor 3 e minimo 2, un record confermato esiste su almeno due broker (l'ISR è spiegata nell'articolo sui fondamenti).

Il prezzo è la disponibilità: se due broker su tre cadono, le scritture si fermano invece di accettare un record con una sola copia. È una scelta, non un difetto, e va fatta consapevolmente per i topic che contengono dati che non puoi rigenerare. Quanta latenza in più costi non ha una cifra valida in assoluto: dipende da rete e dischi del tuo cluster, e va misurata con un tuo test.

Retry e duplicati: il modello

Il caso che trasforma i retry in duplicati è il più subdolo: il broker scrive e conferma, ma la conferma non torna al client. Dal punto di vista del producer c'è un record senza risposta e non sa se sia stato scritto. Il modello seguente mette a confronto tre produttori davanti a due guasti: la richiesta che si perde prima di arrivare e la conferma che si perde al ritorno. Il broker del modello scrive e, se il producer è idempotente, ricorda l'ultima sequenza vista per ogni producer id e ignora quelle già viste.

import java.util.*;

public class Consegna {
    enum Guasto { RICHIESTA_PERSA, ACK_PERSO }

    static class Broker {
        final List<String> log = new ArrayList<>();
        final Map<Long, Integer> ultimaSequenza = new HashMap<>();   // per ogni producer id

        // Restituisce true se il broker conferma. Con idempotenza scarta le sequenze già viste.
        boolean scrivi(long pid, int seq, String valore, boolean idempotente) {
            if (idempotente && seq <= ultimaSequenza.getOrDefault(pid, -1)) return true;  // duplicato
            log.add(valore);
            ultimaSequenza.put(pid, seq);
            return true;
        }
    }

    // Invia un record; al primo tentativo accade il guasto indicato. Il secondo tentativo va a buon fine.
    static List<String> prova(Guasto guasto, boolean ritenta, boolean idempotente) {
        Broker broker = new Broker();
        long pid = 7;
        int seq = 0;                                           // stessa sequenza a ogni tentativo
        if (guasto == Guasto.RICHIESTA_PERSA) {
            // la richiesta non arriva al broker: nessuna scrittura
        } else {
            broker.scrivi(pid, seq, "R1", idempotente);        // il broker scrive, ma l'ack si perde
        }
        if (ritenta) broker.scrivi(pid, seq, "R1", idempotente);   // il producer non ha avuto risposta
        return broker.log;
    }

    public static void main(String[] args) {
        String[] nomi = {"senza retry          ", "retry, senza idempot.", "retry + idempotenza  "};
        boolean[][] modo = {{false, false}, {true, false}, {true, true}};
        System.out.println("                      richiesta persa | ack perso");
        for (int i = 0; i < 3; i++) {
            System.out.println(nomi[i] + " " + prova(Guasto.RICHIESTA_PERSA, modo[i][0], modo[i][1])
                + " | " + prova(Guasto.ACK_PERSO, modo[i][0], modo[i][1]));
        }
    }
}

// Output:
//                       richiesta persa | ack perso
// senza retry           [] | [R1]
// retry, senza idempot. [R1] | [R1, R1]
// retry + idempotenza   [R1] | [R1]

La tabella dice tutto in sei celle. Senza retry, una richiesta persa significa un record perso ([]) mentre una conferma persa lascia un record scritto, perché il broker lo aveva già. Con i retry, la richiesta persa si risolve ma la conferma persa produce [R1, R1]: il duplicato. Con l'idempotenza entrambe le colonne danno [R1], perché il secondo tentativo porta la stessa coppia (producer id, sequenza) e il broker lo riconosce. Un broker vero fa molto di più (tiene le ultime cinque richieste per producer, rifiuta le sequenze fuori ordine), ma il principio è questo: la deduplicazione sta sul broker, non nel tuo codice.

C'è un limite da conoscere. L'identità del producer vale per la sessione: se il processo si riavvia ottiene un nuovo producer id e il broker non può riconoscere come duplicato ciò che è stato inviato prima del riavvio. Il Javadoc di KafkaProducer lo dice così: «the producer can only guarantee idempotence for messages sent within a single session». Solo con un transactional.id si ottiene un recupero che attraversa più sessioni, ed è l'argomento dell'articolo sulle transazioni. Sempre il Javadoc conferma che dalla 3.0 enable.idempotence vale true per default.

Ordine e richieste in volo

Per andare veloce il producer tiene più richieste in volo verso lo stesso broker (max.in.flight.requests.per.connection, 5). Se la prima fallisce e la seconda no, il retry della prima arriva dopo: le due richieste si invertono. La documentazione del parametro retries lo dice esplicitamente: con retry attivi, idempotenza spenta e più di una richiesta in volo, «the records in the second batch may appear first».

import java.util.*;

public class Ordine {
    static class Broker {
        final List<String> log = new ArrayList<>();
        final boolean idempotente;
        int attesa = 0;                                   // prossima sequenza accettata
        Broker(boolean idempotente) { this.idempotente = idempotente; }

        boolean ricevi(int seq, String valore) {
            if (idempotente) {
                if (seq != attesa) return false;          // fuori sequenza: rifiutato
                attesa++;
            }
            log.add(valore);
            return true;
        }
    }

    // Due richieste in volo verso la stessa partizione: B1 (sequenza 0) si perde, B2 (sequenza 1) arriva.
    static List<String> simula(boolean idempotente) {
        Broker broker = new Broker(idempotente);
        boolean b2Accettato = broker.ricevi(1, "B2");     // arriva per prima
        broker.ricevi(0, "B1");                           // il producer ritenta B1
        if (!b2Accettato) broker.ricevi(1, "B2");         // rifiutato: lo rinvia, ora nell'ordine giusto
        return broker.log;
    }

    public static void main(String[] args) {
        System.out.println("retry, senza idempotenza: " + simula(false));
        System.out.println("retry, con idempotenza:   " + simula(true));
    }
}

// Output:
// retry, senza idempotenza: [B2, B1]
// retry, con idempotenza:   [B1, B2]

Senza idempotenza il log finisce in ordine [B2, B1]. Con l'idempotenza il broker del modello accetta solo la sequenza attesa: rifiuta B2 (sequenza 1 mentre attende la 0), poi accetta B1 e infine B2 ritentato, quindi [B1, B2]. Qui sta anche la ragione del tetto a 5: la documentazione spiega che il broker conserva al massimo cinque batch per producer, e se le richieste in volo fossero di più «previous batches may be removed on broker side». Dalla 4.0, come dicono le note di upgrade, l'impostazione dell'idempotenza non ripiega più da sola quando max.in.flight.requests.per.connection supera 5.

Timeout: il tetto che comanda i retry

L'impostazione che decide quanto a lungo si riprova è delivery.timeout.ms (120000 ms), non retries: è il limite superiore di tempo tra il ritorno di send() e l'esito, retry compresi. La documentazione di retries consiglia di lasciarlo al valore predefinito e di governare il comportamento con il timeout, il quale deve essere almeno la somma di request.timeout.ms (30000 ms) e linger.ms. Scegli il timeout in base a quanto puoi aspettare prima di dire all'utente che è fallito: 120 secondi sono un'eternità per una richiesta web, e per un checkout probabilmente conviene abbassarlo e gestire l'errore nella callback.

Il consumatore: dove si salva la posizione

Con enable.auto.commit=true (il default, ogni 5000 ms) l'offset viene salvato dentro la chiamata a poll(), non da un timer a metà elaborazione. Il Javadoc di KafkaConsumer precisa la condizione per restare at-least-once: devi consumare tutti i record restituiti da ogni poll() prima del successivo. Se non lo fai, scrive il Javadoc, l'offset salvato può superare la posizione effettivamente consumata, con record saltati. Il caso tipico è l'elaborazione affidata a un altro thread. Il modello mostra i due casi.

import java.util.*;

public class CommitAuto {
    // Modello del consumer: l'auto-commit salva la posizione dentro poll(), non in background.
    static class Consumer {
        long posizione = 0, committed = 0;
        List<Long> poll(int quanti) {
            committed = posizione;                          // auto-commit: salva quanto restituito PRIMA
            List<Long> batch = new ArrayList<>();
            for (int i = 0; i < quanti; i++) batch.add(posizione++);
            return batch;
        }
    }

    public static void main(String[] args) {
        // Caso 1: si elabora tutto il batch prima del poll successivo, poi il processo muore a metà del secondo.
        Consumer c1 = new Consumer();
        List<Long> primo = c1.poll(5);                      // offset 0..4
        List<Long> elaborati = new ArrayList<>(primo);      // elaborati tutti e cinque
        List<Long> secondo = c1.poll(5);                    // qui avviene il commit di 5
        elaborati.addAll(secondo.subList(0, 2));            // 5 e 6 elaborati, poi crash
        System.out.println("caso 1: committed=" + c1.committed + ", elaborati=" + elaborati
            + " -> riparte da " + c1.committed + ", rielabora 5 e 6, nulla perso");

        // Caso 2: il batch passa a un altro thread, il poll successivo parte subito.
        Consumer c2 = new Consumer();
        List<Long> batch = c2.poll(5);                      // 0..4 affidati al worker
        List<Long> fatti = new ArrayList<>(batch.subList(0, 3)); // il worker ha finito 0, 1, 2
        c2.poll(5);                                         // commit di 5 mentre il worker è a metà
        List<Long> persi = new ArrayList<>(batch.subList(3, 5));
        System.out.println("caso 2: committed=" + c2.committed + ", elaborati=" + fatti
            + " -> riparte da " + c2.committed + ", offset persi: " + persi);
    }
}

// Output:
// caso 1: committed=5, elaborati=[0, 1, 2, 3, 4, 5, 6] -> riparte da 5, rielabora 5 e 6, nulla perso
// caso 2: committed=5, elaborati=[0, 1, 2] -> riparte da 5, offset persi: [3, 4]

Nel primo caso il commit di 5 avviene all'inizio del secondo poll, quando tutto il primo batch è già elaborato; il crash dopo l'offset 6 costringe a rielaborare 5 e 6, e nulla va perso. Nel secondo il worker ha finito solo 0, 1 e 2 quando il poll successivo salva 5: dopo il crash si riparte da 5 e gli offset 3 e 4 non verranno mai elaborati. Il modello non è il consumatore reale (non ha il timer da 5000 ms, né il fetch), ma ricalca la regola del Javadoc, e la lezione è che l'auto-commit è sicuro solo se l'elaborazione è sequenziale. Altrimenti passa al commit manuale.

Il commit manuale ha due forme: commitSync() salva la posizione corrente dopo l'elaborazione, commitSync(Map) salva offset espliciti per partizione. In entrambe il valore giusto è offset del record + 1. Il listato qui sotto usa il MockConsumer del client Kafka 4.4.0, che esegue il codice vero della libreria senza broker: elabora un batch, salva con commitSync, elabora un secondo batch senza salvarlo e simula un rebalance che toglie una partizione. Il ConsumerRebalanceListener salva gli offset della partizione che sta per perdere.

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.time.Duration;
import java.util.*;

public class CommitManuale {
    public static void main(String[] args) {
        MockConsumer<String, String> consumer = new MockConsumer<>("earliest");   // client reale, senza broker
        TopicPartition p0 = new TopicPartition("ordini", 0), p1 = new TopicPartition("ordini", 1);
        Map<TopicPartition, OffsetAndMetadata> daCommitare = new HashMap<>();

        consumer.subscribe(List.of("ordini"), new ConsumerRebalanceListener() {
            @Override public void onPartitionsRevoked(Collection<TopicPartition> perse) {
                Map<TopicPartition, OffsetAndMetadata> mie = new HashMap<>();
                for (TopicPartition tp : perse)
                    if (daCommitare.containsKey(tp)) mie.put(tp, daCommitare.remove(tp));
                if (!mie.isEmpty()) consumer.commitSync(mie);          // salva prima di perdere la partizione
                System.out.println("revocate " + perse + ", salvato " + offsets(mie));
            }
            @Override public void onPartitionsAssigned(Collection<TopicPartition> nuove) {
                System.out.println("assegnate " + nuove);
            }
        });
        consumer.rebalance(List.of(p0, p1));
        consumer.updateBeginningOffsets(Map.of(p0, 0L, p1, 0L));
        aggiungi(consumer, 0, 3);
        elabora(consumer, daCommitare);
        consumer.commitSync(daCommitare);                              // commit manuale, dopo l'elaborazione
        System.out.println("committed: " + offsets(consumer.committed(Set.of(p0, p1))));
        daCommitare.clear();

        aggiungi(consumer, 3, 5);
        elabora(consumer, daCommitare);                                // elaborati ma NON ancora salvati
        System.out.println("da salvare: " + offsets(daCommitare));
        consumer.rebalance(List.of(p0));                               // p1 passa a un altro consumatore
    }

    static void aggiungi(MockConsumer<String, String> c, long da, long a) {
        for (long i = da; i < a; i++) {
            c.addRecord(new ConsumerRecord<>("ordini", 0, i, "a", "ordine-0-" + i));
            c.addRecord(new ConsumerRecord<>("ordini", 1, i, "b", "ordine-1-" + i));
        }
    }

    static void elabora(MockConsumer<String, String> c, Map<TopicPartition, OffsetAndMetadata> daCommitare) {
        for (ConsumerRecord<String, String> r : c.poll(Duration.ofMillis(10))) {
            // qui andrebbe la logica di business; poi si annota il PROSSIMO offset da leggere
            daCommitare.put(new TopicPartition(r.topic(), r.partition()), new OffsetAndMetadata(r.offset() + 1));
        }
    }

    static String offsets(Map<TopicPartition, OffsetAndMetadata> m) {
        TreeMap<String, String> ordinata = new TreeMap<>();
        m.forEach((tp, o) -> ordinata.put(tp.toString(), o == null ? "null" : String.valueOf(o.offset())));
        return ordinata.toString();
    }
}

// Output:
// assegnate [ordini-0, ordini-1]
// committed: {ordini-0=3, ordini-1=3}
// da salvare: {ordini-0=5, ordini-1=5}
// revocate [ordini-1], salvato {ordini-1=5}
// assegnate []

La seconda riga mostra l'offset committed dopo il primo batch: 3 per ciascuna partizione, cioè l'offset dell'ultimo record (2) più uno. Dopo il secondo batch gli offset da salvare sono 5, ma nessuno li ha ancora commitati. Il rebalance simulato revoca ordini-1 e il listener lo salva: «salvato {ordini-1=5}». Chi riceverà quella partizione ripartirà dal 5 e non rielaborerà i record 3 e 4. Il MockConsumer è un doppio di test, quindi non dimostra come si comporta il coordinatore su un cluster vero; dimostra l'uso corretto dell'API e il suo contratto.

Rebalance: fermare tutti o fermare solo chi cambia

Quando un consumatore entra o esce, il gruppo riassegna le partizioni. Con il protocollo classico il comportamento dipende dall'assegnatore. L'elenco predefinito di partition.assignment.strategy è [RangeAssignor, CooperativeStickyAssignor] e, come dice la documentazione, il gruppo usa il RangeAssignor finché non fai un rolling restart passando al cooperativo: il RangeAssignor fa un rebalance eager, in cui ogni consumatore rilascia tutte le sue partizioni e il gruppo si ferma. Il CooperativeStickyAssignor (KIP-429) revoca solo le partizioni che cambiano proprietario.

import java.util.*;

public class Rebalance {
    static final int PARTIZIONI = 6;

    public static void main(String[] args) {
        // Stato prima: c1 ha 0,1,2 e c2 ha 3,4,5. Entra c3: l'obiettivo è 2 partizioni a testa.
        Map<String, List<Integer>> prima = new LinkedHashMap<>();
        prima.put("c1", List.of(0, 1, 2));
        prima.put("c2", List.of(3, 4, 5));
        int obiettivo = PARTIZIONI / 3;

        // Eager: tutti rilasciano tutto, poi si riassegna da zero.
        int revocateEager = PARTIZIONI;

        // Cooperative (modello sticky): ognuno tiene fino a 'obiettivo' partizioni, cede il resto a c3.
        List<Integer> cedute = new ArrayList<>();
        for (List<Integer> mie : prima.values())
            cedute.addAll(mie.subList(obiettivo, mie.size()));

        System.out.println("eager:       revocate " + revocateEager + " partizioni su " + PARTIZIONI
            + ", tutto il gruppo si ferma");
        System.out.println("cooperative: revocate " + cedute.size() + " partizioni " + cedute
            + ", le altre " + (PARTIZIONI - cedute.size()) + " continuano a leggere");
    }
}

// Output:
// eager:       revocate 6 partizioni su 6, tutto il gruppo si ferma
// cooperative: revocate 2 partizioni [2, 5], le altre 4 continuano a leggere

Con sei partizioni e un terzo membro che deve riceverne due, il rebalance eager ferma tutto il gruppo e revoca sei partizioni; il cooperativo ne revoca solo due (la 2 e la 5), mentre le altre quattro continuano a leggere. Il modello usa un criterio sticky semplificato e ignora il secondo giro di messaggi del protocollo reale, che serve a riassegnare le partizioni revocate. Esiste anche il nuovo protocollo (KIP-848), che si attiva con group.protocol=consumer: l'assegnazione la decide il coordinatore sul broker e non c'è più la barriera globale. Il default resta classic nella documentazione 4.4, quindi devi sceglierlo tu.

Resta un'impostazione che causa rebalance «inspiegabili»: max.poll.interval.ms (300000 ms). Se il tuo ciclo non chiama poll() entro quell'intervallo, il consumatore lascia il gruppo di sua iniziativa e il successivo commitSync fallisce con CommitFailedException. Il Javadoc propone due leve: abbassare max.poll.records (default 500) per accorciare il lavoro per iterazione, oppure spostare l'elaborazione su un altro thread, ricordando però la regola dell'auto-commit vista sopra.

Che cosa regolare davvero

La tabella raccoglie i parametri di tuning che ha senso toccare, con i default della documentazione 4.4. Nessuna cifra di throughput qui: ogni valore dipende dal carico, e la sola maniera onesta di sceglierlo è provare.

Una regola dal documento dei producer vale più di ogni tabella: la compressione agisce su lotti interi, quindi dipende da linger.ms e batch.size. Scegli un algoritmo (gzip, snappy, lz4, zstd) solo dopo aver misurato CPU e dimensione dei messaggi sui tuoi dati.

Che cosa sai fare adesso

Sai leggere la configurazione predefinita di un producer 4.4 e perché è già sicura, e sai riconoscere i tre modi di romperla: acks=1 con l'idempotenza non dichiarata, più di cinque richieste in volo, min.insync.replicas lasciato a 1. Sai spiegare un duplicato come una conferma persa e un riordino come un retry fuori sequenza. Sul consumatore sai quando l'auto-commit è sicuro, scrivi un commit manuale con l'offset +1 e un listener che salva prima di perdere una partizione, e sai che il rebalance classico di default ferma tutto il gruppo. Per le scritture atomiche su più topic passa alle transazioni; per i messaggi che non riesci a elaborare, alla dead letter queue. Se arrivi da un altro punto della serie, i collegamenti sono: KRaft e i controller, lo Schema Registry, il monitoraggio, Kafka in produzione.