09 - Monitoring Kafka: JMX Exporter, Prometheus e Grafana Dashboard

Come sai che il cluster sta male prima che te lo dicano gli utenti

Un broker lento non produce un errore: produce un ritardo. Una partizione con la replica indietro non fa suonare nulla finché non cade anche il leader. Un gruppo di consumatori che non regge il ritmo non si ferma: accumula lag, e dopo qualche ora i dati che legge sono vecchi di una giornata. Il monitoraggio di Kafka serve a trasformare questi guasti silenziosi in numeri con una soglia e una persona che riceve l'avviso. L'obiettivo di questo articolo è modesto e verificabile: sapere quali sono le poche metriche che meritano un allarme, da dove si leggono, e come si scrive la regola.

Una premessa sul metodo. Qui non c'è un cluster Kafka né un Prometheus: nessuna dashboard è stata vista. Ho eseguito invece due modelli in Java 21 (un finto exporter che applica regole a MBean veri della JVM, e un classificatore dell'andamento del lag) e ho controllato la sintassi YAML dei file di configurazione con un parser. I nomi delle metriche vengono dalla documentazione «Monitoring» di Kafka 4.4, le regole dell'exporter dalla documentazione di JMX Exporter 1.7.0 e gli allarmi da quella di Prometheus, consultate il 2026-11-12. Le cifre d'esempio nei listati sono inventate per la prova e lo dichiaro dove compaiono.

Le poche metriche che hanno una soglia

La documentazione di Kafka elenca centinaia di metriche. Per un allarme ne bastano poche, perché hanno un valore atteso che non dipende dal tuo carico:

• kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions: partizioni in cui l'ISR è più piccola dell'insieme delle repliche. Il valore atteso è 0.
• kafka.server:type=ReplicaManager,name=UnderMinIsrPartitionCount: partizioni in cui l'ISR è sotto min.insync.replicas. Atteso 0.
• kafka.controller:type=KafkaController,name=OfflinePartitionsCount: partizioni senza leader, così come le vede quel controller.
• kafka.controller:type=KafkaController,name=ActiveControllerCount: la documentazione dice che in tutto il cluster un solo nodo dovrebbe avere 1.
• Il lag dei gruppi di consumatori, che non è una metrica del broker (lo vediamo più sotto).

Il criterio che le tiene insieme è la domanda «quanto sono vicino a perdere o a non servire dati?». Le prime due misurano la perdita di margine di durabilità (la seconda è già un problema operativo: con acks=all le scritture su quelle partizioni falliscono, come visto nell'articolo sui producer avanzati). La terza misura il non servire. La quarta controlla che il cluster abbia un cervello solo, e nasce con KRaft e il suo quorum di controller. Tutte hanno un valore giusto, 0 o 1, quindi la soglia non richiede esperienza del carico: è lì che sono più utili. Il resto (latenza delle richieste, byte in ingresso, code interne) è contesto da mostrare in una dashboard e non da allarmare finché non hai un riferimento storico.

Come escono dal broker: JMX, e un exporter che le traduce

Kafka espone le metriche come MBean JMX, ciascuno con un nome nella forma dominio:type=...,name=.... La documentazione precisa che l'accesso JMX remoto è spento di default: si accende con la variabile JMX_PORT oppure con le proprietà standard della JVM, e va protetto in produzione. Prometheus non parla JMX, quindi serve un traduttore: il JMX Exporter, che per la sua stessa documentazione si usa di solito come Java agent dentro il processo del broker, evitando la configurazione di JMX remoto.

L'exporter funziona per regole. Ogni regola ha un pattern (un'espressione regolare), un name (il nome della metrica Prometheus), eventualmente labels, value e type. La documentazione descrive l'ingresso su cui si applica il pattern: dominio<proprietà1=valore1, proprietà2=valore2><chiavi>attributo: valore. Le regole si valutano in ordine, vince la prima che combacia, e gli attributi senza regola non vengono raccolti. Il listato riproduce questo meccanismo su MBean veri della JVM: registra quattro MBean con nomi identici a quelli di Kafka (i valori sono inventati), costruisce per ciascuno la stringa di ingresso e applica tre regole.

import javax.management.*;
import java.lang.management.ManagementFactory;
import java.util.*;
import java.util.regex.*;

public class RegoleJmx {
    public interface ValoreMBean { long getValue(); }               // un MBean con l'attributo "Value"
    public static class Valore implements ValoreMBean {
        final long v; Valore(long v) { this.v = v; }
        public long getValue() { return v; }
    }

    // Una regola come nella documentazione dell'exporter: pattern, name, labels (con gruppi di cattura).
    record Regola(Pattern pattern, String name, Map<String, String> labels) {}

    public static void main(String[] args) throws Exception {
        MBeanServer server = ManagementFactory.getPlatformMBeanServer();
        // MBean con gli stessi nomi di quelli documentati da Kafka; i valori sono inventati per la prova.
        server.registerMBean(new Valore(0), new ObjectName("kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions"));
        server.registerMBean(new Valore(1), new ObjectName("kafka.controller:type=KafkaController,name=ActiveControllerCount"));
        server.registerMBean(new Valore(12), new ObjectName("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec,topic=ordini"));
        server.registerMBean(new Valore(7), new ObjectName("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec,topic=click"));

        List<Regola> regole = List.of(
            new Regola(Pattern.compile("kafka.server<type=ReplicaManager, name=UnderReplicatedPartitions><>Value: (.*)"),
                "kafka_server_replicamanager_underreplicatedpartitions", Map.of()),
            new Regola(Pattern.compile("kafka.controller<type=KafkaController, name=ActiveControllerCount><>Value: (.*)"),
                "kafka_controller_activecontrollercount", Map.of()),
            new Regola(Pattern.compile("kafka.server<type=BrokerTopicMetrics, name=MessagesInPerSec, topic=(.+)><>Value: (.*)"),
                "kafka_server_messages_in", Map.of("topic", "$1")));

        for (ObjectName on : new TreeSet<>(server.queryNames(new ObjectName("kafka.*:*"), null))) {
            // La forma documentata dell'ingresso: domain<proprietà><>attributo: valore
            StringBuilder props = new StringBuilder();
            for (String p : on.getKeyPropertyListString().split(",")) props.append(props.length() == 0 ? "" : ", ").append(p);
            String ingresso = on.getDomain() + "<" + props + "><>Value: " + server.getAttribute(on, "Value");
            for (Regola r : regole) {
                Matcher m = r.pattern().matcher(ingresso);
                if (!m.matches()) continue;
                StringBuilder riga = new StringBuilder(r.name());
                if (!r.labels().isEmpty()) {
                    riga.append("{");
                    r.labels().forEach((k, v) -> riga.append(k).append("=\"").append(v.replace("$1", m.group(1))).append("\""));
                    riga.append("}");
                }
                System.out.println(riga + " " + m.group(m.groupCount()));
                break;                                               // la prima regola che combacia vince
            }
        }
    }
}

// Output:
// kafka_controller_activecontrollercount 1
// kafka_server_messages_in{topic="click"} 7
// kafka_server_messages_in{topic="ordini"} 12
// kafka_server_replicamanager_underreplicatedpartitions 0

Il risultato è una piccola uscita in stile Prometheus: il numero di controller attivi, le partizioni sotto-replicate e il tasso di messaggi in ingresso con l'etichetta topic estratta dal gruppo di cattura. Le due righe kafka_server_messages_in hanno etichette diverse perché il pattern cattura il nome del topic, e il valore è quello del gruppo dopo di esso. Le tre cose da portarsi via sono nel codice. Il pattern termina con Value: (.*) perché l'ingresso contiene anche il valore. Una proprietà con nome variabile, come topic=(.+), diventa un'etichetta e non una metrica distinta. E break dopo la prima corrispondenza replica la regola «vince la prima». Il modello non è l'exporter: non ha cache, non gestisce dati compositi e non conosce i tipi. Per questo la configurazione vera che trovi sotto è uno schema, validato nella sintassi e non eseguito.

# Schema da adattare (non eseguito): regole dell'exporter nel formato della documentazione JMX Exporter 1.7.0.
# Le regole sono valutate in ordine e si ferma alla prima che combacia; gli attributi senza regola non vengono raccolti.
rules:
  - pattern: 'kafka.server<type=ReplicaManager, name=UnderReplicatedPartitions><>Value: (.*)'
    name: kafka_server_replicamanager_underreplicatedpartitions
    value: '$1'
    type: GAUGE
  - pattern: 'kafka.server<type=ReplicaManager, name=UnderMinIsrPartitionCount><>Value: (.*)'
    name: kafka_server_replicamanager_underminisrpartitioncount
    value: '$1'
    type: GAUGE
  - pattern: 'kafka.controller<type=KafkaController, name=OfflinePartitionsCount><>Value: (.*)'
    name: kafka_controller_offlinepartitionscount
    value: '$1'
    type: GAUGE
  - pattern: 'kafka.controller<type=KafkaController, name=ActiveControllerCount><>Value: (.*)'
    name: kafka_controller_activecontrollercount
    value: '$1'
    type: GAUGE
  - pattern: 'kafka.server<type=BrokerTopicMetrics, name=MessagesInPerSec, topic=(.+)><>OneMinuteRate: (.*)'
    name: kafka_server_messagesinpersec_oneminuterate
    value: '$2'
    labels:
      topic: '$1'
    type: GAUGE

Si attiva aggiungendo il Java agent al broker, per esempio con KAFKA_OPTS="-javaagent:/opt/kafka/jmx_prometheus_javaagent-1.7.0.jar=9404:/opt/kafka/exporter.yml", secondo il formato della guida rapida dell'exporter (porta, due punti, file di configurazione). La stessa guida dice che senza un host indicato il server HTTP si lega a tutte le interfacce: se i broker sono su una rete aperta, limita la porta con il firewall o lega l'exporter a un'interfaccia interna. La verifica è un curl sulla porta, alla ricerca del nome della metrica.

Il lag: un numero che va letto in movimento

Il lag di un gruppo è la differenza tra il log end offset di ogni partizione e l'offset committed del gruppo (la distinzione vista nell'articolo sui fondamenti). Non è tra le metriche dei broker che hai appena mappato. Lo vedi con kafka-consumer-groups.sh --describe --group, lo leggi lato client con la metrica records-lag-max di kafka.consumer:type=consumer-fetch-manager-metrics, oppure con un exporter che interroga i gruppi; di quest'ultimo ne esistono più d'uno e non ne raccomando uno, perché non ne ho verificato la manutenzione.

Qualunque sia la fonte, l'errore da evitare è un allarme su una soglia assoluta («più di 50.000 record»). Un lag di 100 può essere il normale ritardo di un gruppo che segue da vicino, e un lag di 100.000 può essere un gruppo che sta recuperando dopo una fermata. Quello che conta è la direzione, in rapporto al ritmo di scrittura. Il modello confronta tre gruppi sullo stesso intervallo: scritti e letti sono contatori cumulativi campionati ogni 30 secondi, con numeri inventati.

import java.util.*;

public class LagTrend {
    // Ogni campione: record scritti e record consumati (cumulativi) a intervalli di 30 secondi.
    static void analizza(String nome, long[] scritti, long[] consumati) {
        long[] lag = new long[scritti.length];
        for (int i = 0; i < lag.length; i++) lag[i] = scritti[i] - consumati[i];       // LEO - offset committed
        long inizio = lag[0], fine = lag[lag.length - 1];
        double prodRate = (scritti[scritti.length - 1] - scritti[0]) / (30.0 * (lag.length - 1));
        double consRate = (consumati[consumati.length - 1] - consumati[0]) / (30.0 * (lag.length - 1));
        String verdetto;
        if (fine > inizio) verdetto = "CRESCE: il gruppo non regge il ritmo";
        else if (fine < inizio) verdetto = "SCENDE: sta recuperando, " + (consRate > prodRate
            ? "svuotato in circa " + Math.round(fine / (consRate - prodRate)) + " s" : "");
        else verdetto = "STABILE";
        System.out.printf(Locale.ROOT, "%-26s lag %s -> %s | scritti %.1f/s, letti %.1f/s | %s%n",
            nome, inizio, fine, prodRate, consRate, verdetto);
    }

    public static void main(String[] args) {
        // A e C finiscono con lo stesso lag (100), ma uno è stabile e l'altro sta scendendo; B ha un lag che cresce.
        analizza("A: ritmo uguale", new long[] {1000, 4000, 7000, 10000}, new long[] {900, 3900, 6900, 9900});
        analizza("B: consumo troppo lento", new long[] {1000, 4000, 7000, 10000}, new long[] {900, 2900, 4900, 6900});
        analizza("C: sta recuperando", new long[] {1000, 4000, 7000, 10000}, new long[] {200, 3700, 7200, 10000 - 100});
    }
}

// Output:
// A: ritmo uguale            lag 100 -> 100 | scritti 100.0/s, letti 100.0/s | STABILE
// B: consumo troppo lento    lag 100 -> 3100 | scritti 100.0/s, letti 66.7/s | CRESCE: il gruppo non regge il ritmo
// C: sta recuperando         lag 800 -> 100 | scritti 100.0/s, letti 107.8/s | SCENDE: sta recuperando, svuotato in circa 13 s

Il gruppo A e il gruppo C finiscono entrambi con lag 100, ma A è stabile e C sta scendendo: leggendo 107,8 record al secondo contro 100 scritti, svuoterà il residuo in circa 13 secondi. Il gruppo B ha un lag di partenza identico ad A, ma legge 66,7 record al secondo contro 100 scritti e arriva a 3100: ogni 30 secondi accumula 1000 record in più, e non si fermerà da solo. Se vuoi un allarme sul lag, fallo sulla tendenza (il lag cresce da N minuti) e, per il livello assoluto, su una soglia legata a quanto tempo di dati puoi permetterti di non aver letto: la retention del topic decide il punto oltre il quale i record vengono cancellati prima di essere letti.

Gli allarmi: la sintassi di Prometheus e le soglie

Una regola di allarme Prometheus ha un nome, un'espressione expr, una durata for e, a scelta, labels e annotations. La documentazione spiega che for impone di aspettare che l'espressione resti vera per quel tempo prima che l'allarme scatti: nel frattempo è in stato pending. Nelle annotazioni si usano {{ $labels.nome }} per i valori delle etichette e {{ $value }} per il valore dell'espressione. Le regole qui sotto coprono le cinque condizioni dell'inizio, con le sole soglie che non dipendono dal carico.

# Schema da adattare (non eseguito): regole di allarme nel formato della documentazione Prometheus.
groups:
  - name: kafka.critical
    rules:
      - alert: KafkaBrokerDown
        expr: up{job="kafka-brokers"} == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Broker {{ $labels.instance }} non raggiungibile"
      - alert: KafkaOfflinePartitions
        expr: sum(kafka_controller_offlinepartitionscount) > 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "{{ $value }} partizioni senza leader"
      - alert: KafkaControllerAttivi
        expr: sum(kafka_controller_activecontrollercount) != 1
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "I controller attivi sono {{ $value }}, dovrebbe essere 1"
  - name: kafka.warning
    rules:
      - alert: KafkaUnderReplicated
        expr: sum(kafka_server_replicamanager_underreplicatedpartitions) > 0
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "{{ $value }} partizioni con ISR incompleta da più di 5 minuti"
      - alert: KafkaSottoMinIsr
        expr: sum(kafka_server_replicamanager_underminisrpartitioncount) > 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Partizioni sotto min.insync.replicas: le scritture con acks=all falliscono"

Guarda come è scelto for. Un controller attivo diverso da uno e una partizione offline sono condizioni da pochi minuti al massimo, perché durante una normale elezione possono durare qualche secondo: for: 1m o 2m evita falsi allarmi ma non ritarda l'intervento. Le partizioni sotto-replicate sono un avviso, non un'emergenza, finché non scendono sotto min.insync.replicas: per questo la regola ha for: 5m e gravità warning, mentre quella su UnderMinIsr è critical ed è breve. I tempi vanno tarati sui tuoi: sono un punto di partenza e non un valore consigliato dalla documentazione. Per verificare che la regola scatti nel caso in cui serve, la documentazione di Prometheus prevede test unitari delle regole con promtool, che qui non ho potuto eseguire perché non è installato.

Dashboard: i pannelli, non gli identificatori

Sulle dashboard Grafana pronte da importare non scrivo identificatori numerici: non li ho verificati e un elenco di ID non controllati è esattamente il tipo di dato che invecchia in silenzio. Costruisci invece, partendo dalle metriche che già raccogli, sei pannelli: stato di ogni broker (up), partizioni sotto-replicate e sotto min-ISR, partizioni offline, controller attivi, tasso di messaggi in ingresso per topic, e il lag per gruppo con la tendenza. Tutti rispondono a una domanda a cui dovresti saper rispondere in dieci secondi: il cluster serve? sta perdendo margine? i consumatori seguono?

Che cosa sai fare adesso

Sai scegliere le cinque metriche che meritano un allarme e il perché di ciascuna. Sai da dove le espone Kafka (MBean JMX) e come un exporter le traduce con regole a pattern, nell'ordine in cui vengono valutate. Sai che il lag si legge in movimento e non in assoluto, e come si calcola il tempo di svuotamento. Sai scrivere regole di allarme con for sensato e annotazioni che dicono il valore. Il passo dopo il monitoraggio è dimensionare il cluster in modo che gli allarmi suonino raramente, nell'articolo su sizing e disaster recovery; per decidere che cosa fare dei messaggi che non si riescono a elaborare, nell'articolo sulla dead letter queue. Se arrivi da un altro punto della serie, i collegamenti sono: Flink e Iceberg.