04 - Exactly-Once Semantics in Kafka: Transazioni e Coordinamento

Il servizio che elabora un ordine e cade a metà

Hai un servizio che legge ordini da un topic, li arricchisce e scrive il risultato su un altro topic. Per ogni batch fa tre cose in fila: scrive l'output, salva l'offset letto, passa al batch successivo. Se il processo muore tra la prima e la seconda, al riavvio ripete il batch e l'output compare due volte. Se invertissi i passi (prima l'offset, poi l'output) un crash nel mezzo farebbe perdere il batch. Con due operazioni separate non esiste un ordine giusto: serve un modo per farle diventare una sola.

È ciò che le transazioni di Kafka offrono, ed è il cuore di quella che la documentazione chiama exactly-once semantics. La formula completa richiede quattro elementi che lavorano insieme: un producer idempotente, un transactional.id, un consumatore configurato con isolation.level=read_committed e enable.auto.commit=false, e la scrittura dell'offset del consumatore dentro la transazione. Vale solo per i dati che restano dentro Kafka: un database esterno non ne fa parte, e lo vedremo in fondo. Fatti e valori di configurazione vengono dalla documentazione di Kafka 4.4 (consultata il 2026-11-12) e dal Javadoc del client; i comportamenti li ho riprodotti con modelli in Java 21 e con le classi Mock* del client Kafka 4.4.0, senza un broker.

Perché due operazioni separate non bastano: il modello

Il modello seguente confronta due versioni dello stesso servizio. Elabora tre record di input (offset da 0 a 2) e cade prima di salvare l'offset. Alla ripartenza rielabora dall'offset salvato, che è ancora 0. La differenza è se gli output del primo tentativo sono già visibili a un lettore oppure sono rimasti dentro una transazione mai completata.

import java.util.*;

public class Atomicita {
    // Elabora tre record di input e scrive tre output; il processo cade prima della conferma.
    // Poi riparte dall'offset salvato e rielabora. Cosa vede un lettore dell'output?
    static List<String> corsa(boolean transazionale) {
        List<String> output = new ArrayList<>();         // output visibile ai lettori
        List<String> inTransazione = new ArrayList<>();  // scritto, non ancora visibile
        final long offsetSalvato = 0;

        // Primo tentativo: scrive gli output...
        for (long o = offsetSalvato; o < 3; o++) {
            if (transazionale) inTransazione.add("out-" + o); else output.add("out-" + o);
        }
        // ...e cade PRIMA di salvare l'offset (e, nel caso transazionale, prima del commit).
        if (transazionale) inTransazione.clear();        // la transazione viene abortita: i record restano nascosti

        // Riavvio: riparte da 0 perché l'offset non è stato salvato.
        for (long o = offsetSalvato; o < 3; o++) output.add("out-" + o);
        return output;
    }

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

// Output:
// senza transazione: [out-0, out-1, out-2, out-0, out-1, out-2]
// con transazione:   [out-0, out-1, out-2]

Senza transazione il lettore vede sei record: i tre del primo tentativo più i tre rielaborati. Con la transazione il primo tentativo viene abortito (la riga inTransazione.clear() rappresenta l'abort) e il lettore vede solo i tre record del secondo. Il modello è volutamente riduttivo: non mostra come Kafka renda quei record invisibili. Quello lo fanno i marker, e li vedremo tra poco. Il punto da fissare qui è la condizione: output e offset devono stare nella stessa transazione, altrimenti non c'è atomicità.

L'API transazionale, nell'ordine in cui va chiamata

Il Javadoc del KafkaProducer fissa l'ordine: con transactional.id impostato, initTransactions() va chiamato una volta prima di ogni altra operazione; poi, per ogni transazione, beginTransaction(), le send(), sendOffsetsToTransaction() e infine commitTransaction() oppure abortTransaction(). Un producer transazionale ha una sola transazione aperta alla volta e tutti i suoi invii devono stare dentro una transazione.

Il listato qui sotto percorre la sequenza con il MockProducer di kafka-clients-4.4.0.jar (Maven Central, SHA-1 verificato): è codice della libreria reale, ma è un doppio di test, non parla con un broker. Quello che dimostra è il contratto dell'API, non la semantica distribuita. Nella vita vera l'offset si ricava dal consumatore con consumer.groupMetadata(), e qui un MockConsumer lo fornisce.

import org.apache.kafka.clients.consumer.MockConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Map;

public class ApiTransazionale {
    public static void main(String[] args) {
        MockProducer<String, String> producer =
            new MockProducer<>(true, null, new StringSerializer(), new StringSerializer());

        try {
            producer.beginTransaction();                       // senza initTransactions()
        } catch (IllegalStateException e) {
            System.out.println("beginTransaction prima di initTransactions: " + e.getMessage());
        }

        producer.initTransactions();
        Map<TopicPartition, OffsetAndMetadata> offset =
            Map.of(new TopicPartition("ordini-input", 0), new OffsetAndMetadata(3));
        MockConsumer<String, String> consumer = new MockConsumer<>("earliest");   // fornisce groupMetadata()

        producer.beginTransaction();                           // transazione 1: va a buon fine
        producer.send(new ProducerRecord<>("ordini-elaborati", "o1", "ok-1"));
        producer.send(new ProducerRecord<>("ordini-elaborati", "o2", "ok-2"));
        producer.sendOffsetsToTransaction(offset, consumer.groupMetadata());     // la posizione del consumatore, nella stessa transazione
        producer.commitTransaction();
        System.out.println("transazione 1 committed: " + producer.transactionCommitted()
            + ", record visibili: " + producer.history().size()
            + ", offset committati: " + producer.consumerGroupOffsetsHistory().size());

        producer.beginTransaction();                           // transazione 2: viene abortita
        producer.send(new ProducerRecord<>("ordini-elaborati", "o3", "ok-3"));
        producer.abortTransaction();
        System.out.println("transazione 2 abortita: " + producer.transactionAborted()
            + ", record visibili: " + producer.history().size());
    }
}

// Output:
// beginTransaction prima di initTransactions: MockProducer hasn't been initialized for transactions.
// transazione 1 committed: true, record visibili: 2, offset committati: 1
// transazione 2 abortita: true, record visibili: 2

La prima riga di output mostra l'errore se salti initTransactions(): beginTransaction() lancia IllegalStateException (il testo del messaggio è quello del MockProducer; il Javadoc del producer vero documenta la stessa eccezione per questo caso). La transazione 1 invia due record e l'offset di input 3 con sendOffsetsToTransaction: dopo il commit la cronologia contiene due record e un offset. La transazione 2 invia un terzo record e la abortisce: i record visibili restano due, perché quello abortito non entra nella cronologia. Su un broker vero il record abortito è scritto nel log e nascosto dai marker.

Per la gestione degli errori il Javadoc distingue due famiglie. ProducerFencedException, OutOfOrderSequenceException e AuthorizationException non sono recuperabili: l'unica mossa è chiudere il producer. Per gli altri KafkaException si chiama abortTransaction() e si riprova. La sezione «Design» aggiunge una precisazione che è facile dimenticare: dopo un abort bisogna anche riportare indietro la posizione del consumatore, perché i record di quel batch vanno riletti. Sulle categorie d'errore (retriable, abortable, fatali) più recenti vale la pagina di Design 4.4.

I marker e chi legge: read_committed e LSO

Un producer transazionale scrive i suoi record normalmente nel log della partizione, ma quando la transazione finisce il broker aggiunge un marker di COMMIT o di ABORT. Un consumatore con isolation.level=read_committed (il default è read_uncommitted) scarta i record delle transazioni abortite e non supera il last stable offset (LSO), cioè il primo offset di una transazione ancora aperta. La documentazione del consumatore lo dice: con read_committed, poll() restituisce record «up to the last stable offset».

import java.util.*;

public class Isolamento {
    enum Tipo { DATO, COMMIT, ABORT }
    record Voce(Tipo tipo, String txn, String valore) {}      // txn == null: record non transazionale

    static List<Voce> log = new ArrayList<>();
    static void dato(String txn, String v) { log.add(new Voce(Tipo.DATO, txn, v)); }
    static void chiudi(String txn, Tipo esito) { log.add(new Voce(esito, txn, null)); }

    // LSO: primo offset di una transazione ancora aperta (o la fine del log se non ce ne sono).
    static int lso() {
        Set<String> chiuse = new HashSet<>();
        for (Voce v : log) if (v.tipo() != Tipo.DATO) chiuse.add(v.txn());
        for (int i = 0; i < log.size(); i++) {
            Voce v = log.get(i);
            if (v.tipo() == Tipo.DATO && v.txn() != null && !chiuse.contains(v.txn())) return i;
        }
        return log.size();
    }

    static List<String> leggi(boolean readCommitted) {
        Map<String, Tipo> esito = new HashMap<>();
        for (Voce v : log) if (v.tipo() != Tipo.DATO) esito.put(v.txn(), v.tipo());
        List<String> visti = new ArrayList<>();
        int limite = readCommitted ? lso() : log.size();
        for (int i = 0; i < limite; i++) {
            Voce v = log.get(i);
            if (v.tipo() != Tipo.DATO) continue;                          // i marker non arrivano all'applicazione
            if (readCommitted && v.txn() != null && esito.get(v.txn()) == Tipo.ABORT) continue;
            visti.add(v.valore());
        }
        return visti;
    }

    public static void main(String[] args) {
        dato(null, "x");                 // 0  non transazionale
        dato("T1", "a");                 // 1
        dato("T1", "b");                 // 2
        dato("T2", "c");                 // 3
        chiudi("T1", Tipo.COMMIT);       // 4
        dato("T2", "d");                 // 5
        chiudi("T2", Tipo.ABORT);        // 6
        dato("T3", "e");                 // 7  ancora aperta
        System.out.println("read_uncommitted: " + leggi(false));
        System.out.println("read_committed:   " + leggi(true) + "  (LSO=" + lso() + ")");

        // Una transazione lenta blocca chi legge anche i dati committed che la seguono.
        log.clear();
        dato("T1", "a");                 // 0  T1 resta aperta
        dato("T2", "f");                 // 1
        chiudi("T2", Tipo.COMMIT);       // 2  T2 è già committed
        System.out.println("T1 aperta, T2 committed -> read_committed vede " + leggi(true) + " (LSO=" + lso() + ")");
        chiudi("T1", Tipo.COMMIT);
        System.out.println("dopo il commit di T1       -> read_committed vede " + leggi(true));
    }
}

// Output:
// read_uncommitted: [x, a, b, c, d, e]
// read_committed:   [x, a, b]  (LSO=7)
// T1 aperta, T2 committed -> read_committed vede [] (LSO=0)
// dopo il commit di T1       -> read_committed vede [a, f]

Nella prima scena il log ha record di tre transazioni più uno non transazionale. Un lettore read_uncommitted vede tutto (x a b c d e); uno read_committed vede solo x a b: c e d appartengono a T2, abortita, mentre e è di T3, ancora aperta, e sta oltre l'LSO (7). La seconda scena è la conseguenza che pesa in produzione: T1 è aperta all'offset 0 e T2 ha già fatto commit, ma l'LSO è 0, quindi read_committed non vede nulla, nemmeno f. Solo dopo il commit di T1 compaiono entrambi. Una transazione lenta o dimenticata blocca i lettori di quella partizione, e per questo esistono i limiti di tempo: transaction.timeout.ms (60000 ms di default sul producer) e il tetto del broker transaction.max.timeout.ms (900000 ms), con cui il coordinatore abortisce da solo le transazioni troppo lunghe.

Il coordinatore e il fencing degli zombie

Il transaction coordinator gira nei broker e tiene lo stato delle transazioni nel topic interno __transaction_state, che ha 50 partizioni (transaction.state.log.num.partitions) e, per default, replication factor 3 e transaction.state.log.min.isr 2. Per questo la documentazione avverte che le transazioni, per default, richiedono un cluster di almeno tre broker; in sviluppo si abbassano i due parametri.

Il transactional.id identifica un'istanza logica del producer attraverso i riavvii. Quando una nuova istanza chiama initTransactions() con lo stesso identificatore, il coordinatore aumenta l'epoca, aborta l'eventuale transazione lasciata aperta e, da quel momento, rifiuta i commit di chi ha ancora l'epoca vecchia. È la protezione contro lo «zombie»: un producer che è stato in pausa (una pausa del garbage collector, una partizione di rete) e si risveglia dopo che il suo sostituto è già partito.

import java.util.*;

public class Fencing {
    static final Map<String, Integer> epoca = new HashMap<>();     // stato nel coordinatore delle transazioni

    static int initTransactions(String transactionalId) {
        return epoca.merge(transactionalId, 1, Integer::sum);      // ogni init alza l'epoca
    }

    static String commit(String transactionalId, int epocaProducer) {
        return epocaProducer < epoca.get(transactionalId) ? "ProducerFencedException" : "commit accettato";
    }

    public static void main(String[] args) {
        int a = initTransactions("ordini-0");                      // il producer A parte
        System.out.println("A: epoca " + a);
        int b = initTransactions("ordini-0");                      // A si blocca, parte B con lo stesso id
        System.out.println("B: epoca " + b);
        System.out.println("A tenta il commit: " + commit("ordini-0", a));
        System.out.println("B tenta il commit: " + commit("ordini-0", b));
    }
}

// Output:
// A: epoca 1
// B: epoca 2
// A tenta il commit: ProducerFencedException
// B tenta il commit: commit accettato

L'epoca di A è 1, quella di B 2. Quando A, risvegliato, tenta il commit, il coordinatore (qui un Map) vede l'epoca 1 minore di 2 e risponde con ProducerFencedException; B passa. Il modello non copre il fatto che dalla 4.0 l'epoca viene incrementata a ogni transazione, non solo a ogni inizializzazione: è il protocollo «server-side defense» di KIP-890, attivo per default sul server dalla 4.0 e governato dal flag transaction.version. Come regola per scegliere l'identificatore il Javadoc dice che di solito si deriva dall'identità dello shard di un'applicazione partizionata e con stato; la sezione Design aggiunge di usare un producer per ogni consumatore.

Kafka Streams: la stessa cosa con una proprietà

Se il tuo flusso è un'applicazione Kafka Streams non scrivi nulla di tutto questo: imposti processing.guarantee a exactly_once_v2 e la libreria gestisce producer transazionale, consumatore read_committed e offset nella transazione. Il listato crea la configurazione con la libreria 4.4.0 (kafka-streams-4.4.0.jar) e stampa due cose che conviene sapere a memoria: i valori ammessi e l'intervallo di commit.

import org.apache.kafka.streams.StreamsConfig;
import java.util.*;

public class StreamsEos {
    static StreamsConfig config(String garanzia) {
        Map<String, Object> p = new HashMap<>();
        p.put(StreamsConfig.APPLICATION_ID_CONFIG, "arricchimento-ordini");
        p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");   // non viene contattato
        if (garanzia != null) p.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, garanzia);
        return new StreamsConfig(p);
    }

    public static void main(String[] args) {
        for (String g : new String[] {null, StreamsConfig.EXACTLY_ONCE_V2}) {
            StreamsConfig c = config(g);
            System.out.println("processing.guarantee = " + c.getString(StreamsConfig.PROCESSING_GUARANTEE_CONFIG)
                + ", commit.interval.ms = " + c.getLong(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG));
        }
    }
}

// Output:
// processing.guarantee = at_least_once, commit.interval.ms = 30000
// processing.guarantee = exactly_once_v2, commit.interval.ms = 100

Con la garanzia predefinita il commit avviene ogni 30000 ms; con exactly_once_v2 la libreria porta l'intervallo a 100 ms, perché il commit coincide con il commit della transazione e con il momento in cui l'output diventa visibile ai lettori read_committed. Un intervallo più alto riduce il numero di transazioni ma allunga la latenza con cui i dati diventano visibili: è questo il compromesso, e lo regoli tu. In 4.4 i valori ammessi sono due, at_least_once e exactly_once_v2; la versione 1 non c'è più. Per i dettagli di Streams, compresi i broker minimi, vale la relativa pagina della serie.

Quanto costa, e quando non basta

Non ti do qui una percentuale di throughput perduto: nessuna cifra attendibile esiste fuori da una misura sul tuo hardware e sul tuo carico, e quelle che circolano senza descrizione del test vanno trattate con sospetto. I costi sono qualitativi ma concreti. Ogni transazione aggiunge scambi con il coordinatore e scritture di marker, quindi conviene transazioni con più record. I lettori read_committed vedono i dati più tardi, al commit. Le transazioni lunghe bloccano i lettori. E il parametro più importante da regolare è la durata: grandi lotti per transazione diminuiscono il costo per record e aumentano il ritardo di visibilità.

Il confine più importante è che le garanzie finiscono dove finisce Kafka. La documentazione lo dice a proposito della consegna verso altri sistemi: serve la cooperazione di quel sistema. Se il tuo servizio, dopo aver letto un ordine, scrive su PostgreSQL, la scrittura nel database non fa parte della transazione Kafka. Le due strade sensate sono rendere idempotente il consumatore (una chiave univoca che il database rifiuta la seconda volta, vedi l'articolo sull'idempotenza nei consumer) oppure pubblicare gli eventi dallo stato del database con il pattern outbox. Spesso la prima costa meno di qualsiasi transazione.

Che cosa sai fare adesso

Sai spiegare perché scrivere l'output e salvare l'offset in due passi non può essere né sicuro né atomico, e come una transazione li unisce. Conosci l'ordine delle chiamate dell'API transazionale, quali eccezioni sono fatali e quali si risolvono con un abort. Sai leggere un log con marker: perché un lettore read_committed salta i record abortiti e si ferma all'LSO, e perché una transazione aperta a lungo blocca tutti. Sai cosa fa un transactional.id e perché due istanze con lo stesso non convivono. Sai quando affidarti a exactly_once_v2 in Streams e che la garanzia non attraversa il confine verso un database esterno. Se vuoi spiegare i duplicati dall'altra parte, quelli generati dai retry del producer, parti dall'articolo precedente. Se arrivi da un altro punto della serie, i collegamenti sono: i fondamenti di topic e partizioni, Flink e Iceberg.