11 - Kafka in Produzione: Sizing, Retention, Replica e Disaster Recovery

Dai tre broker di prova a un cluster che deve restare in piedi

Un cluster di prova con tre broker, valori predefiniti e nessuna retention pensata sembra funzionare, finché non arrivano il carico vero, un disco pieno o un riavvio fatto di fretta. Metterlo in produzione significa rispondere a domande concrete: quanti broker e quanto disco servono per il mio flusso, che cosa cambio rispetto ai default, come aggiorno il cluster senza fermare le scritture, come ricostruisco il servizio se perdo un intero datacenter. Questo articolo risponde con il metodo, non con numeri universali: nessuno dei valori «tipici» di throughput per broker che circolano ha una fonte, e per questo non li uso. Ti do invece i conti da fare con i tuoi dati e le impostazioni verificate nella documentazione di Kafka 4.4 (consultata il 2026-11-12).

Il materiale eseguito è di due tipi. Un calcolo in Java 21 che dimensiona rete, disco e memoria da un'ipotesi di carico (le cifre sono ipotesi dichiarate, non misure) e un modello che mostra perché il rolling restart ha una regola d'oro. Tutto ciò che richiede un cluster, compreso MirrorMaker 2, è uno schema da adattare e non è stato eseguito.

Il dimensionamento è aritmetica sul tuo flusso

Parti da tre numeri che solo tu conosci: quanti MB al secondo scrivono i produttori, per quanti giorni vuoi conservare i dati, quanti gruppi di consumatori leggono tutto. Da lì derivano le grandezze che contano. Ogni byte scritto da un client viene scritto replication factor volte nel cluster, una dal client e le altre dalla replica tra broker: il traffico in ingresso del cluster è scrittura × RF. In uscita escono le repliche (scrittura × (RF − 1)) più la lettura di ogni gruppo (scrittura × gruppi). Il disco è scrittura × secondi di retention × RF, più uno spazio libero che non vuoi esaurire. E per la memoria la documentazione offre una stima esplicita: per bufferizzare 30 secondi di attività, write_throughput × 30, perché Kafka si appoggia alla cache di pagine del sistema operativo.

import java.util.Locale;

public class Dimensionamento {
    public static void main(String[] args) {
        // IPOTESI da sostituire con i tuoi valori: nessuno di questi numeri è una misura.
        double scritturaMBs = 50;          // flusso di scrittura dei client, in MB/s
        int replicationFactor = 3;
        int gruppiConsumatori = 2;         // ogni gruppo legge tutto il flusso
        int giorniRetention = 7;
        double margine = 0.30;             // spazio libero da tenere, 30%
        int broker = 3;

        double ingressoCluster = scritturaMBs * replicationFactor;                          // client + repliche
        double uscitaCluster = scritturaMBs * (replicationFactor - 1) + scritturaMBs * gruppiConsumatori; // repliche + letture
        double discoTotaleTB = scritturaMBs * giorniRetention * 86_400 * replicationFactor / 1_000_000.0; // MB -> TB (decimali)
        double discoConMargineTB = discoTotaleTB / (1 - margine);
        double cache30s = scritturaMBs * 30 / 1000.0;                                       // GB: write_throughput * 30 s

        System.out.printf(Locale.ROOT, "rete in ingresso, cluster:   %.0f MB/s  (%.0f MB/s per broker con %d broker)%n",
            ingressoCluster, ingressoCluster / broker, broker);
        System.out.printf(Locale.ROOT, "rete in uscita, cluster:     %.0f MB/s  (%.0f MB/s per broker)%n",
            uscitaCluster, uscitaCluster / broker);
        System.out.printf(Locale.ROOT, "disco totale (RF %d, %d giorni): %.2f TB, con margine %.2f TB, per broker %.2f TB%n",
            replicationFactor, giorniRetention, discoTotaleTB, discoConMargineTB, discoConMargineTB / broker);
        System.out.printf(Locale.ROOT, "memoria per 30 s di scritture: %.1f GB%n", cache30s);

        // Controllo aritmetico con altri dati: 100 MB/s, 7 giorni, RF 3
        double vecchio = 100.0 * 7 * 86_400 * 3 / 1_000_000.0;
        System.out.printf(Locale.ROOT, "controllo: 100 MB/s x 7 giorni x RF 3 = %.2f TB (decimali), %.1f TB per 3 broker%n", vecchio, vecchio / 3);
    }
}

// Output:
// rete in ingresso, cluster:   150 MB/s  (50 MB/s per broker con 3 broker)
// rete in uscita, cluster:     200 MB/s  (67 MB/s per broker)
// disco totale (RF 3, 7 giorni): 90.72 TB, con margine 129.60 TB, per broker 43.20 TB
// memoria per 30 s di scritture: 1.5 GB
// controllo: 100 MB/s x 7 giorni x RF 3 = 181.44 TB (decimali), 60.5 TB per 3 broker

Con quell'ipotesi, tre broker vedono 50 MB/s in ingresso e circa 67 MB/s in uscita ciascuno, servono 90,72 TB di disco per sette giorni a fattore di replica 3, che salgono a 129,60 TB con il 30% di margine, cioè 43,20 TB per broker, e 1,5 GB di memoria per tenere calda mezza minuto di scrittura. Il numero da confrontare con la capacità reale della rete e dei dischi che hai è l'ultimo tra i conti di rete: se il traffico per broker si avvicina a ciò che la scheda di rete o i dischi sanno reggere (una misura che devi fare tu sul tuo hardware), servono più broker. L'ultima riga controlla l'aritmetica con altri dati: 100 MB/s per sette giorni a RF 3 sono 181,44 TB decimali, e quindi 60,5 TB per broker su tre. Se leggi altrove 173 TB per lo stesso caso, è una conversione sbagliata o in tebibyte.

Per i dischi la documentazione aggiunge che il throughput dei dischi è «generalmente il collo di bottiglia», e che avere più dischi è meglio. Per le partizioni vale la regola vista nell'articolo sui fondamenti: serve abbastanza parallelismo per i consumatori e abbastanza distribuzione del carico, tenendo conto che ogni partizione deve stare per intero su un solo server. Un limite ufficiale c'è sui nomi: la documentazione assume che le partizioni non superino 100.000 per topic, e per questo i nomi dei topic non possono superare 249 caratteri. Non ci sono invece, nella documentazione, cifre del tipo «mai più di N partizioni per broker».

Configurazione del broker: cambia poco, e sai perché

Molte guide propongono lunghi server.properties «ottimizzati». I default di Kafka 4.4 sono più ragionevoli di quanto si pensi, e conviene cambiarli solo con una ragione misurata. Leggili prima di toccarli:

• num.network.threads vale 3 e num.io.threads vale 8; si alzano quando i thread risultano saturi, non per abitudine.
• log.flush.interval.messages ha come default il massimo di un long e log.flush.interval.ms non è impostato: Kafka lascia il flush al sistema operativo, e la durabilità la danno le repliche con acks=all, non il flush forzato.
• min.insync.replicas vale 1 e offsets.topic.replication.factor vale 3: il primo va portato a 2 per i topic che contano, come nell'articolo sui producer.
• controlled.shutdown.enable e auto.leader.rebalance.enable valgono entrambi true: lo spegnimento ordinato e il ritorno della leadership alle repliche preferite sono già attivi.
• auto.create.topics.enable vale true: in produzione, un refuso nel nome di un topic ne crea uno nuovo senza che nessuno se ne accorga, quindi di solito si disattiva.

L'elenco dice una cosa utile: delle cinque voci, solo due (min.insync.replicas e auto.create.topics.enable) sono quasi sempre da cambiare in produzione, e riguardano la sicurezza dei dati, non la velocità. Le altre si toccano quando il monitoraggio (vedi l'articolo su JMX e Prometheus) mostra un problema. Per i controller KRaft la documentazione stima che 5 GB di memoria e 5 GB di disco per il log dei metadati siano sufficienti per un cluster tipico (vedi l'articolo su KRaft). Per la JVM e il filesystem la pagina «Hardware and OS» dà indicazioni su limiti di file descriptor, flush del sistema operativo e scelta tra XFS ed ext4: leggila per il tuo sistema, senza copiare valori da un'altra installazione.

Un topic di dominio, come si crea

I valori che definiscono un topic sono pochi: partizioni, fattore di replica, min.insync.replicas e retention. Il blocco sotto è la creazione di un topic di pagamenti con le scelte discusse; la retention a 30 giorni è una scelta del dominio (quanto tempo ti serve per riprocessare o per un'indagine), non un default. Il secondo comando mostra l'aumento delle partizioni, con la precisazione della documentazione operativa: se i dati sono partizionati con hash(chiave) modulo il numero di partizioni, aumentarle cambia la mappatura e i record con la stessa chiave possono finire in partizioni diverse; inoltre, i consumatori con auto.offset.reset=latest possono perdere record prodotti sulle nuove partizioni prima di scoprirle, e la documentazione raccomanda by_duration:<durata>.

# Schema da adattare (non eseguito: serve un cluster raggiungibile in kafka1:9092).
# Un topic di dominio: tre repliche, due devono essere in sincrono per accettare una scrittura con acks=all.
bin/kafka-topics.sh --bootstrap-server kafka1:9092 --create --topic pagamenti-confermati \
  --partitions 12 --replication-factor 3 \
  --config min.insync.replicas=2 \
  --config retention.ms=2592000000          # 30 giorni, scelta del dominio

# Aumentare le partizioni in seguito è possibile, ma sposta le chiavi (vedi l'articolo sui fondamenti)
bin/kafka-topics.sh --bootstrap-server kafka1:9092 --alter --topic pagamenti-confermati --partitions 24

# Riportare la leadership sulle repliche preferite dopo i riavvii (solo se auto.leader.rebalance.enable=false)
bin/kafka-leader-election.sh --bootstrap-server kafka1:9092 --election-type preferred --all-topic-partitions

Il terzo comando serve solo se hai spento auto.leader.rebalance.enable: con il default true il cluster riporta da solo la leadership sulle repliche preferite, il comportamento descritto nella sezione «Balancing leadership». Non mettere mai un tipo di elezione unclean in una procedura di riavvio: l'elezione unclean permette a una replica fuori dall'ISR di diventare leader, e la documentazione del parametro unclean.leader.election.enable (default false) dice che può causare perdita di dati.

Riavviare i broker senza fermare le scritture

Aggiornare la versione, cambiare una configurazione o patchare il sistema richiede di riavviare i broker. La procedura documentata per l'aggiornamento è: un broker alla volta, fermalo, aggiorna, riavvialo, e verifica che il cluster si comporti come atteso prima del successivo. Lo spegnimento ordinato (controlled.shutdown.enable è true di default) sincronizza i log su disco per evitare la ricostruzione al riavvio e sposta la leadership su altre repliche prima di fermarsi, portando l'indisponibilità di ogni partizione a pochi millisecondi, come dice la documentazione. La regola d'oro è quella che tutti scrivono e pochi capiscono: non riavviare il broker successivo finché il precedente non è rientrato nell'ISR.

import java.util.*;

public class RollingRestart {
    enum Stato { ATTIVO_IN_ISR, SPENTO, IN_RECUPERO }              // IN_RECUPERO: è tornato ma non ha ancora ricopiato il log

    static final int MIN_ISR = 2;

    static long isr(Stato[] broker) { return Arrays.stream(broker).filter(s -> s == Stato.ATTIVO_IN_ISR).count(); }

    static void stampa(String evento, Stato[] b) {
        long n = isr(b);
        System.out.printf("  %-34s ISR=%d  scritture acks=all: %s%n", evento, n, n >= MIN_ISR ? "ok" : "RIFIUTATE");
    }

    static void sequenza(String titolo, boolean aspetta) {
        System.out.println(titolo);
        Stato[] b = {Stato.ATTIVO_IN_ISR, Stato.ATTIVO_IN_ISR, Stato.ATTIVO_IN_ISR};   // RF=3, min.insync.replicas=2
        stampa("cluster sano", b);
        for (int i = 0; i < 2; i++) {
            b[i] = Stato.SPENTO;          stampa("broker " + i + " spento", b);
            b[i] = Stato.IN_RECUPERO;     stampa("broker " + i + " riavviato, copia il log", b);
            if (aspetta) { b[i] = Stato.ATTIVO_IN_ISR; stampa("broker " + i + " rientrato nell'ISR", b); }
        }
    }

    public static void main(String[] args) {
        sequenza("riavvio che aspetta l'ISR completa:", true);
        sequenza("riavvio senza aspettare:", false);
    }
}

// Output:
// riavvio che aspetta l'ISR completa:
//   cluster sano                       ISR=3  scritture acks=all: ok
//   broker 0 spento                    ISR=2  scritture acks=all: ok
//   broker 0 riavviato, copia il log   ISR=2  scritture acks=all: ok
//   broker 0 rientrato nell'ISR        ISR=3  scritture acks=all: ok
//   broker 1 spento                    ISR=2  scritture acks=all: ok
//   broker 1 riavviato, copia il log   ISR=2  scritture acks=all: ok
//   broker 1 rientrato nell'ISR        ISR=3  scritture acks=all: ok
// riavvio senza aspettare:
//   cluster sano                       ISR=3  scritture acks=all: ok
//   broker 0 spento                    ISR=2  scritture acks=all: ok
//   broker 0 riavviato, copia il log   ISR=2  scritture acks=all: ok
//   broker 1 spento                    ISR=1  scritture acks=all: RIFIUTATE
//   broker 1 riavviato, copia il log   ISR=1  scritture acks=all: RIFIUTATE

Nella sequenza ordinata, quando il broker 0 è stato riavviato non conta ancora nell'ISR: il cluster resta a due repliche in sincrono, il minimo, e continua ad accettare scritture; solo dopo il rientro si passa al broker 1. Nella sequenza frettolosa il broker 0 è tornato ma sta ancora copiando il log; fermare il 1 lascia un'unica replica in sincrono, sotto il minimo, e tutte le scritture con acks=all vengono rifiutate. È l'errore che la regola previene, e lo vedi nel monitoraggio come UnderMinIsrPartitionCount sopra zero. Nel modello gli «stati» sono tre e il passaggio da IN_RECUPERO a rientrato è istantaneo se aspetti; in un cluster vero dipende dalla quantità di dati da ricopiare, ed è per questo che non esiste un'attesa fissa corretta: si aspetta il segnale, non il tempo.

Disaster recovery: MirrorMaker 2

Le repliche dentro un cluster proteggono da un guasto di broker, non dalla perdita di un datacenter. Per quello serve un secondo cluster in un'altra sede e un processo che copia i dati: MirrorMaker 2, costruito su Kafka Connect, descritto dalla documentazione come strumento di geo-replica. Si dichiarano i cluster, i flussi di replica (primary->secondary.enabled), i topic da copiare (espressioni regolari) e i gruppi di consumatori di cui replicare lo stato. I topic copiati cambiano nome: per default, con la DefaultReplicationPolicy, il topic ordini-effettuati del cluster primary diventa primary.ordini-effettuati nel secondario, per non mescolare record di origini diverse.

# Schema da adattare (non eseguito): replica attivo/passivo, dall'esempio della documentazione Kafka 4.4.
clusters = primary, secondary
primary.bootstrap.servers = broker3-primary:9092
secondary.bootstrap.servers = broker5-secondary:9092

# Il flusso va in una direzione sola: dal primario al secondario
primary->secondary.enabled = true
primary->secondary.topics = ordini-.*, pagamenti-.*
primary->secondary.groups = servizio-.*

# Lanciato nel datacenter del secondario ("consuma dal remoto, produci in locale"):
#   bin/connect-mirror-maker.sh connect-mirror-maker.properties --clusters secondary

Tre dettagli, tutti dalla documentazione. Primo, la raccomandazione «consuma dal remoto, produci in locale»: il processo va lanciato vicino al cluster di destinazione (--clusters secondary), perché i producer soffrono le reti lente più dei consumatori. Secondo, i processi MirrorMaker condividono la configurazione tramite il cluster di destinazione, quindi due processi con elenchi di topic diversi verso lo stesso cluster entrano in conflitto: tieni un solo file di configurazione. Terzo, l'exactly-once è supportato per i cluster MirrorMaker dedicati dalla versione 3.5.0, ma va abilitato (exactly.once.source.support) e richiede la comunicazione interna tra i nodi.

Resta il problema più delicato: gli offset. Un record ha un offset nel primario e un altro nel secondario, quindi un consumatore che passa al secondario non può riusare i propri offset. Il modo per tradurli è nella KIP-382: la classe RemoteClusterUtils espone translateOffsets(properties, targetClusterAlias, consumerGroupId, timeout), che trova gli offset locali corrispondenti all'ultimo checkpoint del gruppo di origine, supponendo la DefaultReplicationPolicy. «Ultimo checkpoint» vuol dire che la traduzione ha la risoluzione dei checkpoint: dopo un failover alcuni record possono essere riletti, quindi il consumatore deve tollerare i duplicati (vedi l'articolo sull'idempotenza). Non ho eseguito un failover: la procedura va provata su due cluster di prova prima di fidarsi.

Che cosa sai fare adesso

Sai calcolare rete, disco e memoria di un cluster dai tuoi dati, e riconosci un conto sbagliato per un fattore di unità. Sai quali impostazioni del broker conviene cambiare davvero e quali sono già al valore giusto. Sai creare un topic con min.insync.replicas e retention scelte per ragioni di dominio, e che aumentare le partizioni sposta le chiavi. Sai riavviare i broker uno alla volta aspettando l'ISR, e perché mai un'elezione unclean. Sai che MirrorMaker 2 copia i topic con un prefisso e traduce gli offset con i checkpoint, ma sai anche che un failover si prova prima. Se un messaggio non si lascia elaborare dopo tutto questo, la risposta è la dead letter queue.