01 - Apache Kafka Fondamenti: Topics, Partizioni, Offset e Consumer Group

Perché una coda che cancella non basta

Immagina un negozio online. Ogni ordine deve raggiungere il magazzino, la fatturazione e il servizio che calcola le statistiche, e ciascuno lavora al suo ritmo: il magazzino in tempo reale, le statistiche una volta all'ora. Tra tre mesi arriverà un quarto servizio, per i suggerimenti di acquisto, che vorrà rileggere anche gli ordini del passato. Con una coda tradizionale, in cui il messaggio sparisce quando un consumatore lo conferma, il quarto servizio parte da zero e il passato è perso.

Apache Kafka risolve il problema cambiando il modello: non una coda che si svuota, ma un log in cui i record si aggiungono sempre in fondo e restano al loro posto finché non scade la retention. Leggere non cancella nulla. Ogni lettore ricorda per conto suo fino a dove è arrivato e può tornare indietro. Questa è la differenza che tiene in piedi tutto il resto dell'articolo, e la introduzione della documentazione ufficiale la presenta nello stesso modo: eventi conservati per tutto il tempo che serve e riletti quando occorre.

Una precisazione per non fare torto alle code classiche: anche RabbitMQ offre oggi gli stream, log che si possono rileggere finché non scadono, quindi il confronto giusto non è «Kafka contro le code» ma «log contro coda con conferma». In questo primo articolo della serie costruisci le basi: topic, partizioni, offset, gruppi di consumatori, replica, retention. Per ognuno trovi un piccolo modello in Java 21, che ho compilato ed eseguito, e i valori di configurazione letti nella documentazione di Kafka 4.4 (la versione corrente, pubblicata il 22 ottobre 2026, consultata il 2026-11-12).

Topic e partizioni: il log che si scrive solo in fondo

Un topic è il nome sotto cui i produttori pubblicano e i consumatori leggono, per esempio ordini-effettuati. È suddiviso in partizioni, e la partizione è l'unità che conta davvero: un log ordinato, immutabile, in cui ogni record riceve un numero progressivo chiamato offset (0, 1, 2, ...). L'ordine è garantito dentro la partizione e non tra partizioni diverse: la documentazione scrive che «events with the same event key (e.g., a customer or vehicle ID) are written to the same partition» e che chi legge una partizione vede gli eventi nell'ordine in cui sono stati scritti.

Come sceglie il produttore la partizione? Se il record ha una chiave, la scelta dipende da un hash della chiave; se la chiave manca, il partizionatore predefinito riempie una partizione fino a batch.size byte e poi passa a un'altra (è la descrizione di partitioner.class nella documentazione dei producer). Il modello qui sotto riproduce la prima regola con l'hashCode della stringa e il resto della divisione per il numero di partizioni. Kafka usa un'altra funzione di hash, quindi le partizioni scelte da un broker vero saranno diverse, ma la proprietà che conta è la stessa: stessa chiave, stessa partizione.

import java.util.*;

public class Partizioni {
    record Record(long offset, String chiave, String valore) {}

    static class Topic {
        final List<List<Record>> partizioni = new ArrayList<>();
        Topic(int n) { for (int i = 0; i < n; i++) partizioni.add(new ArrayList<>()); }

        static int dove(String chiave, int n) {            // modello: hash della chiave modulo n
            return Math.floorMod(chiave.hashCode(), n);
        }
        void scrivi(String chiave, String valore) {
            List<Record> log = partizioni.get(dove(chiave, partizioni.size()));
            log.add(new Record(log.size(), chiave, valore)); // l'offset è la posizione nel log
        }
    }

    public static void main(String[] args) {
        Topic ordini = new Topic(3);
        String[] clienti = {"anna", "marco", "anna", "luca", "marco", "anna", "luca", "anna"};
        for (int i = 0; i < clienti.length; i++) ordini.scrivi(clienti[i], "ordine-" + i);

        for (int p = 0; p < 3; p++) {
            System.out.print("partizione " + p + ":");
            for (Record r : ordini.partizioni.get(p))
                System.out.print(" [" + r.offset() + " " + r.chiave() + " " + r.valore() + "]");
            System.out.println();
        }

        int spostate = 0, totale = 1000;
        for (int k = 0; k < totale; k++) {
            String chiave = "cliente-" + k;
            if (Topic.dove(chiave, 6) != Topic.dove(chiave, 12)) spostate++;
        }
        System.out.println("chiavi che cambiano partizione passando da 6 a 12: " + spostate + " su " + totale);
    }
}

// Output:
// partizione 0: [0 anna ordine-0] [1 anna ordine-2] [2 anna ordine-5] [3 anna ordine-7]
// partizione 1: [0 luca ordine-3] [1 luca ordine-6]
// partizione 2: [0 marco ordine-1] [1 marco ordine-4]
// chiavi che cambiano partizione passando da 6 a 12: 500 su 1000

Il metodo dove è tutto il partizionamento: con tre partizioni, anna finisce sempre nella 0, luca nella 1 e marco nella 2. Guarda gli offset: nella partizione 0 i quattro ordini di Anna hanno offset 0, 1, 2, 3 e compaiono nell'ordine in cui sono stati inviati (ordine-0, ordine-2, ordine-5, ordine-7). Questo è ciò che significa «ordinamento per chiave»: i suoi ordini non si scambieranno mai di posto, mentre tra Anna e Marco non c'è alcun ordine garantito. L'ultima riga dell'output è la trappola: passando da 6 a 12 partizioni, 500 chiavi su 1000 cambiano partizione. L'output è deterministico perché String.hashCode è specificato.

Da qui discende una regola pratica che nessun cluster ti ricorda. La documentazione operativa descrive come aumentare le partizioni (kafka-topics.sh --alter --partitions) e avverte che, se i dati sono partizionati con hash(chiave) modulo il numero di partizioni, la mappatura cambia: record con la stessa chiave possono finire in partizioni diverse dopo l'espansione, con effetto sull'ordine per le chiavi già esistenti, e Kafka non redistribuisce i dati già scritti. Per questo conviene scegliere il numero con un po' di margine fin dall'inizio, ragionando su due quantità che puoi misurare: quanti consumatori vuoi far girare in parallelo (non ne serviranno più delle partizioni, vedi sotto) e quanti MB al secondo deve reggere il topic, divisi per quanto regge una partizione sul tuo hardware. Nessun numero «tipico» vale al posto di una prova sul tuo disco e sulla tua rete.

Broker, replica e il momento in cui un record è «scritto»

Un broker è un server Kafka. Ogni partizione ha un leader, che serve letture e scritture, e zero o più follower che copiano il log del leader. L'insieme delle repliche che stanno al passo si chiama ISR (in-sync replicas). Un follower esce dall'ISR se non chiede dati o non raggiunge la fine del log del leader per più di replica.lag.time.max.ms, che per default vale 30 secondi (documentazione dei broker, 4.4).

La definizione che regge il resto è questa, ricavata dalla sezione «Design» della documentazione: un record è committed quando tutte le repliche dell'ISR l'hanno applicato, e un record committed non va perso finché almeno una di quelle repliche resta viva. I consumatori non vedono un record prima di quel momento, e nemmeno se le repliche in-sync sono meno di min.insync.replicas. Il default di min.insync.replicas è 1: se vuoi che acks=all significhi davvero «almeno due copie», con replication factor 3 imposta 2 a livello di topic. Il costo è la disponibilità: con due broker spenti su tre il topic rifiuta le scritture, ma non perde nulla.

Gli offset che incontrerai nei comandi e nei grafici hanno nomi simili e significati diversi. Li conviene separare subito:

• Log end offset (LEO): l'offset che avrà il prossimo record scritto nel log di una replica.
• High watermark (HW): il limite di visibilità. Per convenzione è l'offset del primo record non ancora visibile ai consumatori, cioè tutto ciò che sta prima è già replicato sull'ISR.
• Offset corrente: il prossimo record che quel consumatore leggerà con la poll() seguente.
• Offset committed: il valore salvato per il gruppo nel topic interno __consumer_offsets; è il punto da cui si riparte dopo un crash o un cambio di assegnazione.

I primi due vivono sul broker e dicono quanto il log è avanzato e quanto è visibile; gli altri due vivono sul lato del consumatore e dicono quanto è stato letto e quanto è stato salvato. Il lag che vedrai nei cruscotti è la distanza tra LEO e offset committed del gruppo: più è grande, più il gruppo è indietro. Attenzione a un errore comune: l'offset committed non è «l'ultimo record elaborato» ma il prossimo da leggere, cioè ultimo elaborato più uno, come specifica il Javadoc di KafkaConsumer: «the committed offset should be the next message your application will consume».

import java.util.*;

public class OffsetCommit {
    public static void main(String[] args) {
        // Il leader ha scritto 10 record (offset 0..9), il LEO è 10.
        // I follower dell'ISR hanno copiato fino a 8 e 9 record: il più lento fissa l'high watermark.
        long leo = 10;
        long[] followerLeo = {8, 9};
        long hw = Math.min(leo, Math.min(followerLeo[0], followerLeo[1]));
        System.out.println("LEO=" + leo + " HW=" + hw + " -> il consumatore vede gli offset 0.." + (hw - 1));

        // Il log è ora completamente replicato: 10 record visibili.
        long committed = 0;                      // offset salvato: il PROSSIMO record da leggere
        long lag = leo - committed;
        System.out.println("committed=" + committed + " lag=" + lag);

        // Scena 1: elabora i record 0..5, poi COMMIT, poi si ferma: ripartenza senza duplicati.
        for (long o = 0; o <= 5; o++) System.out.println("  elaboro offset " + o);
        committed = 5 + 1;                       // ultimo elaborato + 1
        System.out.println("commit -> " + committed + ", lag=" + (leo - committed));

        // Scena 2: elabora 6..8, poi il processo muore PRIMA del commit.
        for (long o = 6; o <= 8; o++) System.out.println("  elaboro offset " + o);
        System.out.println("crash prima del commit: l'offset salvato resta " + committed);

        // Alla ripartenza il consumatore legge dall'offset salvato: 6, 7, 8 sono rielaborati.
        System.out.println("ripartenza da offset " + committed + ": 6, 7, 8 arrivano di nuovo");
    }
}

// Output:
// LEO=10 HW=8 -> il consumatore vede gli offset 0..7
// committed=0 lag=10
//   elaboro offset 0
//   elaboro offset 1
//   elaboro offset 2
//   elaboro offset 3
//   elaboro offset 4
//   elaboro offset 5
// commit -> 6, lag=4
//   elaboro offset 6
//   elaboro offset 7
//   elaboro offset 8
// crash prima del commit: l'offset salvato resta 6
// ripartenza da offset 6: 6, 7, 8 arrivano di nuovo

Nella prima parte l'high watermark vale 8, il minimo tra i LEO delle repliche (10, 9 e 8): il consumatore vede gli offset da 0 a 7 anche se il leader ne ha scritti 10. Nella seconda parte il consumatore elabora fino all'offset 5 e salva 5 + 1: quell'«più uno» è la riga che conta. Poi elabora 6, 7 e 8 e muore prima del commit. Alla ripartenza legge dall'offset salvato e quei tre record arrivano di nuovo. È la semantica at-least-once della documentazione: nessuna perdita, possibili duplicati. Un broker vero fa in più il commit su __consumer_offsets e la riassegnazione automatica della partizione a un altro consumatore del gruppo; la logica dell'offset salvato è però esattamente questa.

Il consumer group: come si scala la lettura

Tutti i consumatori con lo stesso group.id formano un gruppo e si dividono le partizioni del topic. La regola è una sola: una partizione è assegnata a un solo consumatore del gruppo alla volta. Ne segue che il parallelismo massimo di un gruppo è il numero di partizioni: con sei partizioni, il settimo consumatore non lavora, resta in attesa e subentra se un altro cade. Gruppi diversi sono indipendenti: ciascuno ha i suoi offset e riceve tutti i record del topic, ed è così che magazzino e fatturazione leggono lo stesso flusso senza disturbarsi. Che cosa sia un evento di dominio e perché un ordine pubblica un fatto invece di chiamare altri servizi è nell'articolo sui fondamenti dell'event-driven; perché lo stesso record può arrivare due volte a un gruppo e come difendersi è nell'articolo sull'idempotenza dei consumer.

import java.util.*;

public class GruppoConsumer {
    // Modello della assegnazione "a intervalli" (stessa idea del RangeAssignor di Kafka):
    // ogni consumatore riceve n/m partizioni consecutive, i primi n%m ne ricevono una in più.
    static Map<String, List<Integer>> assegna(int partizioni, List<String> membri) {
        Map<String, List<Integer>> mappa = new LinkedHashMap<>();
        int base = partizioni / membri.size(), extra = partizioni % membri.size(), p = 0;
        for (int i = 0; i < membri.size(); i++) {
            List<Integer> mie = new ArrayList<>();
            for (int k = 0; k < base + (i < extra ? 1 : 0); k++) mie.add(p++);
            mappa.put(membri.get(i), mie);
        }
        return mappa;
    }

    public static void main(String[] args) {
        for (int m : new int[] {1, 3, 4, 9}) {
            List<String> membri = new ArrayList<>();
            for (int i = 1; i <= m; i++) membri.add("c" + i);
            System.out.println(m + " consumatori su 6 partizioni -> " + assegna(6, membri));
        }
        // Due gruppi sullo stesso topic: ciascuno ha le sue assegnazioni e i suoi offset.
        System.out.println("gruppo inventario  -> " + assegna(6, List.of("i1", "i2")));
        System.out.println("gruppo fatturazione -> " + assegna(6, List.of("f1", "f2", "f3")));
    }
}

// Output:
// 1 consumatori su 6 partizioni -> {c1=[0, 1, 2, 3, 4, 5]}
// 3 consumatori su 6 partizioni -> {c1=[0, 1], c2=[2, 3], c3=[4, 5]}
// 4 consumatori su 6 partizioni -> {c1=[0, 1], c2=[2, 3], c3=[4], c4=[5]}
// 9 consumatori su 6 partizioni -> {c1=[0], c2=[1], c3=[2], c4=[3], c5=[4], c6=[5], c7=[], c8=[], c9=[]}
// gruppo inventario  -> {i1=[0, 1, 2], i2=[3, 4, 5]}
// gruppo fatturazione -> {f1=[0, 1], f2=[2, 3], f3=[4, 5]}

Con un consumatore le sei partizioni sono tutte sue; con tre ne toccano due a testa; con quattro la divisione non è esatta e i primi due ne prendono due, gli altri una; con nove, tre consumatori restano con la lista vuota. È la prova che aggiungere consumatori oltre il numero di partizioni non serve a nulla. L'ultima coppia di righe mostra due gruppi sullo stesso topic con assegnazioni indipendenti. Il modello copia la logica del RangeAssignor; ricorda però che nel protocollo classico l'elenco predefinito di partition.assignment.strategy è [RangeAssignor, CooperativeStickyAssignor] e vale il primo, mentre con group.protocol=consumer (il nuovo protocollo, KIP-848) l'assegnazione la decide il coordinatore sul broker. Che cosa succede quando un consumatore entra o esce, cioè il ribilanciamento, è il tema dell'articolo sui producer e consumer avanzati.

Retention e compaction: quanto resta nel log

I record non si cancellano al consumo, quindi qualcosa deve decidere quando sparire. Le impostazioni di topic (documentazione «Topic Configs», 4.4) sono due politiche:

• cleanup.policy=delete con retention.ms (default 604800000, cioè 7 giorni) e retention.bytes (default -1, nessun limite): sparisce il segmento più vecchio quando supera il tempo o la dimensione.
• cleanup.policy=compact: per ogni chiave resta solo l'ultimo record, con delete.retention.ms (default 1 giorno) che fissa quanto sopravvivono i tombstone, cioè i record con valore null che cancellano una chiave.

Le due politiche rispondono a domande diverse. La retention a tempo dice «tieni la storia per sette giorni» e va bene per flussi di eventi come i clic o gli ordini; la compaction dice «tieni lo stato attuale di ogni chiave» e va bene per anagrafiche, prezzi e configurazioni. Nota che la cancellazione agisce per segmento (default 1 GiB o 7 giorni, segment.bytes e segment.ms) e mai sul segmento attivo: un topic con pochi dati non vedrà sparire nulla finché il segmento non si chiude.

import java.util.*;

public class Compattazione {
    record Record(long offset, String chiave, String valore) {}

    // Modello della compaction: tiene l'ultimo record per ogni chiave, senza cambiare gli offset.
    static List<Record> compatta(List<Record> log) {
        Map<String, Long> ultimo = new HashMap<>();
        for (Record r : log) ultimo.put(r.chiave(), r.offset());
        List<Record> pulito = new ArrayList<>();
        for (Record r : log) if (ultimo.get(r.chiave()) == r.offset()) pulito.add(r);
        return pulito;
    }

    public static void main(String[] args) {
        List<Record> log = new ArrayList<>();
        String[][] scritture = {
            {"user123", "mario@example.com"}, {"user456", "anna@example.com"},
            {"user123", "mario.rossi@example.com"}, {"user456", null}};   // null = tombstone
        for (String[] s : scritture) log.add(new Record(log.size(), s[0], s[1]));

        System.out.println("prima:  " + log);
        System.out.println("dopo:   " + compatta(log));
    }
}

// Output:
// prima:  [Record[offset=0, chiave=user123, valore=mario@example.com], Record[offset=1, chiave=user456, valore=anna@example.com], Record[offset=2, chiave=user123, valore=mario.rossi@example.com], Record[offset=3, chiave=user456, valore=null]]
// dopo:   [Record[offset=2, chiave=user123, valore=mario.rossi@example.com], Record[offset=3, chiave=user456, valore=null]]

Il log iniziale ha quattro record su due chiavi. Dopo la compaction restano l'offset 2 per user123 (il valore aggiornato) e l'offset 3 per user456, che è un tombstone: il valore è null, ma il record resta nel log per delete.retention.ms, così i consumatori in ritardo fanno in tempo a vedere la cancellazione. Il modello conserva anche gli offset originali (2 e 3, non 0 e 1), come garantisce la documentazione: «the offset for a message never changes». Un broker vero compatta in background con i thread del log cleaner e mai il segmento attivo; il tuo consumatore che parte da zero ricostruisce lo stato leggendo un record per chiave.

Topic interni e tre comandi per guardare dentro

Kafka conserva il proprio stato in topic che si chiamano come i tuoi. __consumer_offsets contiene gli offset committed (50 partizioni per default, offsets.topic.num.partitions; il gruppo è assegnato a una di esse in base al suo group.id); __transaction_state contiene lo stato delle transazioni (50 partizioni, transaction.state.log.num.partitions); in modalità KRaft i metadati del cluster stanno nel log __cluster_metadata (la documentazione lo cita nei comandi kafka-dump-log.sh). Entrambi i parametri da 50 hanno la nota «should not change after deployment»: sono scelte da fare prima del primo avvio.

I comandi che seguono sono lo schema con cui creare un topic con sei partizioni e tre repliche, descriverlo per vedere leader, repliche e ISR di ogni partizione, e misurare il lag di un gruppo. Non li ho eseguiti, perché non c'è un cluster in questo ambiente, e non riporto un'uscita inventata: il formato dell'output cambia con la versione, e la colonna da guardare per prima è LAG.

# Schema da adattare: richiede un cluster raggiungibile in kafka1:9092
kafka-topics.sh --create \
  --bootstrap-server kafka1:9092 \
  --topic ordini-effettuati \
  --partitions 6 \
  --replication-factor 3 \
  --config min.insync.replicas=2

kafka-topics.sh --describe \
  --bootstrap-server kafka1:9092 \
  --topic ordini-effettuati

kafka-consumer-groups.sh \
  --bootstrap-server kafka1:9092 \
  --describe \
  --group servizio-inventario

Il --describe del topic è la prima diagnosi quando qualcosa non va: se per una partizione l'ISR è più corta delle repliche, un follower è rimasto indietro. Il --describe del gruppo mostra per ogni partizione l'offset corrente, il log end offset e la differenza. Se il lag cresce a ogni controllo il gruppo non regge il ritmo: aggiungi consumatori (fino al numero di partizioni) o rendi più veloce l'elaborazione. Se il gruppo risulta vuoto, non ha membri attivi.

Che cosa sai fare adesso

Sai spiegare perché Kafka non è una coda che si svuota e perché la partizione è l'unità di ordine e di parallelismo. Sai prevedere dove finisce un record con una chiave data e perché cambiare il numero di partizioni sposta le chiavi. Distingui LEO, high watermark, offset corrente e offset committed, e sai che quest'ultimo è il prossimo da leggere. Sai quanti consumatori hanno senso per un gruppo e che due gruppi sullo stesso topic non si disturbano. Il passo successivo è capire chi coordina il cluster ora che ZooKeeper non c'è più, nell'articolo su KRaft, e come si rendono affidabili la scrittura e la lettura, nell'articolo sulle transazioni. Se arrivi da un altro punto della serie, i collegamenti sono: Kafka Streams, Kafka Connect, il monitoraggio, Kafka in produzione.