Rozmiar klastra: ilu brokerów i jaki rozmiar

Ustalanie rozmiaru klastra Kafki rozpoczyna się od dwóch podstawowych pytań: Jaka jest przepustowość docelowa szybkość zapisu (MB/s) i jakie jest wymagane przechowywanie (jak długo dane muszą pozostać dostępne)? Odpowiedź na te dwa pytania określa liczbę brokerów, rozmiar dysku i potrzebną sieć.

Formuła numeru brokera

Każdy broker Kafka może w sposób zrównoważony zarządzać ok 100-200 MB/s przepustowości zapisz na standardowym sprzęcie (dysk NVMe SSD, sieć 10 Gb/s). Ta wartość bardzo zależy z profilu obciążenia (rozmiar wiadomości, liczba partycji, ustawienia potwierdzeń). 

# Formula per il sizing del cluster
#
# Throughput totale in scrittura: T_write = messaggi_al_secondo * dimensione_media_messaggio
# Throughput totale con replica:  T_total = T_write * replication_factor
# Numero di broker:               N_broker = ceil(T_total / throughput_per_broker)
#
# Esempio pratico:
# - 100.000 messaggi/s * 1 KB/messaggio = 100 MB/s throughput in scrittura
# - Replication factor 3:                 100 * 3 = 300 MB/s throughput totale
# - Throughput per broker: 150 MB/s       300 / 150 = 2 broker
# - Aggiungi 30% di margine:              ceil(2 * 1.3) = 3 broker
#
# Per la memoria RAM:
# - Kafka usa la page cache del SO per le letture recenti (hot data)
# - Regola empirica: 64-128 GB RAM per broker (o almeno 50% del dataset "caldo")
# - JVM heap: 6-8 GB (non di piu: il resto serve alla page cache)
#
# Per il disco:
# Disco_per_broker = (T_write * retention_giorni * 86400 * replication_factor) / N_broker

# Esempio: 100 MB/s, 7 giorni di retention, replication factor 3, 3 broker
# = (100 * 7 * 86400 * 3) / 3 = 181.440.000 MB ~= 173 TB totale / 3 broker = ~58 TB per broker
# Arrotonda a 80 TB con margine

Zalecany sprzęt dla brokera Kafka w produkcji

  • Procesor: 16-32 rdzeni (Kafka nie jest związana z procesorem, ale kompresja go wykorzystuje)
  • BARAN: 64-128 GB (pamięć podręczna strony jest niezbędna dla wydajności)
  • Dyskoteka: NVMe SSD RAID 10 lub więcej oddzielnych dysków (montowanych oddzielnie dla równoległych wejść/wyjść)
  • Internet: minimum 10 Gb/s, 25 Gb/s dla klastrów o dużej przepustowości
  • OS: Linux z ext4 lub XFS, system plików zoptymalizowany pod kątem dołączania zapisów sekwencyjnych
  • JVM: Java 21 (LTS), G1GC, sterta 6-8 GB, -XX:MaxGCPauseMillis=20

Optymalna konfiguracja brokera: serwer.properties

Domyślne konfiguracje Kafki są często nieoptymalne dla środowiska produkcyjnego. Oto właściwości najważniejsze, aby dostosować się server.properties: 

# server.properties - configurazione production-ready per Kafka 4.0

# ===== Rete e I/O =====
# Thread per le richieste di rete
num.network.threads=8
# Thread per le operazioni di I/O su disco
num.io.threads=16
# Buffer per richieste/risposte di rete
socket.send.buffer.bytes=1048576      # 1 MB
socket.receive.buffer.bytes=1048576   # 1 MB
socket.request.max.bytes=104857600    # 100 MB

# ===== Log e Disco =====
# Separare i log su piu dischi per I/O parallelo
log.dirs=/disk1/kafka-data,/disk2/kafka-data,/disk3/kafka-data

# Numero di thread per il log recovery all'avvio
num.recovery.threads.per.data.dir=4

# Flush su disco: NON abbassare questi valori, usa acks=all invece
log.flush.interval.messages=10000000
log.flush.interval.ms=1000

# ===== Retention Default =====
# Retention temporale (7 giorni)
log.retention.hours=168
# Retention per dimensione (-1 = illimitata per default)
log.retention.bytes=-1
# Dimensione massima del segmento log (1 GB)
log.segment.bytes=1073741824
# Intervallo di controllo per la retention
log.retention.check.interval.ms=300000

# ===== Replica =====
# Numero minimo di ISR per accettare scritture (con acks=all)
min.insync.replicas=2
# Default replication factor per topic auto-creati
default.replication.factor=3
# Replication factor per topic interni
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2

# ===== Performance Replica =====
# Dimensione buffer per la replica
replica.fetch.max.bytes=10485760      # 10 MB
replica.socket.receive.buffer.bytes=10485760
# Timeout per il leader election
leader.imbalance.check.interval.seconds=300

# ===== Consumer Groups =====
# Timeout sessione consumer
group.initial.rebalance.delay.ms=3000
# ===== Auto Create Topics =====
# DISABILITARE in produzione per evitare topic creati per errore
auto.create.topics.enable=false

Liczba partycji: ile utworzyć

Wybór liczby przegród jest w Kafce jednym z najbardziej krytycznych i najmniej odwracalnych: mogą zwiększyć partycje (z kafka-topics.sh --alter) Ale nie zmniejszaj. Zwiększanie partycji wpływa na sortowanie i przywracanie równowagi. 

# Regola pratica per il numero di partizioni:
# max(throughput_MB_s / 10, consumer_max_paralleli, producer_max_paralleli)
#
# Esempio:
# - Throughput target: 50 MB/s  --> almeno 5 partizioni per throughput
# - Consumer parallelismo max: 12 istanze --> almeno 12 partizioni
# - Scelta: 12 partizioni
#
# Linee guida pratiche:
# - Topic ad alto volume, molti consumer: 12, 24, 48 partizioni
# - Topic a basso volume, pochi consumer: 3, 6 partizioni
# - Topic interni (audit, DLQ): 3 partizioni sono spesso sufficienti
# - Non creare mai piu di 4000-6000 partizioni per broker (overhead memoria)

# Creare un topic con sizing ottimale
kafka-topics.sh --create \
  --bootstrap-server kafka1:9092 \
  --topic pagamenti-confermati \
  --partitions 12 \
  --replication-factor 3 \
  --config min.insync.replicas=2 \
  --config retention.ms=604800000 \
  --config compression.type=snappy

# Verificare la distribuzione delle partizioni tra i broker
kafka-topics.sh --describe \
  --bootstrap-server kafka1:9092 \
  --topic pagamenti-confermati

# Se la distribuzione e sbilanciata (troppi leader sullo stesso broker):
kafka-leader-election.sh \
  --bootstrap-server kafka1:9092 \
  --election-type preferred \
  --all-topic-partitions

Polityka przechowywania: czas vs rozmiar vs zagęszczenie

Polityka przechowywania określa, jak długo rekordy pozostają dostępne w Kafce. Zły wybór prowadzi do dwóch przeciwstawnych problemów: brak dysku (zbyt długie przechowywanie) lub powtórka niemożliwa dla powolnych konsumentów (zbyt krótkie zatrzymanie). 

# Configurazioni di retention per diversi use case

# Use case 1: Event streaming real-time (clickstream, metriche)
# Retention breve, alta velocita, i dati vengono aggregati immediatamente
kafka-topics.sh --create \
  --bootstrap-server kafka1:9092 \
  --topic click-events \
  --partitions 24 \
  --replication-factor 3 \
  --config retention.ms=3600000 \        # 1 ora
  --config retention.bytes=5368709120 \  # 5 GB per partizione
  --config compression.type=lz4          # compressione veloce

# Use case 2: Integrazione tra servizi (domain events)
# Retention media, consumer devono poter recuperare eventi recenti
kafka-topics.sh --create \
  --bootstrap-server kafka1:9092 \
  --topic ordini-effettuati \
  --partitions 12 \
  --replication-factor 3 \
  --config retention.ms=604800000 \  # 7 giorni (default)
  --config compression.type=snappy

# Use case 3: Audit log, compliance
# Retention lunga, dati critici per regolamentazione
kafka-topics.sh --create \
  --bootstrap-server kafka1:9092 \
  --topic audit-trail \
  --partitions 6 \
  --replication-factor 3 \
  --config retention.ms=94608000000 \  # 3 anni
  --config compression.type=gzip       # compressione massima per dati storici

# Use case 4: Change Data Capture (CDC), topic di stato
# Log compaction: mantiene solo ultimo valore per chiave
kafka-topics.sh --create \
  --bootstrap-server kafka1:9092 \
  --topic clienti-profilo \
  --partitions 6 \
  --replication-factor 3 \
  --config cleanup.policy=compact \
  --config min.cleanable.dirty.ratio=0.5 \
  --config delete.retention.ms=86400000  # tombstone retention 24h

Rolling Restart: uaktualnij klaster bez przestojów

Wykonywane są aktualizacje klastra Kafka (wersja oprogramowania, konfiguracja brokera, JVM). z ciągły restart: Restartujesz jednego brokera na raz, czekając na ISR odbuduj całkowicie przed przystąpieniem do następnego brokera. 

# Procedura di rolling restart sicura

# 1. Verifica che il cluster sia sano prima di iniziare
kafka-topics.sh --describe \
  --bootstrap-server kafka1:9092 \
  --under-replicated-partitions
# Output atteso: nessuna partizione elencata

# 2. Per ogni broker (uno alla volta):
# a) Marca il broker come "not preferred" per evitare leader election inutili
kafka-leader-election.sh --bootstrap-server kafka1:9092 \
  --election-type unclean --all-topic-partitions

# b) Riavvia il broker
systemctl restart kafka

# c) ASPETTA che il broker si riunga al cluster e l'ISR si ricostruisca
# Monitora: tutte le under-replicated partitions devono tornare a 0
watch -n 5 "kafka-topics.sh --bootstrap-server kafka1:9092 --describe --under-replicated-partitions"

# d) Solo quando l'ISR e completo, passa al broker successivo

# 3. Dopo il rolling restart, ribilancia i leader
kafka-leader-election.sh \
  --bootstrap-server kafka1:9092 \
  --election-type preferred \
  --all-topic-partitions

# Verifica finale
kafka-broker-api-versions.sh --bootstrap-server kafka1:9092

MirrorMaker 2: Replikacja geograficzna i odzyskiwanie po awarii

LustroMaker 2 (MM2) to komponent Kafki służący do replikacji danych pomiędzy odrębnymi klastrami. Jest zbudowany na Kafka Connect i obsługuje replikację dwukierunkową, synchronizację offsetową i automatyczne przełączanie awaryjne. Służy do:

  • Odzyskiwanie po awarii (DR): Replikuje się z klastra podstawowego do klastra rezerwowego w innym centrum danych
  • Dystrybucja geograficzna: Dane replikowane w wielu regionach w celu zapewnienia małych opóźnień dla lokalnych konsumentów
  • Emigracja: Kontrolowana migracja między klastrami bez przestojów
# mm2.properties - Configurazione MirrorMaker 2
# Replica da cluster "primary" a "secondary" (datacenter DR)

# I due cluster
clusters=primary,secondary

# Connessione ai cluster
primary.bootstrap.servers=kafka-primary-1:9092,kafka-primary-2:9092,kafka-primary-3:9092
secondary.bootstrap.servers=kafka-secondary-1:9092,kafka-secondary-2:9092,kafka-secondary-3:9092

# Abilita la replica primary -> secondary
primary->secondary.enabled=true
# Non abilitare secondary->primary per evitare loop (solo se non e' bidirezionale)
secondary->primary.enabled=false

# Topic da replicare (regex): tutti i topic produzione escludendo interni
primary->secondary.topics=^(?!(__|\.)).*$

# Sincronizza anche le configurazioni dei topic
primary->secondary.sync.topic.configs.enabled=true
primary->secondary.sync.topic.acls.enabled=true

# Sincronizza gli offset dei consumer group
primary->secondary.sync.group.offsets.enabled=true
primary->secondary.sync.group.offsets.interval.seconds=60

# Prefisso per i topic replicati (su secondary avrai "primary.ordini-effettuati")
replication.factor=3

# Performance
tasks.max=4
producer.override.acks=all
producer.override.compression.type=snappy

# Avvio di MirrorMaker 2:
# connect-mirror-maker.sh mm2.properties

Praca awaryjna z MirrorMaker 2: Tłumaczenie offsetowe konsumenckie

Jedną z najpotężniejszych funkcji MM2 jest tłumaczenie offsetowe: przesunięcia w klastrze dodatkowym różnią się od przesunięć w klastrze podstawowym (kopia może nie mieć wszystkie rekordy, jeśli występuje opóźnienie replikacji). MM2 utrzymuje tabelę mapowania w temacie mm2-offset-syncs.primary.internal do tłumaczenia przesunięć. 

# RemoteClusterUtils: tradurre gli offset per il failover
# Da usare durante un failover emergency per far ripartire i consumer
# dal punto corretto sul cluster secondary

from confluent_kafka.admin import AdminClient
import json

# Script di failover: trova l'offset tradotto per ogni consumer group
def translate_offsets_for_failover(primary_group_id, secondary_bootstrap):
    """
    Usa i checkpoint di MM2 per trovare l'offset equivalente
    su secondary per un consumer group originalmente su primary.
    """
    admin = AdminClient({'bootstrap.servers': secondary_bootstrap})

    # MM2 scrive i checkpoint nel topic dedicato
    # Topic: mm2-checkpoints.primary.internal
    # Leggi i checkpoint per il gruppo specifico

    # In alternativa, usa la MM2 RemoteClusterUtils Java API:
    # Map<TopicPartition, OffsetAndMetadata> translated =
    #   RemoteClusterUtils.translateOffsets(
    #     adminClient, "primary", groupId, Duration.ofMinutes(1));

    print(f"Failover completato: consumer group '{primary_group_id}' "
          f"ora punta al cluster secondary con offset tradotti")

# Dopo il failover, cambia il bootstrap.servers dei consumer
# da primary a secondary e ripartono dal punto corretto

Konfigurowanie tematów do produkcji: lista kontrolna

# Script di creazione topic production-ready con tutti i parametri

create_production_topic() {
    TOPIC=$1
    PARTITIONS=$2
    RETENTION_DAYS=$3

    RETENTION_MS=$((RETENTION_DAYS * 24 * 3600 * 1000))

    kafka-topics.sh --create \
        --bootstrap-server kafka1:9092 \
        --topic $TOPIC \
        --partitions $PARTITIONS \
        --replication-factor 3 \
        --config min.insync.replicas=2 \
        --config retention.ms=$RETENTION_MS \
        --config compression.type=snappy \
        --config max.message.bytes=10485760 \  # 10 MB max message
        --if-not-exists

    echo "Topic $TOPIC creato: $PARTITIONS partizioni, RF=3, MIR=2, retention=${RETENTION_DAYS}d"
}

# Crea i topic principali
create_production_topic ordini-effettuati 12 7
create_production_topic pagamenti-confermati 12 30
create_production_topic audit-trail 3 1095        # 3 anni
create_production_topic ordini-effettuati.DLT 3 30  # DLQ

# Verifica di tutti i topic
kafka-topics.sh --list --bootstrap-server kafka1:9092

# Verifica configurazione di un topic specifico
kafka-configs.sh --bootstrap-server kafka1:9092 \
    --entity-type topics \
    --entity-name ordini-effettuati \
    --describe

Kafka na Kubernetesie z operatorem Strimzi

W przypadku wdrożenia na platformie Kubernetes Strimzi jest oficjalnym operatorem referencyjnym (część piaskownicy CNCF). Zarządza tworzeniem, aktualizacją i konfiguracją klastrów Kafka za pośrednictwem niestandardowych definicji zasobów Kubernetes (CRD). 

# kafka-cluster.yaml - Strimzi KafkaNodePool + Kafka CRD
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaNodePool
metadata:
  name: broker
  namespace: kafka
  labels:
    strimzi.io/cluster: production-cluster
spec:
  replicas: 3
  roles:
    - broker
  storage:
    type: persistent-claim
    size: 2Ti       # 2 TB per broker
    class: fast-ssd
  resources:
    requests:
      memory: 32Gi
      cpu: "4"
    limits:
      memory: 64Gi
      cpu: "8"
  jvmOptions:
    -Xms: 6144m
    -Xmx: 6144m
    gcLoggingEnabled: false
---
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: production-cluster
  namespace: kafka
spec:
  kafka:
    version: 4.0.0
    metadataVersion: "4.0"
    replicas: 3
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
    config:
      auto.create.topics.enable: "false"
      default.replication.factor: 3
      min.insync.replicas: 2
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      log.retention.hours: 168
      compression.type: snappy
  zookeeper:
    replicas: 0  # KRaft: niente ZooKeeper in Kafka 4.0

Strojenie wydajności: konfiguracje, które zmieniają wszystko

Typowe anty-wzorce w produkcji

  • acks=0 dla szybkości: gwarancja zerowej trwałości, wiadomości utracone w trybie cichym w przypadku awarii brokera
  • współczynnik replikacji=1: brak redundancji, utrata brokera niszczy dane
  • auto.create.topics.enable=true: tematy utworzone przez literówkę, niemożliwe do sprawdzenia
  • Sterta JVM > 8 GB: powoduje długie przerwy w GC, które pogarszają wydajność brokera
  • Za mało partycji: wąskie gardło związane z równoległością niemożliwe do rozwiązania bez przestojów
# Configurazioni producer per massimo throughput (batch processing)
props.put("batch.size", 65536);          # 64 KB per batch
props.put("linger.ms", 20);              # Aspetta 20ms per riempire il batch
props.put("compression.type", "snappy"); # Snappy: buon bilanciamento velocita/ratio
props.put("buffer.memory", 67108864);    # 64 MB buffer totale

# Configurazioni consumer per massimo throughput
props.put("fetch.min.bytes", 65536);     # 64 KB fetch minimo
props.put("fetch.max.wait.ms", 500);     # Aspetta max 500ms se non ci sono dati
props.put("max.partition.fetch.bytes", 10485760);  # 10 MB per fetch per partizione
props.put("max.poll.records", 500);      # 500 record per poll()

# Per sistemi a bassa latenza (trading, real-time alerts):
props.put("linger.ms", 0);              # Invia immediatamente
props.put("batch.size", 1);             # Nessun batching
props.put("acks", "1");                 # Solo leader ack (senza aspettare ISR)

Podsumowanie: Lista kontrolna dla Kafki w produkcji

  • Klaster KRaft z minimum 3 brokerami (5+ w celu zapewnienia wysokiej dostępności)
  • replication.factor=3, min.insync.replicas=2 na wszystkie krytyczne tematy
  • auto.create.topics.enable=false: jawnie zarządzaj tematami
  • Sterta JVM 6-8 GB, skonfigurowana G1GC, reszta pamięci RAM na pamięć podręczną stron
  • Zaloguj się na dedykowanym dysku (NVMe SSD), niezależnym od systemu operacyjnego
  • Monitoring: JMX Exporter + Prometheus + Grafana (patrz artykuł 9)
  • Alert dotyczący: partycji offline, niedostatecznie zreplikowanych partycji, opóźnień konsumenckich
  • DLQ skonfigurowane dla każdej grupy konsumentów (patrz artykuł 10)
  • MirrorMaker 2 do replikacji geograficznej i odzyskiwania po awarii w dodatkowym centrum danych
  • Procedura ciągłego restartu udokumentowana i regularnie testowana
  • Planowanie wydajności weryfikowane co kwartał w oparciu o trendy wzrostu

Koniec serii: Kolejne kroki

Ukończyłeś całą serię Apache Kafka. Aby dowiedzieć się więcej na powiązane tematy:

  • Architektura sterowana zdarzeniami (seria 39): zastosuj Kafkę jako szkielet wzorców EDA, Saga, CQRS i Outbox w złożonych systemach mikroserwisowych.
  • PostgreSQL AI i Debezium CDC: Użyj Kafki jako potoku do propagowania zmian z bazy danych do usług podrzędnych w czasie rzeczywistym.

Połącz z innymi seriami

  • Architektura sterowana zdarzeniami – wzorzec Saga i CQRS: Kafka jako broker komunikatów w celu wdrożenia projekcji Saga Pattern i CQRS w systemach mikroserwisowych.
  • Monitoring z Prometeuszem i Grafaną (art. 9): posiadane metryki klastra skonfigurowane w tym przewodniku to te udostępniane za pośrednictwem JMX Exporter i wyświetlane w Grafanie.