로컬 KRaft 클러스터 구축

kafka-lab/
├── docker-compose.yml     # 3노드 KRaft 클러스터
├── .env                   # 이미지 태그·클러스터 ID 등 변수
├── create-topics.sh        # 실습용 토픽 생성 (멱등)
└── kcli                    # 컨테이너 안 CLI 를 호스트에서 쓰는 래퍼

# Apache Kafka 4.3.1 · KRaft combined 모드 3노드 클러스터
# ZooKeeper 컨테이너는 없습니다. ZooKeeper 모드는 Kafka 4.0에서 제거되었습니다.
#
# 세 노드가 모두 broker + controller 를 겸합니다(combined).
# 컨트롤러 쿼럼이 3이므로 노드 1대가 죽어도 과반(2)이 남아 클러스터가 유지됩니다.
---
name: kafka-lab                 # compose 프로젝트 이름. 볼륨/네트워크 접두어가 됩니다

# ---------------------------------------------------------------------------
# 세 노드가 공유하는 설정을 YAML 앵커로 한 번만 씁니다.
# 노드별로 다른 값(NODE_ID, ADVERTISED_LISTENERS)만 아래에서 덮어씁니다.
# ---------------------------------------------------------------------------
x-kafka-common: &kafka-common
  image: ${KAFKA_IMAGE}
  restart: unless-stopped
  environment: &kafka-env
    # --- KRaft 롤과 쿼럼 -----------------------------------------------------
    KAFKA_PROCESS_ROLES: 'broker,controller'
    # 컨트롤러 쿼럼 투표자 3명을 {node.id}@{host}:{port} 형식으로 고정합니다.
    # 3이면 1대 손실을 견딥니다(과반 2). 5는 2대까지 견디지만 커밋 지연이 늘어납니다.
    # 짝수는 의미가 없습니다 — 4대는 3대와 같은 1대 손실만 견디면서 비용만 늘어납니다.
    KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka-1:9093,2@kafka-2:9093,3@kafka-3:9093'
    # 컨트롤러 트래픽이 흐르는 리스너 이름. 클라이언트 리스너와 반드시 분리합니다.
    KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'

    # --- 리스너 3종 ----------------------------------------------------------
    # 이름 → 보안 프로토콜 매핑. 리스너 이름은 임의로 정할 수 있으므로
    # 브로커에게 "이 이름이 어떤 프로토콜인지" 알려 주어야 합니다.
    KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
    # 실제로 bind 하는 소켓. 컨테이너 내부 관점의 포트입니다.
    KAFKA_LISTENERS: 'PLAINTEXT://:19092,CONTROLLER://:9093,PLAINTEXT_HOST://:9092'
    # 브로커끼리(복제·컨트롤러 요청) 쓸 리스너. 컨테이너 네트워크 쪽을 씁니다.
    KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'

    # --- 내부 토픽 복제 계수 --------------------------------------------------
    # 기본값은 3인데 단일 노드 예제에서는 1로 내려야 뜹니다.
    # 여기는 브로커가 3대이므로 기본값 그대로 3을 명시해 의도를 드러냅니다.
    # 이 값을 1로 두면 __consumer_offsets 를 가진 브로커 1대가 죽는 순간
    # 컨슈머 그룹 전체가 오프셋을 잃습니다.
    KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
    KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
    # 트랜잭션 로그의 최소 ISR. RF=3에 min.isr=2 가 표준 조합입니다.
    # 1로 두면 트랜잭션 커밋이 유실될 수 있고, 3으로 두면 1대만 빠져도 트랜잭션이 멈춥니다.
    KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
    # Share Groups(KIP-932, 4.2 production-ready) 상태 토픽도 같은 원칙을 적용합니다.
    KAFKA_SHARE_COORDINATOR_STATE_TOPIC_REPLICATION_FACTOR: 3
    KAFKA_SHARE_COORDINATOR_STATE_TOPIC_MIN_ISR: 2

    # --- 사용자 토픽 기본값 --------------------------------------------------
    # 자동 생성 토픽은 파티션 1 / RF 1 이 되어 실습을 망칩니다.
    # 오타 토픽이 조용히 만들어지는 것도 막기 위해 자동 생성을 끕니다.
    KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'false'
    KAFKA_NUM_PARTITIONS: 3
    KAFKA_DEFAULT_REPLICATION_FACTOR: 3
    # 브로커 기본 min.insync.replicas. 토픽 레벨 설정이 이 값을 덮어씁니다.
    KAFKA_MIN_INSYNC_REPLICAS: 2

    # --- 로컬 실습 편의 ------------------------------------------------------
    # 기본값 3000(3초). 컨슈머가 그룹에 처음 붙을 때 이만큼 리밸런스를 지연시켜
    # 여러 컨슈머를 한 번에 묶습니다. 로컬에서는 즉시 시작하는 편이 편해 0으로 둡니다.
    # 프로덕션에서는 절대 0으로 두지 마세요(배포 중 리밸런스가 폭증합니다).
    KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
    # 로그 디렉터리. 아래 named volume 이 이 경로에 마운트됩니다.
    KAFKA_LOG_DIRS: '/var/lib/kafka/data'
    # JVM 힙. 노드 3개가 뜨므로 로컬에서는 512MB 로 묶어 둡니다.
    KAFKA_HEAP_OPTS: '-Xmx512m -Xms512m'
    # 세 노드가 같은 클러스터 ID 를 공유해야 합니다. 다르면 서로를 거부합니다.
    CLUSTER_ID: ${CLUSTER_ID}
  healthcheck:
    # 브로커 API 응답이 오면 준비 완료로 봅니다.
    # depends_on 의 service_healthy 조건이 이 결과를 씁니다.
    test: ['CMD-SHELL', '/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:19092 >/dev/null 2>&1']
    interval: 10s
    timeout: 10s
    retries: 12
    start_period: 30s

services:
  kafka-1:
    <<: *kafka-common
    hostname: kafka-1           # advertised.listeners 의 호스트명과 일치해야 합니다
    container_name: kafka-1
    ports:
      - '29092:9092'            # 호스트 29092 → 컨테이너 9092 (PLAINTEXT_HOST)
    environment:
      <<: *kafka-env
      KAFKA_NODE_ID: 1
      # 컨테이너 내부용 주소와 호스트용 주소를 함께 광고합니다.
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka-1:19092,PLAINTEXT_HOST://localhost:29092'
    volumes:
      - kafka-1-data:/var/lib/kafka/data

  kafka-2:
    <<: *kafka-common
    hostname: kafka-2
    container_name: kafka-2
    ports:
      - '39092:9092'
    environment:
      <<: *kafka-env
      KAFKA_NODE_ID: 2
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka-2:19092,PLAINTEXT_HOST://localhost:39092'
    volumes:
      - kafka-2-data:/var/lib/kafka/data

  kafka-3:
    <<: *kafka-common
    hostname: kafka-3
    container_name: kafka-3
    ports:
      - '49092:9092'
    environment:
      <<: *kafka-env
      KAFKA_NODE_ID: 3
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka-3:19092,PLAINTEXT_HOST://localhost:49092'
    volumes:
      - kafka-3-data:/var/lib/kafka/data

volumes:
  # named volume 을 쓰지 않으면 컨테이너를 지울 때 토픽과 메타데이터가 사라집니다.
  # 클러스터를 완전히 초기화하려면 `docker compose down -v` 로 볼륨까지 지웁니다.
  kafka-1-data:
  kafka-2-data:
  kafka-3-data:

# 이미지 태그를 한 곳에서 관리합니다. 4.3.1 은 이 예제를 검증한 버전입니다.
# GraalVM 네이티브 이미지(apache/kafka-native)는 실험적이며 프로덕션 용도가 아닙니다.
KAFKA_IMAGE=apache/kafka:4.3.1

# 클러스터 ID — 세 노드가 반드시 같은 값을 공유해야 합니다.
# 새로 만들려면: docker run --rm apache/kafka:4.3.1 /opt/kafka/bin/kafka-storage.sh random-uuid
CLUSTER_ID=4L6g3nShT-eMCtK--X86sw

#!/usr/bin/env bash
# 실습용 토픽 생성. 여러 번 실행해도 안전합니다(--if-not-exists).
# 사용법: ./create-topics.sh
set -euo pipefail

# 컨테이너 안에서 실행하므로 컨테이너 네트워크 주소를 씁니다.
BROKER_INTERNAL='kafka-1:19092,kafka-2:19092,kafka-3:19092'
KAFKA_BIN='/opt/kafka/bin'

# 토픽 정의: 이름:파티션수:복제계수:추가설정(쉼표구분, 없으면 빈 문자열)
TOPICS=(
  # 예제 3(무손실 프로듀서) · 예제 4(오프셋 전략)
  "orders:6:3:min.insync.replicas=2"
  # 예제 5(EOS 파이프라인) 입력/출력
  "orders.raw:6:3:min.insync.replicas=2"
  "orders.enriched:6:3:min.insync.replicas=2"
  # 예제 6(DLQ) — DLQ 는 유실이 곧 사고 원인 소실이므로 보관을 길게 둡니다(30일)
  "payments:3:3:min.insync.replicas=2"
  "payments.DLT:3:3:min.insync.replicas=2,retention.ms=2592000000"
  # 예제 9(Streams 집계) 입력/출력
  "clickstream:6:3:min.insync.replicas=2"
  "clicks-per-minute:6:3:min.insync.replicas=2"
  # 예제 4에서 컴팩션 토픽과 비교하기 위한 상태 토픽
  "product-catalog:3:3:cleanup.policy=compact,min.insync.replicas=2"
)

for spec in "${TOPICS[@]}"; do
  IFS=':' read -r name parts rf extra <<< "$spec"

  args=(--bootstrap-server "$BROKER_INTERNAL"
        --create --if-not-exists
        --topic "$name"
        --partitions "$parts"
        --replication-factor "$rf")

  # 추가 설정은 --config 를 항목마다 하나씩 붙여야 합니다.
  if [[ -n "${extra:-}" ]]; then
    IFS=',' read -ra kvs <<< "$extra"
    for kv in "${kvs[@]}"; do
      args+=(--config "$kv")
    done
  fi

  echo "==> create ${name} (partitions=${parts}, rf=${rf}, config=${extra:-none})"
  docker exec kafka-1 "${KAFKA_BIN}/kafka-topics.sh" "${args[@]}"
done

echo
echo '==> 최종 토픽 목록'
docker exec kafka-1 "${KAFKA_BIN}/kafka-topics.sh" \
  --bootstrap-server "$BROKER_INTERNAL" --list

#!/usr/bin/env bash
# 컨테이너 안의 Kafka CLI 를 호스트에서 실행하는 래퍼.
#
# 사용법:
#   ./kcli kafka-topics.sh --list
#   ./kcli kafka-consumer-groups.sh --describe --group order-service
#   ./kcli kafka-metadata-quorum.sh describe --status
#
# --bootstrap-server 를 생략하면 컨테이너 네트워크 주소를 자동으로 넣습니다.
set -euo pipefail

CONTAINER="${KCLI_CONTAINER:-kafka-1}"
BOOTSTRAP="${KCLI_BOOTSTRAP:-kafka-1:19092,kafka-2:19092,kafka-3:19092}"

if [[ $# -lt 1 ]]; then
  echo "usage: $0 <kafka-*.sh> [args...]" >&2
  exit 2
fi

script="$1"; shift

# 이미 --bootstrap-server 또는 --bootstrap-controller 가 있으면 그대로 존중합니다.
needs_bootstrap=1
for a in "$@"; do
  case "$a" in
    --bootstrap-server|--bootstrap-controller) needs_bootstrap=0 ;;
  esac
done

if [[ $needs_bootstrap -eq 1 ]]; then
  set -- --bootstrap-server "$BOOTSTRAP" "$@"
fi

# -i 만 주고 -t 는 주지 않습니다. 파이프로 넘길 때 TTY 가 있으면 깨집니다.
exec docker exec -i "$CONTAINER" "/opt/kafka/bin/${script}" "$@"

# 0. 작업 디렉터리 준비
mkdir -p kafka-lab && cd kafka-lab
# (위의 4개 파일을 이 디렉터리에 저장합니다)
chmod +x create-topics.sh kcli

# 1. (선택) 나만의 클러스터 ID 생성 → .env 의 CLUSTER_ID 에 붙여넣기
docker run --rm apache/kafka:4.3.1 /opt/kafka/bin/kafka-storage.sh random-uuid

# 2. 클러스터 기동. 로그를 보려면 -d 를 빼세요.
docker compose up -d

# 3. 세 노드가 healthy 가 될 때까지 대기 (보통 30~45초)
docker compose ps

# 4. 실습용 토픽 생성
./create-topics.sh

# 컨테이너만 정지 — 데이터(볼륨)는 남습니다. 다시 up 하면 토픽이 그대로 있습니다.
docker compose down

# 데이터까지 완전 초기화 — 볼륨을 지우므로 다음 기동 시 새 클러스터로 포맷됩니다.
docker compose down -v

./kcli kafka-metadata-quorum.sh describe --status

ClusterId:              4L6g3nShT-eMCtK--X86sw
LeaderId:               2
LeaderEpoch:            1
HighWatermark:          312
MaxFollowerLag:         0
MaxFollowerLagTimeMs:   0
CurrentVoters:          [{"id": 1, ...}, {"id": 2, ...}, {"id": 3, ...}]
CurrentObservers:       []

# 각 브로커가 어떤 주소로 자신을 광고하는지 그대로 보여 줍니다.
./kcli kafka-broker-api-versions.sh | grep -E '^kafka-[0-9]+:'

kafka-1:19092 (id: 1 rack: null isFenced: false) -> (
kafka-2:19092 (id: 2 rack: null isFenced: false) -> (
kafka-3:19092 (id: 3 rack: null isFenced: false) -> (

./kcli kafka-topics.sh --describe --topic orders

Topic: orders  TopicId: kkxLGvB1RCel...  PartitionCount: 6  ReplicationFactor: 3
	Configs: min.insync.replicas=2
	Topic: orders	Partition: 0	Leader: 1	Replicas: 1,2,3	Isr: 1,2,3
	Topic: orders	Partition: 1	Leader: 2	Replicas: 2,3,1	Isr: 2,3,1
	Topic: orders	Partition: 2	Leader: 3	Replicas: 3,1,2	Isr: 3,1,2
	Topic: orders	Partition: 3	Leader: 1	Replicas: 1,3,2	Isr: 1,3,2
	Topic: orders	Partition: 4	Leader: 2	Replicas: 2,1,3	Isr: 2,1,3
	Topic: orders	Partition: 5	Leader: 3	Replicas: 3,2,1	Isr: 3,2,1

# (A) 컨테이너 네트워크 경로로 3건 생산
printf 'a\nb\nc\n' | ./kcli kafka-console-producer.sh --topic orders

# (B) 컨테이너 네트워크 경로로 소비 — 3건이 보이면 성공
./kcli kafka-console-consumer.sh --topic orders --from-beginning \
  --timeout-ms 5000 --max-messages 3

# (C) 호스트 경로 확인이 이 예제의 진짜 관문입니다.
#     호스트에 Kafka tarball 이 없다면 컨테이너를 호스트 네트워크로 붙여 확인합니다.
docker run --rm --network host apache/kafka:4.3.1 \
  /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:29092 --list