Guida agli strumenti da riga di comando di Kafka: Riferimento CLI

Gli strumenti da riga di comando di Kafka sono il modo più rapido per rispondere a domande operative di base: questo topic esiste, quale broker guida questa partizione, cosa c'è dentro il topic, perché questo gruppo di consumatori è in ritardo, e questo client può autenticarsi con il cluster? Non servono per ogni attività, e la maggior parte delle modifiche in produzione dovrebbe comunque passare attraverso l'automazione, ma durante un deploy interrotto o una domanda notturna sui dati, la CLI è spesso la via più breve per ottenere i fatti.

Gli esempi seguenti presuppongono che gli script siano nel tuo PATH. In molte installazioni si trovano nella directory bin/ di Kafka, quindi lo stesso comando può essere eseguito come bin/kafka-topics.sh. Per i cluster protetti, la maggior parte dei comandi necessita anche di --command-config client.properties, dove quel file contiene le impostazioni SASL, SSL e altre impostazioni del client.

Strumenti CLI principali di Kafka

Le distribuzioni di Kafka includono tipicamente una directory bin/ contenente vari script ed eseguibili. Ci concentreremo su quelli più usati per gestire Kafka in modo efficace.

1. kafka-topics.sh

Questo è probabilmente lo strumento da riga di comando più usato. Permette di creare, elencare, descrivere, eliminare, modificare e gestire i topic di Kafka. La gestione dei topic è fondamentale per organizzare i flussi di dati all'interno di Kafka.

Sottocomandi e argomenti comuni:

• --create: Crea un nuovo topic.
• --list: Elenca tutti i topic nel cluster.
• --describe: Fornisce informazioni dettagliate su topic specifici.
• --delete: Elimina uno o più topic.
• --alter: Modifica la configurazione di un topic esistente.
• --topic <nome_topic>: Specifica il nome del topic.
• --partitions <num_partizioni>: Imposta il numero di partizioni per un topic (usato con --create).
• --replication-factor <fattore>: Imposta il fattore di replica per un topic (usato con --create).
• --bootstrap-server <host:porta>: Specifica il broker Kafka a cui connettersi.

Esempi:

• Crea un topic chiamato my_topic con 3 partizioni e un fattore di replica di 2:

  kafka-topics.sh --create --topic my_topic --partitions 3 --replication-factor 2 --bootstrap-server kafka-broker-1:9092,kafka-broker-2:9092
  

• Elenca tutti i topic nel cluster:

  kafka-topics.sh --list --bootstrap-server kafka-broker-1:9092
  

• Descrivi un topic chiamato my_topic:

  kafka-topics.sh --describe --topic my_topic --bootstrap-server kafka-broker-1:9092
  

  Questo mostrerà dettagli come partizioni, leader, repliche e ISR (In-Sync Replicas).

• Elimina un topic chiamato old_topic:

  kafka-topics.sh --delete --topic old_topic --bootstrap-server kafka-broker-1:9092
  

  Nota: L'eliminazione dei topic deve essere abilitata nelle configurazioni del broker Kafka (delete.topic.enable=true).

2. kafka-console-producer.sh

Questo strumento permette di inviare messaggi a un topic Kafka dall'input standard. È prezioso per testare i producer, iniettare dati di esempio o pubblicare manualmente messaggi.

Argomenti comuni:

• --topic <nome_topic>: Specifica il topic di destinazione.
• --bootstrap-server <host:porta>: Specifica il broker Kafka a cui connettersi.
• --property <chiave>=<valore>: Permette di impostare proprietà del producer (es. key.serializer, value.serializer).
• --producer-property <chiave>=<valore>: Simile a --property, ma specifico per configurazioni lato producer.

Esempi:

• Invia messaggi a my_topic:

  kafka-console-producer.sh --topic my_topic --bootstrap-server kafka-broker-1:9092
  

  Dopo aver eseguito questo comando, puoi digitare i messaggi riga per riga. Premi Ctrl+C per uscire.

• Invia messaggi con chiavi (formato JSON):

  kafka-console-producer.sh --topic my_topic --bootstrap-server kafka-broker-1:9092 --property parse.key=true --property key.separator=':'
  

  Ora puoi digitare coppie chiave:valore e Kafka le invierà con la chiave specificata.

3. kafka-console-consumer.sh

Questo strumento si sottoscrive a uno o più topic Kafka e stampa i messaggi ricevuti sull'output standard. È essenziale per testare i consumer, ispezionare i dati nei topic e fare debug di applicazioni producer/consumer.

Argomenti comuni:

• --topic <nome_topic>: Specifica il/i topic da cui consumare.
• --bootstrap-server <host:porta>: Specifica il broker Kafka a cui connettersi.
• --group-id <id_gruppo>: Specifica l'ID del gruppo di consumatori. È importante per gestire gli offset e permettere a più consumer di condividere il carico di consumo.
• --from-beginning: Legge i messaggi dall'inizio del log del topic.
• --offset <offset>: Inizia a consumare da un offset specifico.
• --partition <id_partizione>: Consuma da una partizione specifica.
• --property <chiave>=<valore>: Permette di impostare proprietà del consumer (es. value.deserializer).

Esempi:

• Consuma tutti i messaggi da my_topic:

  kafka-console-consumer.sh --topic my_topic --bootstrap-server kafka-broker-1:9092
  

• Consuma messaggi dall'inizio di my_topic per il gruppo di consumatori my_group:

  kafka-console-consumer.sh --topic my_topic --group-id my_group --from-beginning --bootstrap-server kafka-broker-1:9092
  

• Consuma messaggi con offset e chiavi stampati:

  kafka-console-consumer.sh --topic my_topic --bootstrap-server kafka-broker-1:9092 --property print.key=true --property key.separator="\t" --property print.offset=true --property print.headers=true
  

4. kafka-consumer-groups.sh

Questo strumento viene utilizzato per gestire e ispezionare i gruppi di consumatori. È fondamentale per comprendere il ritardo dei consumatori (lag), riassegnare le partizioni e risolvere i problemi di consumo.

Sottocomandi e argomenti comuni:

• --list: Elenca tutti i gruppi di consumatori nel cluster.
• --describe: Fornisce dettagli su gruppi di consumatori specifici, incluso il lag.
• --bootstrap-server <host:porta>: Specifica il broker Kafka a cui connettersi.
• --group <id_gruppo>: Specifica l'ID del gruppo di consumatori.
• --reset-offsets: Reimposta gli offset per un gruppo di consumatori.
• --topic <nome_topic>: Specifica il topic per il reset degli offset.
• --to-earliest: Reimposta gli offset al messaggio più vecchio disponibile.
• --to-latest: Reimposta gli offset al messaggio più recente disponibile.
• --execute: Esegue l'operazione di reset degli offset.

Esempi:

• Elenca tutti i gruppi di consumatori:

  kafka-consumer-groups.sh --list --bootstrap-server kafka-broker-1:9092
  

• Descrivi un gruppo di consumatori my_group e mostra il suo lag:

  kafka-consumer-groups.sh --describe --group my_group --bootstrap-server kafka-broker-1:9092
  

  L'output mostrerà il topic, la partizione, l'offset corrente, l'offset di fine log e il lag.

• Reimposta gli offset per my_group su my_topic al messaggio più vecchio disponibile:

  kafka-consumer-groups.sh --group my_group --topic my_topic --reset-offsets --to-earliest --execute --bootstrap-server kafka-broker-1:9092
  

  Usa questo comando con cautela, poiché influisce su dove i consumatori inizieranno a leggere.

5. kafka-log-dirs.sh

Questo strumento aiuta a ispezionare le directory dei log sui broker Kafka. Può essere utile per comprendere l'utilizzo del disco e localizzare i dati dei topic.

Argomenti comuni:

• --bootstrap-server <host:porta>: Specifica il broker Kafka a cui connettersi.
• --topic <nome_topic>: Filtra l'output per mostrare le directory per un topic specifico.

Esempi:

• Elenca le directory dei log su un broker:

  kafka-log-dirs.sh --bootstrap-server kafka-broker-1:9092
  

• Mostra le directory dei log per un topic specifico:

  kafka-log-dirs.sh --bootstrap-server kafka-broker-1:9092 --topic my_topic
  

6. kafka-preferred-replica-election.sh

Questo script avvia le elezioni delle repliche preferite per i topic. Una replica preferita è il broker scelto come leader per una partizione in base al suo fattore di replica. Se un broker fallisce e una replica non preferita diventa leader, questo strumento può essere utilizzato per riportare la leadership alla replica preferita.

Argomenti comuni:

• --topic <nome_topic>: Specifica il topic per cui eleggere le repliche preferite.
• --broker-list <id_broker1,id_broker2,...>: Specifica un elenco separato da virgole di ID broker.
• --bootstrap-server <host:porta>: Specifica il broker Kafka a cui connettersi.

Esempi:

• Elegge le repliche preferite per my_topic:

  kafka-preferred-replica-election.sh --topic my_topic --bootstrap-server kafka-broker-1:9092
  

• Elegge le repliche preferite per più topic:

  kafka-preferred-replica-election.sh --topic topic1,topic2 --bootstrap-server kafka-broker-1:9092
  

Considerazioni importanti e best practice

• --bootstrap-server è fondamentale: Assicurati sempre di specificare l'argomento --bootstrap-server corretto per connetterti al tuo cluster Kafka. Di solito è un elenco separato da virgole di host:porta per i tuoi broker.
• Ambiente: Questi comandi si trovano tipicamente nella directory bin/ della tua installazione Kafka. Dovrai navigare in questa directory o assicurarti che la directory bin di Kafka sia nel PATH del tuo sistema.
• Permessi: Assicurati che l'utente che esegue questi comandi abbia l'accesso di rete necessario per raggiungere i broker Kafka.
• Configurazione: Molti strumenti CLI possono accettare configurazioni client Kafka tramite argomenti --property o --producer-property/--consumer-property. Questo è utile per sovrascrivere serializer/deserializer predefiniti o impostare altre configurazioni client specifiche.
• Sicurezza: Per cluster Kafka sicuri (es. con autenticazione SSL/TLS o SASL), dovrai passare argomenti aggiuntivi relativi alla sicurezza (come --command-config che punta a un file di proprietà del client) a questi strumenti.
• Eliminazione dei topic: Ricorda che l'eliminazione dei topic è un'operazione delicata e deve essere esplicitamente abilitata nel file server.properties del broker Kafka usando delete.topic.enable=true.

Un modo sicuro per usare la CLI in produzione

Usa la CLI prima come strumento di ispezione e poi come strumento di modifica. --list, --describe e brevi letture dalla console sono a basso rischio. --delete, --alter, aumenti di partizioni e reset degli offset modificano il comportamento del cluster e dovrebbero seguire lo stesso percorso di revisione delle modifiche alle applicazioni, quando possibile.

Una sessione di produzione pratica di solito inizia con un file di configurazione del client:

cat client.properties
# security.protocol=SASL_SSL
# sasl.mechanism=SCRAM-SHA-512
# sasl.jaas.config=...

Poi ogni comando lo include:

kafka-topics.sh --bootstrap-server kafka-1:9093 --command-config client.properties --describe --topic orders

Per i consumer da console, evita di unirti accidentalmente a un gruppo di applicazioni reale. Usa un ID gruppo temporaneo quando ispezioni i dati e usa --max-messages in modo che il comando termini:

kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9093 \
  --command-config client.properties \
  --topic orders \
  --group debug-orders-$(date +%s) \
  --from-beginning \
  --max-messages 5 \
  --property print.key=true \
  --property print.offset=true

Questa piccola abitudine impedisce a un comando di debug di rubare partizioni da un servizio live. Lascia anche una traccia di controllo più pulita perché il nome del gruppo rende ovvia l'intenzione.

La CLI è al suo meglio quando è noiosa: un comando per ispezionare, un comando per confermare e una chiara registrazione di qualsiasi comando che cambi lo stato.

Ricette per la risoluzione dei problemi quotidiani

Se un producer dice di scrivere con successo ma il team dei consumer non vede nulla, inizia dal topic:

kafka-topics.sh --bootstrap-server kafka-1:9092 --describe --topic orders

Conferma il nome del topic, il numero di partizioni, la disponibilità del leader e le repliche in-sync. Un errore di battitura nel nome di un topic può sembrare esattamente come una pipeline interrotta quando la creazione automatica dei topic è abilitata in un cluster di sviluppo. In produzione, un topic con partizioni offline o ISR in contrazione indica un problema del broker o di replica prima di puntare al codice dell'applicazione.

Successivamente, leggi un piccolo campione con un gruppo temporaneo:

kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 \
  --topic orders \
  --group debug-orders-$(date +%s) \
  --max-messages 10 \
  --property print.key=true \
  --property print.timestamp=true \
  --property print.offset=true

Se i record appaiono lì, Kafka ha i dati e il problema è probabilmente il gruppo di consumatori reale, i suoi offset, le sue sottoscrizioni o la sua logica di elaborazione. Se non appaiono record, controlla il topic del producer, i serializer, l'autenticazione e se il producer sta scrivendo su un cluster diverso.

Per domande sul lag, vai direttamente al gruppo:

kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 --describe --group orders-writer

Non fermarti al lag totale. Confronta le partizioni. Una singola partizione con un grande lag significa un problema diverso rispetto a ogni partizione con un lag moderato. Il lag su una singola partizione spesso significa skew delle chiavi o una cattiva assegnazione del consumer. Il lag uniforme di solito significa che l'intera applicazione è più lenta del tasso di input.

Per domande del tipo "cosa è cambiato?", ispeziona la configurazione del topic:

kafka-configs.sh \
  --bootstrap-server kafka-1:9092 \
  --entity-type topics \
  --entity-name orders \
  --describe

Qui è dove cogli le modifiche alla retention, le sorprese della politica di pulizia, le sostituzioni della compressione e le impostazioni della dimensione dei messaggi che differiscono dalle ipotesi del servizio.

L'output della CLI non sostituisce il monitoraggio, ma è eccellente per ridurre l'incertezza. In un incidente reale, alcuni output di comandi incollati nel ticket possono salvare tutti dal dibattere se il topic esiste, se i record sono presenti e se il gruppo si sta effettivamente muovendo.

Comandi che meritano attenzione

Alcuni comandi CLI di Kafka sembrano innocui perché sono brevi. Non sono innocui.

kafka-topics.sh --alter --partitions aumenta solo il numero di partizioni; non lo riduce in seguito se ti penti della modifica. Più partizioni possono aiutare il parallelismo dei consumer, ma possono anche cambiare la distribuzione delle chiavi per i nuovi record e rompere le ipotesi nei sistemi che si aspettavano che tutti gli eventi per un intervallo di chiavi finissero in un insieme più piccolo di partizioni.

kafka-consumer-groups.sh --reset-offsets --execute cambia dove un gruppo leggerà successivamente. Usa prima --dry-run, ferma i consumer interessati e registra i vecchi offset. Reimpostare al più vecchio può riprodurre dati in sistemi che non sono idempotenti. Reimpostare al più recente può saltare dati che l'azienda si aspetta ancora di elaborare.

kafka-topics.sh --delete dipende dalla configurazione e dalla politica del cluster, ma quando l'eliminazione è consentita, dovrebbe essere trattata come l'eliminazione di una tabella del database. Controlla il cluster, controlla il topic e controlla se un altro ambiente utilizza la stessa convenzione di denominazione. Un topic di produzione chiamato orders-test è ancora produzione se i servizi reali dipendono da esso.

Per operazioni ripetibili, inserisci il comando in un runbook o script con variabili esplicite per cluster, topic, gruppo e configurazione del comando. La CLI è ottima per le indagini, ma la mutazione in produzione dovrebbe essere noiosa, revisionata e facile da controllare.