클러스터 크기 조정: 브로커 수 및 크기

Kafka 클러스터 크기 조정은 두 가지 근본적인 질문으로 시작됩니다. 처리량은 얼마입니까? 목표 쓰기 속도(MB/s) 및 필요한 보존 기간은 얼마입니까(데이터를 얼마나 오랫동안 사용할 수 있어야 합니까)? 이 두 질문에 대한 답에 따라 필요한 브로커 수, 디스크 크기 및 네트워크가 결정됩니다.

브로커 번호 공식

각 Kafka 브로커는 대략적으로 지속 가능하게 관리할 수 있습니다. 100-200MB/초 처리량 상용 하드웨어(NVMe SSD 디스크, 10Gbps 네트워크)에 쓰기 이 값은 많이 달라집니다. 워크로드 프로필(메시지 크기, 파티션 수, ack 설정)에서. 

# 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

프로덕션 Kafka 브로커에 권장되는 하드웨어

  • CPU: 16~32개 코어(Kafka는 CPU에 바인딩되지 않지만 압축에서는 이를 사용함)
  • 숫양: 64~128GB(성능을 위해서는 페이지 캐시가 필수)
  • 디스코: NVMe SSD RAID 10개 이상의 개별 디스크(병렬 I/O를 위해 별도로 장착)
  • 그물: 최소 10Gbps, 높은 처리량 클러스터의 경우 25Gbps
  • OS: ext4 또는 XFS를 사용하는 Linux, 순차 추가 쓰기에 최적화된 파일 시스템
  • JVM: Java 21(LTS), G1GC, 6~8GB 힙, -XX:MaxGCPauseMillis=20

최적의 브로커 구성: server.properties

기본 Kafka 구성은 프로덕션에 적합하지 않은 경우가 많습니다. 다음은 속성입니다. 커스터마이징에서 가장 중요한 것은 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

파티션 수: 생성할 개수

파티션 수 선택은 Kafka에서 가장 중요하지만 되돌릴 수 없는 부분 중 하나입니다. 그들은 할 수 있다 증가하다 파티션( kafka-topics.sh --alter) 하지만 감소하지 마십시오. 파티션을 늘리면 정렬 및 재조정에 영향을 미칩니다. 

# 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

보존 정책: 시간 vs 크기 vs 압축

보존 정책은 Kafka에서 레코드를 사용할 수 있는 기간을 결정합니다. 잘못된 선택은 두 가지 상반된 문제, 즉 디스크 부족(너무 긴 보존 기간) 또는 느린 소비자에게는 재생이 불가능합니다(보존이 너무 짧음). 

# 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

롤링 재시작: 가동 중지 시간 없이 클러스터 업그레이드

Kafka 클러스터 업데이트(소프트웨어 버전, 브로커 구성, JVM)가 수행됩니다. 와 롤링 재시작: 한 번에 하나의 브로커를 다시 시작하고 ISR을 기다립니다. 다음 브로커를 진행하기 전에 완전히 재구축하세요. 

# 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: 지리적 복제 및 재해 복구

미러메이커 2(MM2) 개별 클러스터 간에 데이터를 복제하기 위한 Kafka 구성 요소입니다. Kafka Connect를 기반으로 구축되었으며 양방향 복제, 오프셋 동기화를 지원합니다. 그리고 자동 장애 조치. 이는 다음 용도로 사용됩니다:

  • 재해 복구(DR): 기본 클러스터에서 다른 데이터 센터의 대기 클러스터로 복제합니다.
  • 지리적 분포: 로컬 소비자에 대한 짧은 대기 시간을 위해 여러 지역에 걸쳐 데이터가 복제됩니다.
  • 이주: 다운타임 없이 클러스터 간 마이그레이션을 제어합니다.
# 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

MirrorMaker 2를 사용한 장애 조치: 소비자 오프셋 변환

MM2의 가장 강력한 기능 중 하나는 오프셋 변환: 보조 클러스터의 오프셋이 기본 클러스터의 오프셋과 다릅니다(사본에는 복제 지연이 있는 경우 모든 기록). MM2는 주제에서 매핑 테이블을 유지 관리합니다. mm2-offset-syncs.primary.internal 오프셋을 변환합니다. 

# 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

제작을 위한 주제 구성: 체크리스트

# 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

Strimzi Operator를 사용하는 Kubernetes의 Kafka

Kubernetes에 배포하는 경우 스트림지 공식 참조 연산자입니다 (CNCF 샌드박스의 일부). Kafka 클러스터의 생성, 업데이트 및 구성을 관리합니다. 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

성능 튜닝: 모든 것을 변화시키는 구성

프로덕션의 일반적인 안티 패턴

  • 속도는 acks=0: 내구성 보장 제로, 브로커 오류 발생 시 메시지가 자동으로 손실됨
  • 복제.인자=1: 중복 없음, 브로커 손실로 인해 데이터가 파괴됨
  • auto.create.topics.enable=true: 오타로 인해 생성된 토픽, 확인 불가
  • JVM 힙 > 8GB: 브로커의 성능을 저하시키는 GC 긴 일시 중지를 유발합니다.
  • 파티션이 너무 적음: 다운타임 없이는 해결할 수 없는 병렬성 병목 현상
# 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)

요약: 프로덕션 중인 Kafka 체크리스트

  • 최소 3개의 브로커가 있는 KRaft 클러스터(고가용성을 위해 5개 이상)
  • replication.factor=3, min.insync.replicas=2 모든 중요한 주제에 대해
  • auto.create.topics.enable=false: 주제를 명시적으로 관리합니다.
  • JVM 힙 6-8GB, G1GC 구성, 나머지 RAM은 페이지 캐시용
  • OS와 별도로 전용 디스크(NVMe SSD)에 로그온
  • 모니터링: JMX 수출업체 + Prometheus + Grafana(9조 참조)
  • 경고: 오프라인 파티션, 복제되지 않은 파티션, 소비자 지연
  • Consumer Group별 DLQ 구성(10조 참조)
  • 보조 데이터 센터의 지역 복제 및 DR을 위한 MirrorMaker 2
  • 롤링 재시작 절차를 문서화하고 정기적으로 테스트함
  • 성장 추세에 따라 분기별로 용량 계획을 검토합니다.

시리즈 종료: 다음 단계

전체 Apache Kafka 시리즈를 완료했습니다. 관련 주제에 대해 자세히 알아보는 방법은 다음과 같습니다.

  • 이벤트 기반 아키텍처(시리즈 39): Kafka를 EDA 패턴의 백본으로 적용, 복잡한 마이크로서비스 시스템의 Saga, CQRS 및 Outbox.
  • PostgreSQL AI 및 Debezium CDC: Kafka를 파이프라인으로 사용하여 변경 사항 전파 데이터베이스에서 다운스트림 서비스까지 실시간으로 제공됩니다.

다른 시리즈와의 연계

  • 이벤트 중심 아키텍처 – Saga 패턴 및 CQRS: 메시지 브로커로서의 Kafka 마이크로서비스 시스템에서 Saga Pattern 및 CQRS 예측을 구현합니다.
  • Prometheus 및 Grafana를 통한 모니터링(9조): 보유하고 있는 클러스터 측정항목 이 가이드에서 구성한 것은 JMX 내보내기를 통해 노출되고 Grafana에 표시됩니다.