#Step 01 — 클러스터 기동과 첫 메시지

cd kafka/docker
docker compose up -d

docker compose ps

docker compose logs kafka-1 | tail -20

docker exec kafka-1 /opt/kafka/bin/kafka-broker-api-versions.sh \
  --bootstrap-server kafka-1:9092 | grep -E '^kafka-[0-9]'

docker exec kafka-1 /opt/kafka/bin/kafka-metadata-quorum.sh \
  --bootstrap-server kafka-1:9092 describe --status

alias kt='docker exec kafka-1 /opt/kafka/bin/kafka-topics.sh      --bootstrap-server kafka-1:9092'
alias kcg='docker exec kafka-1 /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server kafka-1:9092'
alias kconf='docker exec kafka-1 /opt/kafka/bin/kafka-configs.sh  --bootstrap-server kafka-1:9092'
alias kgo='docker exec kafka-1 /opt/kafka/bin/kafka-get-offsets.sh --bootstrap-server kafka-1:9092'

kt --list

docker exec kafka-1 /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka-1:9092 \
  --create --topic orders \
  --partitions 3 \
  --replication-factor 3 \
  --config retention.ms=604800000 \
  --config min.insync.replicas=2

kt --create --topic bad-rf --partitions 1 --replication-factor 5

kt --describe --topic orders

docker exec -it kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 \
  --topic orders \
  --from-beginning

docker exec -it kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka-1:9092 \
  --topic orders

docker exec -it kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 \
  --topic orders

kgo --topic orders

docker exec -it kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 \
  --topic orders \
  --from-beginning \
  --property print.key=true \
  --property print.partition=true \
  --property print.offset=true \
  --property key.separator=' | '

docker exec -it kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka-1:9092 \
  --topic orders \
  --property parse.key=true \
  --property key.separator=:

docker exec -it kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 \
  --topic oders \
  --from-beginning

kconf --describe --entity-type brokers --entity-name 1 --all \
  | grep auto.create.topics.enable

# 방법 1 — 전용 옵션 (콘솔 프로듀서에만 있음)
docker exec -it kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka-1:9092 --topic orders \
  --request-required-acks 1

# 방법 2 — 임의의 프로듀서 설정을 넘김 (모든 설정에 사용 가능)
docker exec -it kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka-1:9092 --topic orders \
  --producer-property acks=1 \
  --producer-property compression.type=lz4 \
  --producer-property linger.ms=100

docker exec -it kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka-1:9092 --topic orders \
  --producer-property acks=0 2>&1 | grep -E 'acks|linger.ms'

# payments — orders 와 같은 구성. 키는 order_id
kt --create --topic payments \
  --partitions 3 --replication-factor 3 \
  --config retention.ms=604800000 \
  --config min.insync.replicas=2

# order-events — 압축(compact) 토픽. 키마다 "최신 값"만 남긴다
kt --create --topic order-events \
  --partitions 3 --replication-factor 3 \
  --config cleanup.policy=compact \
  --config min.insync.replicas=2

# dlq — 처리 실패 메시지. 순서가 중요하므로 파티션 1개, 보존 30일
kt --create --topic dlq \
  --partitions 1 --replication-factor 3 \
  --config retention.ms=2592000000 \
  --config min.insync.replicas=2

kt --list

kt --describe --topic order-events

kt --describe --topics-with-overrides

docker exec kafka-1 ls /opt/kafka/bin/ | grep '^kafka-' | head -30

docker exec kafka-1 cat /opt/kafka/bin/kafka-topics.sh

kt --list

kt --delete --topic bad-rf

kgo --topic orders

#!/usr/bin/env bash
# =============================================================================
# Step 01 — 클러스터 기동과 첫 메시지 : 실습 스크립트
#
# 실행법:
#   bash step-01-setup/practice.sh          # 통째로 실행
#   bash -x step-01-setup/practice.sh       # 한 줄씩 에코하며 실행 (권장)
#
# 전제:
#   kafka/docker 디렉터리에서 docker compose up -d 가 가능한 상태
#
# 주의:
#   [터미널 B] 주석이 붙은 구간은 대화형이라 이 스크립트에서 실행되지 않습니다.
#   해당 명령은 별도 창에서 직접 실행하세요. 스크립트에는 같은 것을 자동으로
#   확인할 수 있는 비대화형 버전(--timeout-ms / 파이프 입력)을 넣어 두었습니다.
# =============================================================================
set -euo pipefail

BS=kafka-1:9092
K() { docker exec kafka-1 /opt/kafka/bin/"$@"; }
Ki() { docker exec -i kafka-1 /opt/kafka/bin/"$@"; }   # 표준입력을 넘길 때

hr() { echo; echo "=== $* ==================================================="; }

# -----------------------------------------------------------------------------
# [1-0] 클러스터 기동과 healthy 대기
# -----------------------------------------------------------------------------
hr "[1-0] 클러스터 기동"

# compose 파일 위치. 이 스크립트를 리포지토리 어디서 실행하든 동작하도록 계산합니다.
COMPOSE_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../docker" && pwd)"
echo "compose dir: $COMPOSE_DIR"

docker compose -f "$COMPOSE_DIR/docker-compose.yml" up -d

wait_healthy() {
  # 세 브로커가 모두 (healthy) 가 될 때까지 최대 120초 대기.
  # 이 대기 없이 바로 토픽을 만들면 Broker may not be available 로 실패합니다.
  local i cnt
  for i in $(seq 1 24); do
    cnt=$(docker compose -f "$COMPOSE_DIR/docker-compose.yml" ps | grep -c '(healthy)' || true)
    echo "  healthy brokers: $cnt/3  (${i}/24)"
    [ "$cnt" -ge 3 ] && return 0
    sleep 5
  done
  echo "브로커가 healthy 가 되지 않았습니다. docker compose logs kafka-1 을 확인하세요." >&2
  return 1
}
wait_healthy

docker compose -f "$COMPOSE_DIR/docker-compose.yml" ps

# -----------------------------------------------------------------------------
# [1-1] 브로커 3대 인식 확인 + KRaft 쿼럼 상태
# -----------------------------------------------------------------------------
hr "[1-1] 브로커 3대 확인"

# 세 줄이 나와야 정상입니다. 한 줄만 나오면 쿼럼 구성이 잘못된 것입니다.
K kafka-broker-api-versions.sh --bootstrap-server "$BS" | grep -E '^kafka-[0-9]'

echo "--- KRaft quorum status ---"
K kafka-metadata-quorum.sh --bootstrap-server "$BS" describe --status

# -----------------------------------------------------------------------------
# [1-2] 편의 별칭 — 스크립트에서는 alias 대신 함수(K)를 씁니다
# -----------------------------------------------------------------------------
hr "[1-2] 토픽 목록 (아직 비어 있음)"

K kafka-topics.sh --bootstrap-server "$BS" --list

# 참고: 대화형 셸에서는 아래 별칭을 쓰세요.
#   alias kt='docker exec kafka-1 /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka-1:9092'
#   alias kcg='docker exec kafka-1 /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server kafka-1:9092'
#   alias kconf='docker exec kafka-1 /opt/kafka/bin/kafka-configs.sh --bootstrap-server kafka-1:9092'
#   alias kgo='docker exec kafka-1 /opt/kafka/bin/kafka-get-offsets.sh --bootstrap-server kafka-1:9092'

# -----------------------------------------------------------------------------
# [1-3] 첫 토픽 orders 만들기
# -----------------------------------------------------------------------------
hr "[1-3] orders 토픽 생성"

# || true : 이미 있으면 TopicExistsException 이 나는데, set -e 로 죽지 않게 합니다.
#           이 스크립트를 몇 번을 다시 돌려도 안전하도록(idempotent) 만든 것입니다.
K kafka-topics.sh --bootstrap-server "$BS" \
  --create --topic orders \
  --partitions 3 \
  --replication-factor 3 \
  --config retention.ms=604800000 \
  --config min.insync.replicas=2 || true

echo "--- 브로커 수보다 큰 RF 는 즉시 거절됩니다 (일부러 실패시키는 명령) ---"
K kafka-topics.sh --bootstrap-server "$BS" \
  --create --topic bad-rf --partitions 1 --replication-factor 5 || true

# -----------------------------------------------------------------------------
# [1-4] --describe 출력 해독
# -----------------------------------------------------------------------------
hr "[1-4] orders --describe"

K kafka-topics.sh --bootstrap-server "$BS" --describe --topic orders

echo "--- 운영에서 가장 많이 치는 두 변형 (정상이면 아무 출력도 없습니다) ---"
K kafka-topics.sh --bootstrap-server "$BS" --describe --under-replicated-partitions
K kafka-topics.sh --bootstrap-server "$BS" --describe --unavailable-partitions
echo "(위 두 명령의 출력이 비어 있으면 정상입니다)"

# -----------------------------------------------------------------------------
# [1-5] 첫 메시지 — 키 없이
# -----------------------------------------------------------------------------
hr "[1-5] 키 없는 메시지 3건 전송"

# [터미널 A] 대화형 컨슈머 (스크립트에서는 실행하지 않습니다)
#   docker exec -it kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
#     --bootstrap-server kafka-1:9092 --topic orders --from-beginning
#
# [터미널 B] 대화형 프로듀서 (스크립트에서는 실행하지 않습니다)
#   docker exec -it kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
#     --bootstrap-server kafka-1:9092 --topic orders

# 비대화형 대체: 표준 입력을 파이프로 밀어 넣습니다.
printf '%s\n' \
  '{"order_id":"O-1001","customer_id":"C001","amount":39000,"status":"CREATED"}' \
  '{"order_id":"O-1002","customer_id":"C002","amount":15000,"status":"CREATED"}' \
  '{"order_id":"O-1003","customer_id":"C003","amount":72000,"status":"CREATED"}' \
| Ki kafka-console-producer.sh --bootstrap-server "$BS" --topic orders

echo "--- --from-beginning 으로 읽기 (--timeout-ms 로 자동 종료) ---"
# --timeout-ms : 그 시간 동안 새 메시지가 없으면 컨슈머를 끝냅니다.
#                스크립트에서 콘솔 컨슈머를 쓰는 유일하게 안전한 방법입니다.
#                종료 시 TimeoutException 을 던지므로 || true 로 받습니다.
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic orders \
  --from-beginning --timeout-ms 5000 2>&1 | grep -v TimeoutException || true

# -----------------------------------------------------------------------------
# [1-6] 함정 ① --from-beginning 이 없으면 아무것도 안 읽힙니다
# -----------------------------------------------------------------------------
hr "[1-6] 함정 --from-beginning 유무 비교"

echo "--- (A) --from-beginning 있음 ---"
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic orders \
  --from-beginning --timeout-ms 5000 2>&1 | grep 'Processed a total' || true

echo "--- (B) --from-beginning 없음  → Processed a total of 0 messages 가 나옵니다 ---"
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic orders \
  --timeout-ms 5000 2>&1 | grep 'Processed a total' || true

echo "--- 데이터가 없는 것이 아니라 안 읽는 것임을 증명 ---"
K kafka-get-offsets.sh --bootstrap-server "$BS" --topic orders

# -----------------------------------------------------------------------------
# [1-7] 키 있는 메시지 — 같은 키는 같은 파티션
# -----------------------------------------------------------------------------
hr "[1-7] 키 있는 메시지 4건 전송"

printf '%s\n' \
  'C001:{"order_id":"O-1004","customer_id":"C001","amount":21000,"status":"CREATED"}' \
  'C001:{"order_id":"O-1005","customer_id":"C001","amount":33000,"status":"PAID"}' \
  'C006:{"order_id":"O-1006","customer_id":"C006","amount":58000,"status":"CREATED"}' \
  'C002:{"order_id":"O-1007","customer_id":"C002","amount":12000,"status":"CREATED"}' \
| Ki kafka-console-producer.sh --bootstrap-server "$BS" --topic orders \
    --property parse.key=true \
    --property key.separator=:

echo "--- 키/파티션/오프셋을 전부 찍어서 확인 ---"
# C001 두 건이 같은 파티션(2)에 들어갔는지 보세요.
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic orders \
  --from-beginning --timeout-ms 5000 \
  --property print.key=true \
  --property print.partition=true \
  --property print.offset=true \
  --property key.separator=' | ' 2>&1 | grep -v TimeoutException || true

# -----------------------------------------------------------------------------
# [1-8] 함정 ② 오타 난 토픽 이름
# -----------------------------------------------------------------------------
hr "[1-8] 함정 오타 토픽 (auto.create.topics.enable=false)"

# 기본 타임아웃이 60초라 스크립트가 1분 멈춥니다. 5초로 줄입니다.
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic oders \
  --from-beginning --timeout-ms 5000 \
  --consumer-property default.api.timeout.ms=5000 2>&1 | tail -5 || true

echo "--- 브로커의 auto.create.topics.enable 확인 (synonyms 로 출처를 봅니다) ---"
K kafka-configs.sh --bootstrap-server "$BS" \
  --describe --entity-type brokers --entity-name 1 --all \
  | grep auto.create.topics.enable

# -----------------------------------------------------------------------------
# [1-9] 콘솔 프로듀서의 acks — 3.7 기본값은 -1(all)
# -----------------------------------------------------------------------------
hr "[1-9] --producer-property 로 설정을 바꾸고 실제 적용 여부 확인"

# 프로듀서는 기동 시 ProducerConfig values: 로 전체 설정을 덤프합니다.
# "내 설정이 진짜 먹었나"를 확인하는 가장 확실한 방법입니다.
printf '%s\n' 'C001:{"order_id":"O-1008","customer_id":"C001","amount":9000,"status":"CREATED"}' \
| Ki kafka-console-producer.sh --bootstrap-server "$BS" --topic orders \
    --property parse.key=true --property key.separator=: \
    --producer-property acks=0 \
    --producer-property linger.ms=100 2>&1 \
| grep -E '^\s+(acks|linger\.ms|compression\.type) ' || true

# -----------------------------------------------------------------------------
# [1-10] 나머지 코스 토픽 생성
# -----------------------------------------------------------------------------
hr "[1-10] payments / order-events / dlq 생성"

K kafka-topics.sh --bootstrap-server "$BS" \
  --create --topic payments --partitions 3 --replication-factor 3 \
  --config retention.ms=604800000 --config min.insync.replicas=2 || true

# order-events 는 compact. 키마다 최신 값 하나만 남습니다 (Step 09).
K kafka-topics.sh --bootstrap-server "$BS" \
  --create --topic order-events --partitions 3 --replication-factor 3 \
  --config cleanup.policy=compact --config min.insync.replicas=2 || true

# dlq 는 순서가 중요하므로 파티션 1개, 보존 30일.
K kafka-topics.sh --bootstrap-server "$BS" \
  --create --topic dlq --partitions 1 --replication-factor 3 \
  --config retention.ms=2592000000 --config min.insync.replicas=2 || true

K kafka-topics.sh --bootstrap-server "$BS" --list

echo "--- compact 토픽 확인 (Configs 에 cleanup.policy=compact) ---"
K kafka-topics.sh --bootstrap-server "$BS" --describe --topic order-events

echo "--- 브로커 기본값과 다른 설정을 가진 토픽만 ---"
K kafka-topics.sh --bootstrap-server "$BS" --describe --topics-with-overrides

# -----------------------------------------------------------------------------
# [1-11] kafka-ui — 브라우저로 확인 (스크립트에서 할 일 없음)
# -----------------------------------------------------------------------------
hr "[1-11] kafka-ui"
echo "브라우저에서 http://localhost:8080 을 열고 Topics → orders → Messages 를 확인하세요."
echo "위 [1-7] 의 CLI 출력과 정확히 같은 Partition/Offset/Key 가 보여야 합니다."

# -----------------------------------------------------------------------------
# [1-12] CLI 도구 지도
# -----------------------------------------------------------------------------
hr "[1-12] /opt/kafka/bin 의 도구 목록"

K bash -c 'ls /opt/kafka/bin/ | grep "^kafka-" | head -30' 2>/dev/null \
  || docker exec kafka-1 sh -c 'ls /opt/kafka/bin/ | grep "^kafka-" | head -30'

echo "--- kafka-*.sh 는 전부 kafka-run-class.sh 의 얇은 래퍼입니다 ---"
docker exec kafka-1 tail -1 /opt/kafka/bin/kafka-topics.sh

# -----------------------------------------------------------------------------
# [1-13] 정리 — 토픽은 지우지 않습니다. Step 02 가 이 데이터를 씁니다.
# -----------------------------------------------------------------------------
hr "[1-13] 정리 및 상태 검증"

echo "--- 토픽은 정확히 4개여야 합니다 ---"
K kafka-topics.sh --bootstrap-server "$BS" --list

echo "--- orders 의 파티션별 메시지 수 (합계 8건: 1-5 의 3건 + 1-7 의 4건 + 1-9 의 1건) ---"
K kafka-get-offsets.sh --bootstrap-server "$BS" --topic orders

echo
echo "Step 01 완료. orders 의 메시지는 Step 02 에서 세그먼트 파일로 직접 열어 봅니다."
echo "지우지 마세요."

#!/usr/bin/env bash
# =============================================================================
# Step 01 — 연습문제 (6문제)
#
# 실행법:
#   bash step-01-setup/exercise.sh
#
# 각 문제의 "# 여기에 작성:" 아래에 명령을 채워 넣고 실행하세요.
# 정답은 solution.sh 에 있습니다. 먼저 직접 풀어 보세요.
#
# 전제: practice.sh 를 한 번 실행해 orders/payments/order-events/dlq 가 있는 상태
# =============================================================================
set -euo pipefail

BS=kafka-1:9092
K() { docker exec kafka-1 /opt/kafka/bin/"$@"; }
Ki() { docker exec -i kafka-1 /opt/kafka/bin/"$@"; }

hr() { echo; echo "=== $* ==================================================="; }

# -----------------------------------------------------------------------------
# 문제 1. 토픽 만들기
#
#   s01_practice 토픽을 아래 조건으로 만드세요.
#     - 파티션 4개
#     - replication factor 2   ← 코스 공용 토픽(RF 3)과 다릅니다
#     - 보존 기간 1시간
#   만든 뒤 --describe 로 조건이 전부 반영되었는지 검증하세요.
#
#   힌트: retention.ms 는 밀리초입니다. 1시간이 몇 밀리초인지 계산하세요.
#   힌트: --config 는 여러 번 쓸 수 있습니다.
# -----------------------------------------------------------------------------
hr "문제 1 — s01_practice 토픽 생성"

# 여기에 작성:


# 검증 (작성한 명령이 맞으면 PartitionCount: 4, ReplicationFactor: 2 가 보입니다)
# 여기에 작성:


# -----------------------------------------------------------------------------
# 문제 2. 리더가 가장 많은 브로커 찾기
#
#   모든 토픽의 --describe 출력에서 "Leader: N" 을 세어, 어느 브로커가
#   파티션 리더를 가장 많이 맡고 있는지 구하세요.
#
#   힌트: grep -oE 'Leader: [0-9]+' | sort | uniq -c
#   참고: 파티션 배치는 라운드로빈이라 보통 균등하게 나옵니다.
#         직전에 브로커를 재시작했다면 한쪽으로 쏠려 있을 수도 있는데 그것도 정상입니다.
# -----------------------------------------------------------------------------
hr "문제 2 — 브로커별 리더 파티션 수"

# 여기에 작성:


# -----------------------------------------------------------------------------
# 문제 3. --from-beginning 이 있고 없고의 차이를 숫자로
#
#   orders 토픽을 두 번 읽되, 한 번은 --from-beginning 을 주고 한 번은 주지 마세요.
#   각각의 "Processed a total of N messages" 를 출력해 N 을 비교하세요.
#
#   힌트: 스크립트에서 콘솔 컨슈머를 끝내려면 --timeout-ms 5000 이 필요합니다.
#   힌트: 출력이 많으니 | grep 'Processed a total' 로 걸러 내세요.
#   질문: 두 실행이 같은 컨슈머 그룹을 쓰고 있습니까? 왜 그것이 중요합니까?
# -----------------------------------------------------------------------------
hr "문제 3 — --from-beginning 유무 비교"

echo "--- (A) --from-beginning 있음 ---"
# 여기에 작성:


echo "--- (B) --from-beginning 없음 ---"
# 여기에 작성:


# -----------------------------------------------------------------------------
# 문제 4. 같은 키는 같은 파티션으로 간다
#
#   orders 토픽에 키 C007 로 메시지 3건을 보내고,
#   세 건이 모두 같은 파티션에 들어갔음을 증명하세요.
#
#   힌트: 프로듀서에 --property parse.key=true --property key.separator=:
#   힌트: 컨슈머에 --property print.key=true --property print.partition=true
#   힌트: C007 만 보려면 | grep C007
# -----------------------------------------------------------------------------
hr "문제 4 — C007 키 3건의 파티션 확인"

echo "--- 전송 ---"
# 여기에 작성:


echo "--- 확인 ---"
# 여기에 작성:


# -----------------------------------------------------------------------------
# 문제 5. 없는 토픽에 붙었을 때 — 프로듀서와 컨슈머의 차이
#
#   존재하지 않는 토픽 이름(oders)으로 프로듀서를 실행해 메시지 한 건을 보내 보세요.
#   그리고 같은 토픽 이름으로 컨슈머도 실행해 보세요.
#
#   두 에러 메시지를 비교하고, 아래 질문에 답하세요.
#     Q1. 프로듀서는 언제 실패합니까? 실행하자마자입니까, 메시지를 보낼 때입니까?
#     Q2. 그 차이가 실무에서 왜 위험합니까?
#
#   힌트: 타임아웃을 줄이려면
#         프로듀서: --producer-property max.block.ms=5000
#         컨슈머:   --consumer-property default.api.timeout.ms=5000
#   주의: 실패하는 명령이므로 || true 로 감싸야 set -e 에 걸리지 않습니다.
# -----------------------------------------------------------------------------
hr "문제 5 — 없는 토픽: 프로듀서 vs 컨슈머"

echo "--- 프로듀서 ---"
# 여기에 작성:


echo "--- 컨슈머 ---"
# 여기에 작성:


# 여기에 답 작성:
#   A1.
#   A2.


# -----------------------------------------------------------------------------
# 문제 6. 설정이 진짜 먹었는지 확인하기
#
#   콘솔 프로듀서를 아래 두 설정으로 실행하고, 기동 로그에서 두 값이 실제로
#   적용되었는지 확인하세요.
#       acks=0
#       compresion.type=gzip      ← 이 줄은 그대로 두세요. 일부러 이렇게 썼습니다.
#
#   힌트: 프로듀서는 기동 시 "ProducerConfig values:" 블록으로 전체 설정을 덤프합니다.
#   힌트: grep -E '^\s+(acks|compression\.type) ' 로 두 줄만 뽑으세요.
#   힌트: 2>&1 로 표준 에러를 합쳐야 로그가 잡힙니다.
#
#   질문: compression.type 이 gzip 으로 나옵니까? 안 나온다면 왜입니까?
#         그때 Kafka 는 에러를 냅니까, 경고를 냅니까, 아무 말도 안 합니까?
# -----------------------------------------------------------------------------
hr "문제 6 — --producer-property 적용 확인"

# 여기에 작성:


# 여기에 답 작성:
#   A.


# -----------------------------------------------------------------------------
# 뒷정리 — 문제 1에서 만든 토픽을 지웁니다
# -----------------------------------------------------------------------------
hr "뒷정리"

K kafka-topics.sh --bootstrap-server "$BS" --delete --topic s01_practice || true

echo "--- 남아 있어야 할 토픽: dlq / order-events / orders / payments 네 개 ---"
K kafka-topics.sh --bootstrap-server "$BS" --list

#!/usr/bin/env bash
# =============================================================================
# Step 01 — 연습문제 정답과 해설 (6문제)
#
# 실행법:
#   bash step-01-setup/solution.sh
#
# 각 정답 아래의 "# 해설:" 블록이 "왜 그 답인가"를 설명합니다.
# exercise.sh 를 직접 풀어 본 뒤에 여세요.
# =============================================================================
set -euo pipefail

BS=kafka-1:9092
K() { docker exec kafka-1 /opt/kafka/bin/"$@"; }
Ki() { docker exec -i kafka-1 /opt/kafka/bin/"$@"; }

hr() { echo; echo "=== $* ==================================================="; }

# =============================================================================
# 정답 1. s01_practice 토픽 생성
# =============================================================================
hr "정답 1 — s01_practice 토픽 생성"

K kafka-topics.sh --bootstrap-server "$BS" \
  --create --topic s01_practice \
  --partitions 4 \
  --replication-factor 2 \
  --config retention.ms=3600000 || true

K kafka-topics.sh --bootstrap-server "$BS" --describe --topic s01_practice

# 해설:
#   retention.ms 는 밀리초 단위입니다. 1시간 = 60 * 60 * 1000 = 3600000.
#   자주 쓰는 값을 외워 두면 편합니다.
#       1시간   =     3600000
#       1일     =    86400000
#       7일     =   604800000   ← orders / payments 의 기본
#       30일    =  2592000000   ← dlq
#       무제한  =          -1
#
#   --config 는 몇 번이든 반복할 수 있습니다.
#       --config retention.ms=3600000 --config segment.bytes=1048576 ...
#
#   RF 2 의 의미:
#     Replicas 에 브로커가 2개만 나오고 Isr 도 2개입니다.
#     브로커 1대가 죽으면 Isr 이 1이 되어 데이터는 살아 있지만,
#     min.insync.replicas=2 인 토픽이라면 그 순간 acks=all 쓰기가 전부 실패합니다.
#     "RF 3 + min.insync.replicas=2" 조합이 표준인 이유입니다.
#     브로커 1대 손실을 견디면서도 쓰기가 계속되기 때문입니다 (Step 08).
#
#   참고로 이 토픽은 --config min.insync.replicas 를 주지 않았으므로
#   브로커 기본값 2가 적용됩니다. --describe 의 Configs 에는 나오지 않습니다.
#   Configs 는 "토픽 레벨로 명시된 것"만 보여 주기 때문입니다.

# =============================================================================
# 정답 2. 브로커별 리더 파티션 수
# =============================================================================
hr "정답 2 — 브로커별 리더 파티션 수"

K kafka-topics.sh --bootstrap-server "$BS" --describe \
  | grep -oE 'Leader: [0-9]+' | sort | uniq -c | sort -rn

# 해설:
#   기대 출력 (토픽 4개 = 파티션 3+3+3+1 = 10개 + s01_practice 4개 = 14개):
#       5 Leader: 1
#       5 Leader: 2
#       4 Leader: 3
#   숫자는 파티션 배치에 따라 조금씩 다를 수 있습니다.
#
#   왜 세어 봅니까:
#     모든 읽기·쓰기는 리더가 처리합니다. 팔로워는 복제만 합니다.
#     따라서 리더가 한 브로커에 몰리면 그 브로커만 CPU·네트워크·디스크를 쓰고
#     나머지 둘은 놉니다. 클러스터 전체 처리량이 1/3 로 떨어질 수 있습니다.
#
#   언제 쏠립니까:
#     브로커를 재시작하면 그 브로커가 리더였던 파티션들이 다른 브로커로 넘어가고,
#     되살아나도 리더가 자동으로 돌아오지 않습니다(팔로워로 복귀합니다).
#     롤링 재시작을 3대 다 하고 나면 리더가 마지막에 산 브로커를 피해 몰려 있습니다.
#
#   되돌리는 법 (Step 08, 14):
#       kafka-leader-election.sh --bootstrap-server kafka-1:9092 \
#         --election-type preferred --all-topic-partitions
#     "preferred leader" = --describe 의 Replicas 목록 맨 앞 브로커입니다.
#     브로커 설정 auto.leader.rebalance.enable=true(기본값) 면 주기적으로
#     자동 수행되지만, 즉시 필요하면 위 명령을 칩니다.

# =============================================================================
# 정답 3. --from-beginning 유무 비교
# =============================================================================
hr "정답 3 — --from-beginning 유무 비교"

echo "--- (A) --from-beginning 있음 ---"
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic orders \
  --from-beginning --timeout-ms 5000 2>&1 | grep 'Processed a total' || true

echo "--- (B) --from-beginning 없음 ---"
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic orders \
  --timeout-ms 5000 2>&1 | grep 'Processed a total' || true

# 해설:
#   기대 출력:
#       (A) Processed a total of 8 messages
#       (B) Processed a total of 0 messages
#
#   두 실행은 서로 다른 컨슈머 그룹을 씁니다. 이것이 이 문제의 핵심입니다.
#   kafka-console-consumer.sh 는 --group 을 안 주면 실행할 때마다
#   console-consumer-XXXXX (XXXXX 는 난수) 라는 새 그룹을 만듭니다.
#   실제로 쌓여 있는지 확인해 보세요:
#       kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 --list | grep console-consumer
#
#   새 그룹에는 커밋된 오프셋이 없습니다. 그럴 때 어디서부터 읽을지는
#   auto.offset.reset 이 정하고, 기본값은 latest 입니다.
#   latest = "지금 이 순간 이후에 도착하는 것부터". 이미 있는 8건은 건너뜁니다.
#
#   --from-beginning 은 내부적으로 auto.offset.reset=earliest 를 설정합니다.
#
#   왜 위험한가:
#     증상이 "아무 일도 안 일어남"입니다. 에러도, 경고도, 로그도 없습니다.
#     토픽 이름을 의심하고 방화벽을 의심하다가 마지막에야 이 옵션을 떠올립니다.
#
#   더 큰 함정 (Step 06):
#     같은 --group 을 지정해 두 번째로 실행하면 --from-beginning 을 줘도
#     아무것도 안 읽힙니다. 커밋된 오프셋이 이미 있으면 auto.offset.reset 은
#     아예 참조되지 않기 때문입니다. 그때는 --reset-offsets 를 써야 합니다.

# =============================================================================
# 정답 4. C007 키 3건은 모두 같은 파티션
# =============================================================================
hr "정답 4 — C007 키 3건의 파티션"

printf '%s\n' \
  'C007:{"order_id":"O-2001","customer_id":"C007","amount":11000,"status":"CREATED"}' \
  'C007:{"order_id":"O-2002","customer_id":"C007","amount":22000,"status":"PAID"}' \
  'C007:{"order_id":"O-2003","customer_id":"C007","amount":33000,"status":"SHIPPED"}' \
| Ki kafka-console-producer.sh --bootstrap-server "$BS" --topic orders \
    --property parse.key=true --property key.separator=:

K kafka-console-consumer.sh --bootstrap-server "$BS" --topic orders \
  --from-beginning --timeout-ms 5000 \
  --property print.key=true --property print.partition=true --property print.offset=true \
  2>&1 | grep C007 || true

# 해설:
#   기대 출력 — 세 줄 모두 Partition:1 입니다.
#       Partition:1	Offset:2	C007	{"order_id":"O-2001",...}
#       Partition:1	Offset:3	C007	{"order_id":"O-2002",...}
#       Partition:1	Offset:4	C007	{"order_id":"O-2003",...}
#
#   왜 하필 1번입니까? 우연이 아니라 계산 결과입니다.
#   Kafka 의 기본 파티셔너(키가 있을 때)는 이렇게 동작합니다.
#
#       partition = (murmur2(keyBytes) & 0x7fffffff) % numPartitions
#
#   "C007" 의 실제 값:
#       murmur2("C007")            = 642592957
#       & 0x7fffffff (양수화)      = 642592957
#       % 3                        = 1        ← 파티션 1
#
#   & 0x7fffffff 는 최상위 부호 비트를 지워 음수를 양수로 만듭니다.
#   음수에 % 를 하면 Java 에서도 음수가 나와 파티션 번호로 쓸 수 없기 때문입니다.
#
#   여기서 나오는 결론 두 가지:
#     (1) 같은 키는 항상 같은 파티션 → 그 키의 메시지 순서가 보장됩니다.
#     (2) 계산식에 numPartitions 가 들어갑니다. 파티션 수가 바뀌면 결과도 바뀝니다.
#         이것이 Step 03 의 핵심 함정입니다. 3 → 6 으로 늘리면
#         642592957 % 6 = 1 이라 C007 은 마침 그대로지만, C001 같은 키는
#         파티션 2 에서 5 로 이동합니다. 그 순간부터 순서 보장이 깨집니다.

# =============================================================================
# 정답 5. 없는 토픽 — 프로듀서와 컨슈머의 차이
# =============================================================================
hr "정답 5 — 없는 토픽: 프로듀서 vs 컨슈머"

echo "--- 프로듀서 (max.block.ms=5000 으로 대기 단축) ---"
printf '%s\n' 'hello' \
| Ki kafka-console-producer.sh --bootstrap-server "$BS" --topic oders \
    --producer-property max.block.ms=5000 2>&1 | tail -4 || true

echo "--- 컨슈머 ---"
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic oders \
  --from-beginning --timeout-ms 5000 \
  --consumer-property default.api.timeout.ms=5000 2>&1 | tail -4 || true

# 해설:
#   프로듀서:
#       org.apache.kafka.common.errors.TimeoutException:
#         Topic oders not present in metadata after 5000 ms.
#     단, 이 예외는 "첫 메시지를 보낼 때" 납니다. 프로듀서를 실행하면
#     ">" 프롬프트가 정상적으로 뜹니다. 에러가 없습니다.
#
#   컨슈머:
#       WARN ... Error while fetching metadata with correlation id 2 :
#         {oders=UNKNOWN_TOPIC_OR_PARTITION}
#       org.apache.kafka.common.errors.TimeoutException:
#         Topic oders not present in metadata after 5000 ms.
#     컨슈머는 붙자마자 메타데이터를 조회하므로 WARN 이 즉시 반복됩니다.
#
#   A1. 프로듀서는 실행 직후가 아니라 "첫 메시지를 보낼 때" 실패합니다.
#       Kafka 프로듀서는 lazy 하게 동작합니다. send() 를 호출해야 그 토픽의
#       메타데이터를 요청하고, 없으면 max.block.ms(기본 60000) 동안 기다립니다.
#
#   A2. 왜 위험한가:
#       (a) 프로듀서를 켰는데 프롬프트가 떠서 "잘 붙었다"고 착각합니다.
#       (b) 애플리케이션이라면 더 나쁩니다. 스프링 부트가 정상 기동하고
#           헬스체크도 통과합니다. 실제 트래픽이 들어오는 첫 순간에야 터집니다.
#           배포는 성공했는데 5분 뒤 장애 알림이 오는 전형적인 패턴입니다.
#       (c) auto.create.topics.enable=true 인 클러스터라면 최악입니다.
#           오타 토픽 oders 가 그 자리에서 생기고, 메시지가 거기 쌓이고,
#           전송은 "성공"합니다. 아무도 에러를 못 봅니다.
#           컨슈머는 orders 를 보고 있으니 영원히 아무것도 못 받습니다.
#
#   대비책:
#     애플리케이션 기동 시 AdminClient.describeTopics() 로 필요한 토픽이
#     존재하는지 검증하고, 없으면 기동을 실패시키세요.
#     "빨리, 시끄럽게 실패하는 것"이 항상 낫습니다.

# =============================================================================
# 정답 6. 설정 오타는 에러가 아니라 WARN 한 줄
# =============================================================================
hr "정답 6 — --producer-property 적용 확인"

printf '%s\n' 'C001:{"order_id":"O-2004","customer_id":"C001","amount":1000,"status":"CREATED"}' \
| Ki kafka-console-producer.sh --bootstrap-server "$BS" --topic orders \
    --property parse.key=true --property key.separator=: \
    --producer-property acks=0 \
    --producer-property compresion.type=gzip 2>&1 \
| grep -E "(^\s+(acks|compression\.type) |isn't a known config)" || true

# 해설:
#   기대 출력:
#       	acks = 0
#       	compression.type = none
#       WARN The configuration 'compresion.type' was supplied but isn't a known config.
#
#   acks 는 0 으로 잘 적용되었습니다. 그런데 compression.type 은 gzip 이 아니라
#   none 입니다. 문제지의 키가 compresion.type (s 하나 빠짐) 이었기 때문입니다.
#
#   여기서 Kafka 가 한 일:
#     - 에러를 내지 않았습니다.
#     - 기동을 멈추지 않았습니다.
#     - WARN 한 줄을 뱉고, 그 설정을 그냥 무시했습니다.
#     - 메시지는 압축 없이 정상 전송되었습니다.
#
#   이것이 이 코스가 다루는 "조용히 틀리는" 사례의 첫 번째입니다.
#   프로듀서 설정 오타는 대부분 이렇게 지나갑니다. 실무에서 흔한 것들:
#       max.in.flight.request.per.connection   (requests 의 s 누락)
#       compression.codec                      (Kafka 0.8 시절 이름)
#       bootstrap.server                       (servers 의 s 누락)
#   전부 WARN 한 줄로 무시되고, 여러분은 "설정했는데 왜 효과가 없지?" 를 헤맵니다.
#
#   대비책:
#     기동 로그의 "ProducerConfig values:" 블록을 실제로 읽으세요.
#     그리고 "isn't a known config" 를 로그 알림의 검색 조건에 넣으세요.
#     한 줄이지만 반드시 잡아야 하는 WARN 입니다.

# =============================================================================
# 뒷정리
# =============================================================================
hr "뒷정리"

K kafka-topics.sh --bootstrap-server "$BS" --delete --topic s01_practice || true

echo "--- 남아 있어야 할 토픽: dlq / order-events / orders / payments ---"
K kafka-topics.sh --bootstrap-server "$BS" --list

echo "--- orders 파티션별 오프셋 (문제 4·6 에서 4건이 추가되었습니다) ---"
K kafka-get-offsets.sh --bootstrap-server "$BS" --topic orders

echo
echo "orders 의 메시지는 지우지 마세요. Step 02 에서 세그먼트 파일로 직접 열어 봅니다."