#Step 05 — 컨슈머와 컨슈머 그룹

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'

kt --create --topic s05_orders  --partitions 3 --replication-factor 3
kt --create --topic s05_payments --partitions 3 --replication-factor 3

docker exec kafka-1 bash -c '
for i in $(seq 1 300); do
  k="C$(printf "%03d" $((RANDOM % 10 + 1)))"
  echo "$k:{\"order_id\":\"O-$((1000+i))\",\"customer_id\":\"$k\",\"amount\":$((RANDOM % 90000 + 1000)),\"status\":\"CREATED\"}"
done | /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server kafka-1:9092 \
  --topic s05_orders --property parse.key=true --property key.separator=:'

docker exec kafka-1 /opt/kafka/bin/kafka-get-offsets.sh \
  --bootstrap-server kafka-1:9092 --topic s05_orders

kcg --describe --group s05-demo --state

# [터미널 A]
docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 --topic s05_orders \
  --group s05-demo --from-beginning --max-messages 120

# [터미널 B]
kcg --describe --group s05-demo

# [터미널 A]
docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 --topic s05_orders \
  --group s05-demo --timeout-ms 10000 > /dev/null

kcg --describe --group s05-demo

# [터미널 A] [터미널 B] [터미널 C] — 셋 다 실행
docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 --topic s05_orders --group s05-demo

kcg --list

kcg --describe --group s05-demo --members

kcg --describe --group s05-demo --members --verbose

kcg --describe --group s05-demo --state

# [터미널 A] [터미널 B] [터미널 C] [터미널 D] — 넷 다 실행
docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 --topic s05_orders --group s05-four

kcg --describe --group s05-four --members --verbose

kcg --describe --group s05-four

# [터미널 E]
while true; do
  date '+%H:%M:%S'
  docker exec kafka-1 /opt/kafka/bin/kafka-consumer-groups.sh \
    --bootstrap-server kafka-1:9092 --describe --group s05-four --members --verbose \
    2>/dev/null | tail -n +2
  echo '---'
  sleep 1
done

# [터미널 A] [터미널 B]
docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 --include 's05_(orders|payments)' \
  --group s05-range

kcg --describe --group s05-range --members --verbose

# [터미널 A] [터미널 B] — 기존 컨슈머를 죽이고 다시
docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 --include 's05_(orders|payments)' \
  --group s05-rr \
  --consumer-property partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor

kcg --describe --group s05-rr --members --verbose

kcg --describe --group s05-rr --state

# [터미널 A] [터미널 B] [터미널 C]
docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 --topic s05_orders --group s05-coop \
  --consumer-property partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor

kcg --describe --group s05-coop --state

docker cp Practice.java kafka-1:/tmp/
docker exec kafka-1 sh -c 'cd /tmp && java -cp "/opt/kafka/libs/*" Practice.java slow-consumer'

kcg --describe --group s05-slow --state

docker exec kafka-1 sh -c 'cd /tmp && java -cp "/opt/kafka/libs/*" Practice.java static-member'

kcg --describe --group s05-static --state

kcg --describe --group s05-static --members --verbose

kcg --reset-offsets --group s05-demo --topic s05_orders --to-earliest --dry-run

kcg --list

for G in s05-demo s05-four s05-range s05-rr s05-coop s05-slow s05-static; do
  kcg --delete --group "$G" 2>/dev/null || echo "$G 삭제 실패 (활성 멤버가 남아 있는지 확인)"
done

kt --delete --topic s05_orders
kt --delete --topic s05_payments

kcg --list | grep '^s05-' || echo "s05- 그룹 없음 — 정리 완료"
kt --list | grep '^s05_' || echo "s05_ 토픽 없음 — 정리 완료"

#!/usr/bin/env bash
#
# Step 05 — 컨슈머와 컨슈머 그룹 : 본문(5-0 ~ 5-14)의 모든 명령
#
# 실행법:
#   bash practice.sh              # 통째로 실행 (8~12분)
#   bash -x practice.sh           # 한 줄씩 확인하며 실행 (권장)
#
# 이 스크립트는 본문에서 "터미널 A~D 를 여세요" 라고 한 부분을
# 백그라운드 실행(start_consumer)으로 대체합니다.
#
set -euo pipefail

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

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

CONSUMER_PIDS=()

# start_consumer <group> [추가 옵션...]
start_consumer() {
  local group="$1"; shift
  docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
    --bootstrap-server "$BS" --topic s05_orders --group "$group" "$@" \
    > /dev/null 2>&1 &
  CONSUMER_PIDS+=("$!")
  sleep 2   # 그룹 합류를 기다립니다 (group.initial.rebalance.delay.ms=0 이라 빠릅니다)
}

# start_multi <group> <토픽정규식> [추가 옵션...]
start_multi() {
  local group="$1" pattern="$2"; shift 2
  docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
    --bootstrap-server "$BS" --include "$pattern" --group "$group" "$@" \
    > /dev/null 2>&1 &
  CONSUMER_PIDS+=("$!")
  sleep 2
}

stop_all() {
  for p in "${CONSUMER_PIDS[@]:-}"; do
    kill -TERM "$p" 2>/dev/null || true
  done
  CONSUMER_PIDS=()
  # docker exec 로 띄운 자식 프로세스도 정리합니다
  docker exec kafka-1 pkill -f kafka-console-consumer 2>/dev/null || true
  sleep 3
}
trap stop_all EXIT

# ---------------------------------------------------------------------------
# [5-0] 실습 준비
# ---------------------------------------------------------------------------
hr "[5-0] 실습 준비"

for T in s05_orders s05_payments; do
  K kafka-topics.sh --bootstrap-server "$BS" --delete --topic "$T" 2>/dev/null || true
done
sleep 3

K kafka-topics.sh --bootstrap-server "$BS" --create --topic s05_orders \
  --partitions 3 --replication-factor 3
K kafka-topics.sh --bootstrap-server "$BS" --create --topic s05_payments \
  --partitions 3 --replication-factor 3

seed() {  # $1=토픽 $2=건수
  docker exec kafka-1 bash -c "
    for i in \$(seq 1 $2); do
      k=\"C\$(printf '%03d' \$((RANDOM % 10 + 1)))\"
      echo \"\$k:{\\\"order_id\\\":\\\"O-\$((1000+i))\\\",\\\"customer_id\\\":\\\"\$k\\\",\\\"amount\\\":\$((RANDOM % 90000 + 1000)),\\\"status\\\":\\\"CREATED\\\"}\"
    done | /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server kafka-1:9092 \
      --topic $1 --property parse.key=true --property key.separator=:"
}

seed s05_orders 300
seed s05_payments 300

K kafka-get-offsets.sh --bootstrap-server "$BS" --topic s05_orders

# ---------------------------------------------------------------------------
# [5-4] 컨슈머 그룹 만들고 랙 읽기
# ---------------------------------------------------------------------------
hr "[5-4] 랙 확인"

# 120건만 읽고 종료합니다
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic s05_orders \
  --group s05-demo --from-beginning --max-messages 120 > /dev/null

K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s05-demo

# CURRENT-OFFSET  : 그룹이 커밋한 오프셋 (다음에 읽을 위치)
# LOG-END-OFFSET  : 파티션의 마지막 오프셋 + 1
# LAG             : 아직 처리하지 못한 건수 (= LOG-END - CURRENT)
# 컨슈머가 없으면 CONSUMER-ID / HOST / CLIENT-ID 가 '-' 이고
# 맨 위에 "has no active members." 가 붙습니다.

echo "--- 끝까지 읽으면 LAG 이 0 이 됩니다 ---"
set +e
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic s05_orders \
  --group s05-demo --timeout-ms 10000 > /dev/null 2>&1
set -e
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s05-demo

# ⚠️ --from-beginning 을 줘도 커밋된 오프셋이 있으면 그것이 우선합니다.
#    "분명 --from-beginning 을 줬는데 예전 메시지가 안 나온다" 의 원인입니다.

# ---------------------------------------------------------------------------
# [5-5] --list / --members / --state
# ---------------------------------------------------------------------------
hr "[5-5] 세 가지 관점"

start_consumer s05-demo
start_consumer s05-demo
start_consumer s05-demo

echo "--- --list ---"
K kafka-consumer-groups.sh --bootstrap-server "$BS" --list

echo "--- --describe --members ---"
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s05-demo --members

echo "--- --describe --members --verbose  (ASSIGNMENT 컬럼이 핵심) ---"
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s05-demo --members --verbose

echo "--- --describe --state ---"
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s05-demo --state

stop_all

# ---------------------------------------------------------------------------
# [5-6] 핵심 함정 A — 컨슈머 수 > 파티션 수
# ---------------------------------------------------------------------------
hr "[5-6] 함정 A — 남는 컨슈머는 논다"

# 파티션 3개짜리 토픽에 컨슈머 4개
start_consumer s05-four
start_consumer s05-four
start_consumer s05-four
start_consumer s05-four
sleep 3

K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s05-four --members --verbose

echo
echo ">>> #PARTITIONS 가 0 인 컨슈머 (아무것도 할당받지 못한 컨슈머):"
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s05-four --members --verbose \
  | awk 'NR>1 && $5 == 0 {print "    " $2 "  → 놀고 있습니다"}'

echo
echo ">>> --describe (파티션 기준) 에는 이 컨슈머가 아예 나오지 않습니다:"
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s05-four

# 파티션이 병렬성의 상한입니다. 컨슈머를 10개 띄워도 7개는 유휴입니다.
# 에러도 안 나고 경고도 없습니다. 증상은 "인스턴스를 늘렸는데 랙이 그대로" 입니다.

# ---------------------------------------------------------------------------
# [5-7] 리밸런싱 관찰
# ---------------------------------------------------------------------------
hr "[5-7] 리밸런싱 — 컨슈머 하나를 죽인다"

VICTIM="${CONSUMER_PIDS[0]}"
echo "  죽일 컨슈머 PID = $VICTIM"
kill -TERM "$VICTIM" 2>/dev/null || true

for i in $(seq 1 8); do
  date '+%H:%M:%S'
  K kafka-consumer-groups.sh --bootstrap-server "$BS" \
    --describe --group s05-four --members --verbose 2>/dev/null | tail -n +2
  echo '---'
  sleep 1
done

# "Warning: Consumer group 's05-four' is rebalancing." 가 잠깐 뜨고
# 2초쯤 뒤 재배분이 끝납니다.
# ★ 놀고 있던 네 번째 컨슈머가 파티션을 이어받는 것을 확인하십시오.
#
# Ctrl+C / SIGTERM → LeaveGroup 을 보내므로 즉시 리밸런싱
# kill -9          → 하트비트가 끊긴 것을 감지할 때까지 session.timeout.ms(45초) 대기

stop_all

# ---------------------------------------------------------------------------
# [5-8] 할당 전략 — Range vs RoundRobin
# ---------------------------------------------------------------------------
hr "[5-8] Range vs RoundRobin"

echo "--- RangeAssignor (기본) — 두 토픽 구독, 컨슈머 2개 ---"
start_multi s05-range 's05_(orders|payments)'
start_multi s05-range 's05_(orders|payments)'
sleep 3
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s05-range --members --verbose
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s05-range --state
stop_all

# Range 는 토픽마다 독립적으로 계산하므로 앞쪽 컨슈머에 쏠립니다 → 4 : 2

echo "--- RoundRobinAssignor — 같은 조건 ---"
RR=org.apache.kafka.clients.consumer.RoundRobinAssignor
start_multi s05-rr 's05_(orders|payments)' --consumer-property partition.assignment.strategy=$RR
start_multi s05-rr 's05_(orders|payments)' --consumer-property partition.assignment.strategy=$RR
sleep 3
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s05-rr --members --verbose
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s05-rr --state
stop_all

# RoundRobin 은 모든 토픽의 파티션을 한 줄로 세워 배분하므로 3 : 3 으로 균등합니다.
# ⚠️ 같은 그룹에 서로 다른 전략을 섞으면 InconsistentGroupProtocolException 이 납니다.
#    그래서 그룹 이름을 s05-range / s05-rr 로 나눴습니다.

# ---------------------------------------------------------------------------
# [5-9] 핵심 함정 B — Eager vs Cooperative
# ---------------------------------------------------------------------------
hr "[5-9] 함정 B — Cooperative 리밸런싱"

COOP=org.apache.kafka.clients.consumer.CooperativeStickyAssignor
start_consumer s05-coop --consumer-property partition.assignment.strategy=$COOP
start_consumer s05-coop --consumer-property partition.assignment.strategy=$COOP
start_consumer s05-coop --consumer-property partition.assignment.strategy=$COOP
sleep 3

K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s05-coop --state
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s05-coop --members --verbose

echo "--- 하나를 죽여 점진적 리밸런싱을 봅니다 ---"
kill -TERM "${CONSUMER_PIDS[2]}" 2>/dev/null || true
for i in 1 2 3 4; do
  date '+%H:%M:%S'
  K kafka-consumer-groups.sh --bootstrap-server "$BS" \
    --describe --group s05-coop --members --verbose 2>/dev/null | tail -n +2
  echo '---'
  sleep 1
done
stop_all

# Cooperative 는 이동하는 파티션만 반납합니다.
# 컨슈머 로그에 "Revoked partitions (owned - assigned): []" 가 찍히는 것이 증거입니다.
# ⚠️ Eager → Cooperative 전환은 반드시 2단계 롤링으로:
#    1단계: partition.assignment.strategy=CooperativeSticky,Range  (전체 배포)
#    2단계: partition.assignment.strategy=CooperativeSticky        (전체 배포)

# ---------------------------------------------------------------------------
# [5-10] 핵심 함정 C — max.poll.interval.ms 초과
# ---------------------------------------------------------------------------
hr "[5-10] 함정 C — poll timeout 으로 조용히 쫓겨남"

docker cp "$(dirname "$0")/Practice.java" kafka-1:/tmp/

echo "  이 시나리오는 스스로 끝나지 않습니다(그것이 함정의 본질입니다)."
echo "  60초 동안만 돌리고 끊습니다."
echo "  실행 중에 다른 창에서 아래를 반복하면 STATE 가 요동치는 것이 보입니다:"
echo "      kcg --describe --group s05-slow --state"
echo

set +e
docker exec kafka-1 sh -c \
  'cd /tmp && timeout 60 java -cp "/opt/kafka/libs/*" Practice.java slow-consumer'
set -e

K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s05-slow 2>/dev/null || true

# 일어난 일:
#   poll() 5건 → 처리 20초 → 10초 시점에 코디네이터가 제거
#   → 컨슈머가 스스로 LeaveGroup → commitSync() 가 CommitFailedException
#   → 재합류 → 커밋 안 된 같은 5건을 다시 받음 → 무한 반복
# LAG 이 줄지 않고 generationId 만 계속 오릅니다. 에러처럼 보이지 않습니다.
#
# 해결 순서:
#   1) max.poll.records 축소  ← 가장 먼저
#   2) 처리 최적화
#   3) max.poll.interval.ms 증가  ← 마지막 수단
#      (이 값을 늘리면 rebalance.timeout.ms 도 함께 늘어 Eager 정지 시간이 길어집니다)

echo "--- MAX_POLL_RECORDS=1 로 해결되는지 확인 ---"
set +e
docker exec -e MAX_POLL_RECORDS=1 kafka-1 sh -c \
  'cd /tmp && timeout 30 java -cp "/opt/kafka/libs/*" Practice.java slow-consumer'
set -e

# ---------------------------------------------------------------------------
# [5-12] 정적 멤버십
# ---------------------------------------------------------------------------
hr "[5-12] 정적 멤버십 (group.instance.id)"

docker exec kafka-1 sh -c 'cd /tmp && java -cp "/opt/kafka/libs/*" Practice.java static-member'

echo "--- 종료 직후 상태 (STATE 가 Stable, #MEMBERS 가 1 로 남아 있습니다) ---"
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s05-static --state
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s05-static --members --verbose

echo "--- 30초 안에 다시 실행하면 리밸런싱 없이 같은 파티션을 받습니다 ---"
docker exec kafka-1 sh -c 'cd /tmp && java -cp "/opt/kafka/libs/*" Practice.java static-member'

# ⚠️ 정적 멤버십은 진짜 장애의 감지도 늦춥니다.
#    컨슈머가 진짜 죽어도 session.timeout.ms 동안 그 파티션은 아무도 소비하지 않습니다.
#    session.timeout.ms 는 "재시작 시간 + 여유" 정도로만 잡으십시오.

# ---------------------------------------------------------------------------
# [5-13] --reset-offsets 은 Step 06 에서
# ---------------------------------------------------------------------------
hr "[5-13] --reset-offsets 맛보기"

K kafka-consumer-groups.sh --bootstrap-server "$BS" --reset-offsets \
  --group s05-demo --topic s05_orders --to-earliest --dry-run

# --dry-run 은 계산만 하고 적용하지 않습니다. 적용은 --execute.
# ⚠️ 그룹에 활성 멤버가 있으면 실패합니다:
#    Error: Assignments can only be reset if the group is inactive, but the current state is Stable.
# 오프셋 관리 전체는 Step 06 에서 다룹니다.

# ---------------------------------------------------------------------------
# [5-14] 정리
# ---------------------------------------------------------------------------
hr "[5-14] 정리"

stop_all
echo "  정적 멤버십 그룹은 session.timeout.ms 동안 멤버로 남으므로 잠시 기다립니다..."
sleep 35

for G in s05-demo s05-four s05-range s05-rr s05-coop s05-slow s05-static; do
  K kafka-consumer-groups.sh --bootstrap-server "$BS" --delete --group "$G" 2>/dev/null \
    || echo "  $G 삭제 실패 (활성 멤버가 남아 있는지 확인)"
done

for T in s05_orders s05_payments; do
  K kafka-topics.sh --bootstrap-server "$BS" --delete --topic "$T" 2>/dev/null || true
done
sleep 3

K kafka-consumer-groups.sh --bootstrap-server "$BS" --list | grep '^s05-' \
  || echo "s05- 그룹 없음 — 정리 완료"
K kafka-topics.sh --bootstrap-server "$BS" --list | grep '^s05_' \
  || echo "s05_ 토픽 없음 — 정리 완료"

echo
echo "Step 05 완료. 그룹을 안 지우고 넘어가면 Step 06 에서 '왜 메시지가 안 읽히지?' 로 헤맵니다."

#!/usr/bin/env bash
#
# Step 05 — 컨슈머와 컨슈머 그룹 : 연습문제 7문제
#
# 실행법:
#   bash exercise.sh            # 준비 구간만 돌고 각 문제에서 멈춥니다
#   (권장) 이 파일을 열어 놓고 "# 여기에 작성:" 자리를 채운 뒤 그 부분만 복사해 실행
#
# 정답은 solution.sh 에 있습니다. 먼저 직접 풀어 보십시오.
#
set -euo pipefail

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

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

PIDS=()

start_consumer() {   # start_consumer <group> [옵션...]
  local g="$1"; shift
  docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
    --bootstrap-server "$BS" --topic s05_ex_orders --group "$g" "$@" >/dev/null 2>&1 &
  PIDS+=("$!")
  sleep 2
}

stop_all() {
  for p in "${PIDS[@]:-}"; do kill -TERM "$p" 2>/dev/null || true; done
  PIDS=()
  docker exec kafka-1 pkill -f kafka-console-consumer 2>/dev/null || true
  sleep 3
}

seed() {   # seed <토픽> <건수>
  docker exec kafka-1 bash -c "
    for i in \$(seq 1 $2); do
      k=\"C\$(printf '%03d' \$((RANDOM % 10 + 1)))\"
      echo \"\$k:{\\\"order_id\\\":\\\"O-\$((1000+i))\\\",\\\"customer_id\\\":\\\"\$k\\\",\\\"amount\\\":10000,\\\"status\\\":\\\"CREATED\\\"}\"
    done | /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server kafka-1:9092 \
      --topic $1 --property parse.key=true --property key.separator=:"
}

cleanup_ex() {
  stop_all
  sleep 5
  for G in $(K kafka-consumer-groups.sh --bootstrap-server "$BS" --list 2>/dev/null | grep '^s05-ex-' || true); do
    K kafka-consumer-groups.sh --bootstrap-server "$BS" --delete --group "$G" 2>/dev/null || true
  done
  for T in $(K kafka-topics.sh --bootstrap-server "$BS" --list 2>/dev/null | grep '^s05_ex_' || true); do
    K kafka-topics.sh --bootstrap-server "$BS" --delete --topic "$T" 2>/dev/null || true
  done
  echo "s05_ex_ 토픽 / s05-ex- 그룹 정리 완료"
}
trap stop_all EXIT

# ---------------------------------------------------------------------------
# 준비
# ---------------------------------------------------------------------------
hr "준비"
cleanup_ex
sleep 3

K kafka-topics.sh --bootstrap-server "$BS" --create --topic s05_ex_orders \
  --partitions 3 --replication-factor 3
K kafka-topics.sh --bootstrap-server "$BS" --create --topic s05_ex_payments \
  --partitions 3 --replication-factor 3
seed s05_ex_orders 200
seed s05_ex_payments 200
# 데이터가 부족해지면 위 seed 두 줄만 다시 실행하십시오.


# ===========================================================================
# 문제 1. 컨슈머를 1 → 2 → 3 → 4 개로 늘리며 할당 변화 기록
# ===========================================================================
# 그룹 s05-ex-scale 에 컨슈머를 하나씩 늘려 가며
# --members --verbose 로 각 단계의 ASSIGNMENT 를 기록하십시오.
#
# 표를 채우십시오:
#   컨슈머 1개 → 할당 분포 [ ____ ]
#   컨슈머 2개 → 할당 분포 [ ____ ]
#   컨슈머 3개 → 할당 분포 [ ____ ]
#   컨슈머 4개 → 할당 분포 [ ____ ]
#
# 힌트: start_consumer s05-ex-scale
hr "문제 1"

# 여기에 작성:
#   for n in 1 2 3 4; do
#     start_consumer s05-ex-scale
#     echo "=== 컨슈머 ${n}개 ==="
#     K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe \
#        --group ______ --members --verbose
#   done


# ===========================================================================
# 문제 2. kill -9 vs kill -TERM — 리밸런싱까지 걸리는 시간
# ===========================================================================
# 컨슈머 3개를 띄운 뒤 하나를 죽이고, 재배분이 끝날 때까지의 시간을 재십시오.
# kill -TERM 과 kill -9 를 각각 측정해 비교합니다.
#
# 힌트: --members --verbose 의 줄 수가 3 → 2 로 줄고
#       다시 "각 컨슈머가 파티션을 가진 상태" 가 되면 재배분 완료입니다.
# 힌트: date +%s 로 시각을 찍으십시오.
hr "문제 2"

# 여기에 작성: (SIGTERM)
#   start_consumer s05-ex-term; start_consumer s05-ex-term; start_consumer s05-ex-term
#   T0=$(date +%s)
#   kill -____ "${PIDS[0]}"
#   while [ "$(K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe \
#              --group s05-ex-term --members 2>/dev/null | tail -n +2 | wc -l)" -ne 2 ]; do
#     sleep 0.2
#   done
#   echo "SIGTERM: $(( $(date +%s) - T0 ))초"

# 여기에 작성: (SIGKILL — 같은 방식, kill -9)
#


# ===========================================================================
# 문제 3. LAG 이 가장 큰 파티션 찾기 (한 줄 명령)
# ===========================================================================
# --describe 출력에서 LAG 이 가장 큰 파티션 한 줄만 뽑는 명령을 만드십시오.
#
# 힌트: LAG 은 6번째 컬럼입니다.
# 힌트: 헤더 줄과 LAG 이 '-' 인 줄을 걸러야 합니다.
hr "문제 3"

# (문제용 랙 만들기 — 절반만 읽습니다)
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic s05_ex_orders \
  --group s05-ex-lag --from-beginning --max-messages 60 >/dev/null

# 여기에 작성:
#   K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s05-ex-lag \
#     | awk '______________________________' | ______ | ______


# ===========================================================================
# 문제 4. 같은 그룹에 서로 다른 할당 전략을 섞으면?
# ===========================================================================
# 그룹 s05-ex-mix 에
#   - 컨슈머 1: RangeAssignor (기본)
#   - 컨슈머 2: RoundRobinAssignor
# 를 붙이면 어떻게 됩니까? 나오는 예외 이름을 적으십시오.
#
# 힌트: 콘솔 컨슈머는 이 예외를 stderr 로 뱉고 종료합니다.
#       > /dev/null 로 버리면 안 보입니다. 2>&1 | grep 으로 잡으십시오.
#
# 예외 이름: org.apache.kafka.common.errors.____________________
hr "문제 4"

# 여기에 작성:
#


# ===========================================================================
# 문제 5. Range vs RoundRobin 의 불균형
# ===========================================================================
# s05_ex_orders 와 s05_ex_payments 를 함께 구독하는 컨슈머 2개를
# Range 로 한 번, RoundRobin 으로 한 번 띄워 #PARTITIONS 분포를 비교하십시오.
#
# 힌트: --include 's05_ex_(orders|payments)'
# 힌트: 그룹 이름을 반드시 다르게 하십시오 (문제 4 의 이유).
#
# 답:  Range → ____ : ____ ,   RoundRobin → ____ : ____
hr "문제 5"

# 여기에 작성:
#


# ===========================================================================
# 문제 6. max.poll.interval.ms 초과 재현하고 해결하기
# ===========================================================================
# (a) Practice.java slow-consumer 를 40초간 돌려
#     CommitFailedException 과 "같은 메시지를 다시 받는 것" 을 확인하십시오.
# (b) 환경변수 MAX_POLL_RECORDS=1 만 주고 다시 돌려 문제가 사라지는지 보십시오.
# (c) 왜 max.poll.interval.ms 를 늘리는 것이 마지막 수단입니까? 한 문장으로.
#
# 힌트: docker exec -e TOPIC=s05_ex_orders -e MAX_POLL_RECORDS=1 kafka-1 sh -c '...'
# 힌트: 이 시나리오는 스스로 끝나지 않으므로 timeout 으로 감싸십시오.
# 힌트: Practice.java 는 TOPIC 환경변수를 읽습니다. 안 주면 s05_orders 를 봅니다.
hr "문제 6"

docker cp "$(dirname "$0")/Practice.java" kafka-1:/tmp/

# 여기에 작성: (a)
#

# 여기에 작성: (b)
#

# (c) 답:
#


# ===========================================================================
# 문제 7. 정적 멤버십 — generationId 가 오르지 않는 것 확인
# ===========================================================================
# Practice.java static-member 를 두 번 연속 실행하고,
# 두 번의 generationId 가 같은지 확인하십시오.
# 그다음 group.instance.id 없이(consume 시나리오) 두 번 실행해 비교하십시오.
#
# 힌트: generationId 는 컨슈머 로그의
#       "Successfully joined group with generation Generation{generationId=N" 에 있습니다.
#       프로그램도 마지막에 직접 출력합니다.
#
# 답:  정적 멤버 → generationId ____ , ____   (변화 ____)
#      동적 멤버 → generationId ____ , ____   (변화 ____)
hr "문제 7"

# 여기에 작성:
#


# ===========================================================================
# 정리
# ===========================================================================
hr "정리"
# 중간에 멈췄다면 이 함수만 따로 호출하십시오.
cleanup_ex

K kafka-consumer-groups.sh --bootstrap-server "$BS" --list | grep '^s05-' \
  || echo "s05- 그룹 없음 — 정리 완료"
K kafka-topics.sh --bootstrap-server "$BS" --list | grep '^s05_' \
  || echo "s05_ 토픽 없음 — 정리 완료"

#!/usr/bin/env bash
#
# Step 05 — 컨슈머와 컨슈머 그룹 : 연습문제 정답과 해설
#
# 실행법:
#   bash solution.sh
#
# exercise.sh 를 먼저 풀어 본 뒤에 여십시오.
#
set -euo pipefail

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

PIDS=()
start_consumer() {   # start_consumer <group> [옵션...]
  local g="$1"; shift
  docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
    --bootstrap-server "$BS" --topic s05_ex_orders --group "$g" "$@" >/dev/null 2>&1 &
  PIDS+=("$!")
  sleep 2
}
start_multi() {      # start_multi <group> <pattern> [옵션...]
  local g="$1" p="$2"; shift 2
  docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
    --bootstrap-server "$BS" --include "$p" --group "$g" "$@" >/dev/null 2>&1 &
  PIDS+=("$!")
  sleep 2
}
stop_all() {
  for p in "${PIDS[@]:-}"; do kill -TERM "$p" 2>/dev/null || true; done
  PIDS=()
  docker exec kafka-1 pkill -f kafka-console-consumer 2>/dev/null || true
  sleep 3
}
trap stop_all EXIT

members() { K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group "$1" --members --verbose; }

seed() {
  docker exec kafka-1 bash -c "
    for i in \$(seq 1 $2); do
      k=\"C\$(printf '%03d' \$((RANDOM % 10 + 1)))\"
      echo \"\$k:{\\\"order_id\\\":\\\"O-\$((1000+i))\\\",\\\"customer_id\\\":\\\"\$k\\\",\\\"amount\\\":10000,\\\"status\\\":\\\"CREATED\\\"}\"
    done | /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server kafka-1:9092 \
      --topic $1 --property parse.key=true --property key.separator=:"
}

# 준비
for T in s05_ex_orders s05_ex_payments; do
  K kafka-topics.sh --bootstrap-server "$BS" --delete --topic "$T" 2>/dev/null || true
done
sleep 3
K kafka-topics.sh --bootstrap-server "$BS" --create --topic s05_ex_orders   --partitions 3 --replication-factor 3
K kafka-topics.sh --bootstrap-server "$BS" --create --topic s05_ex_payments --partitions 3 --replication-factor 3
seed s05_ex_orders 200
seed s05_ex_payments 200


# ===========================================================================
# 정답 1. 컨슈머 1 → 4 개
# ===========================================================================
hr "정답 1 — 할당 변화"

for n in 1 2 3 4; do
  start_consumer s05-ex-scale
  sleep 2
  echo "=== 컨슈머 ${n}개 ==="
  members s05-ex-scale | tail -n +2 | awk '{printf "   %-30s #PARTITIONS=%s  %s\n", substr($2,1,28), $5, $6}'
done
stop_all
sleep 3
K kafka-consumer-groups.sh --bootstrap-server "$BS" --delete --group s05-ex-scale 2>/dev/null || true

# 정답 (할당 분포):
#   컨슈머 1개 → [3]           한 컨슈머가 P0,P1,P2 전부
#   컨슈머 2개 → [2, 1]        Range 는 앞쪽에 더 줍니다
#   컨슈머 3개 → [1, 1, 1]     완전 균등
#   컨슈머 4개 → [1, 1, 1, 0]  ★ 네 번째는 0개
#
# 해설:
#   ★ 이 문제의 핵심은 "네 번째가 논다" 보다
#     "--describe 화면에는 네 번째 컨슈머가 아예 나오지 않는다" 는 점입니다.
#     --describe 는 "파티션" 기준으로 줄을 만들기 때문에,
#     파티션을 하나도 안 가진 컨슈머는 출력에 등장할 자리가 없습니다.
#     --members 를 봐야만 존재를 알 수 있습니다.
#
#   ★ 파티션이 병렬성의 상한입니다.
#     "인스턴스를 2배로 늘렸는데 랙이 그대로다" 의 정체가 이것이고,
#     에러도 경고도 나지 않습니다.
#
#   ★ 유휴 컨슈머가 완전히 무의미하지는 않습니다.
#     다른 컨슈머가 죽으면 즉시 이어받는 핫 스탠바이가 됩니다.
#     의도한 것이면 괜찮지만, 처리량을 늘리려고 띄웠다면 잘못된 것입니다.
#
#   ★ 컨슈머 수는 파티션 수의 약수로 두십시오.
#     파티션 6에 컨슈머 4면 2,2,1,1 로 불균등합니다.
#     오토스케일링을 쓴다면 파티션을 12나 24처럼 약수가 많은 수로 잡으십시오.


# ===========================================================================
# 정답 2. kill -TERM vs kill -9
# ===========================================================================
hr "정답 2 — 리밸런싱까지 걸리는 시간"

wait_for_members() {   # wait_for_members <group> <기대 멤버수> <타임아웃초>
  local g="$1" want="$2" limit="$3" t0
  t0=$(date +%s)
  while true; do
    local n
    n=$(K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe \
          --group "$g" --members 2>/dev/null | tail -n +2 | grep -c . || true)
    [ "$n" -eq "$want" ] && break
    [ $(( $(date +%s) - t0 )) -ge "$limit" ] && break
    sleep 1
  done
  echo $(( $(date +%s) - t0 ))
}

echo "--- SIGTERM (Ctrl+C 와 동일) ---"
start_consumer s05-ex-term; start_consumer s05-ex-term; start_consumer s05-ex-term
sleep 3
kill -TERM "${PIDS[0]}" 2>/dev/null || true
echo "  재배분 완료까지: $(wait_for_members s05-ex-term 2 90)초"
stop_all

echo "--- SIGKILL ---"
start_consumer s05-ex-kill; start_consumer s05-ex-kill; start_consumer s05-ex-kill
sleep 3
docker exec kafka-1 pkill -9 -f 'kafka-console-consumer' 2>/dev/null || true
kill -9 "${PIDS[0]}" 2>/dev/null || true
echo "  재배분 완료까지: $(wait_for_members s05-ex-kill 2 90)초"
stop_all

# 참고 결과:
#   SIGTERM :  약 1.2초
#   SIGKILL : 약 46초  (= session.timeout.ms 45000 + 알파)
#
# 해설:
#   ★ SIGTERM 은 컨슈머가 close() → LeaveGroup 을 보내고 죽습니다.
#     코디네이터가 즉시 리밸런싱을 시작하므로 1~2초입니다.
#   ★ SIGKILL 은 아무 인사도 없이 프로세스가 사라집니다.
#     코디네이터는 하트비트가 session.timeout.ms(45초) 동안 안 오는 것을
#     확인한 뒤에야 그 멤버를 제거합니다. 그 45초 동안
#     ★ 죽은 컨슈머가 맡던 파티션은 아무도 소비하지 않습니다. 랙이 쌓입니다.
#
#   ★ 실무 함의:
#     쿠버네티스의 terminationGracePeriodSeconds 가 짧으면
#     SIGTERM 직후 SIGKILL 이 날아갑니다.
#     그러면 매 배포마다 파드 하나당 45초씩 파티션이 방치됩니다.
#     컨슈머 애플리케이션은 SIGTERM 을 받아 consumer.wakeup() → close() 를
#     반드시 수행해야 하고, 그레이스 기간은 그 시간보다 넉넉해야 합니다.


# ===========================================================================
# 정답 3. LAG 최대 파티션 찾기
# ===========================================================================
hr "정답 3 — LAG 최대 파티션"

K kafka-console-consumer.sh --bootstrap-server "$BS" --topic s05_ex_orders \
  --group s05-ex-lag --from-beginning --max-messages 60 >/dev/null

K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s05-ex-lag \
  | awk 'NR>1 && $6 ~ /^[0-9]+$/ {print $6, $0}' | sort -rn | head -1

# 정답:
#   awk 'NR>1 && $6 ~ /^[0-9]+$/ {print $6, $0}' | sort -rn | head -1
#
# 해설:
#   ★ 컬럼 위치: 1 GROUP / 2 TOPIC / 3 PARTITION / 4 CURRENT-OFFSET /
#                5 LOG-END-OFFSET / 6 LAG / 7 CONSUMER-ID / 8 HOST / 9 CLIENT-ID
#     LAG 은 6번째입니다.
#   ★ NR>1 로 헤더를 걸러야 합니다.
#   ★ $6 ~ /^[0-9]+$/ 로 '-' 를 걸러야 합니다.
#     커밋된 오프셋이 없는 파티션은 CURRENT-OFFSET 과 LAG 이 '-' 로 나옵니다.
#   ★ sort -rn 은 첫 컬럼(우리가 앞에 붙인 LAG)을 숫자로 내림차순 정렬합니다.
#
#   ★ 이 한 줄이 실무에서 hot partition 을 찾는 첫 번째 명령입니다.
#     "특정 고객의 주문만 처리가 늦다" 는 신고가 들어오면
#     이 명령으로 한 파티션의 LAG 만 큰 것을 확인하고,
#     그 파티션에 어떤 키가 몰려 있는지를 봅니다 (Step 04 의 함정 C).
#     전체 LAG 평균만 보면 절대 안 보이는 문제입니다.


# ===========================================================================
# 정답 4. 할당 전략을 섞으면
# ===========================================================================
hr "정답 4 — InconsistentGroupProtocolException"

RR=org.apache.kafka.clients.consumer.RoundRobinAssignor

start_consumer s05-ex-mix     # 기본 = RangeAssignor
sleep 2

echo "  같은 그룹에 RoundRobinAssignor 컨슈머를 붙입니다:"
set +e
docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server "$BS" --topic s05_ex_orders --group s05-ex-mix \
  --consumer-property partition.assignment.strategy=$RR \
  --timeout-ms 10000 2>&1 | grep -i -m2 'Exception\|incompatible' || echo "  (예외를 못 잡았습니다. 다시 시도하십시오)"
set -e
stop_all

# 정답:
#   org.apache.kafka.common.errors.InconsistentGroupProtocolException:
#     The group member's supported protocols are incompatible with those of existing members.
#
# 해설:
#   ★ 코디네이터는 "모든 멤버가 공통으로 지원하는 전략" 중 하나를 고릅니다.
#     겹치는 것이 하나도 없으면 그룹이 아예 구성되지 않습니다.
#
#   ★ 그래서 할당 전략을 바꾸는 배포는 반드시 2단계 롤링입니다:
#       1단계: partition.assignment.strategy=NewAssignor,OldAssignor   (전체 배포)
#              → 공통 전략이 여전히 OldAssignor 이므로 동작이 안 바뀝니다.
#       2단계: partition.assignment.strategy=NewAssignor               (전체 배포)
#              → 이제 공통 전략이 NewAssignor 로 바뀌며 전환됩니다.
#
#   ★ Eager(Range/RoundRobin/Sticky) → Cooperative 전환에도 똑같이 적용됩니다.
#     Cooperative 는 프로토콜 자체가 달라서, 한 번에 바꾸면
#     배포 도중 그룹이 InconsistentGroupProtocolException 으로 멈춥니다.
#     진행 상황은 --describe --state 의 ASSIGNMENT-STRATEGY 컬럼으로 확인합니다.


# ===========================================================================
# 정답 5. Range vs RoundRobin
# ===========================================================================
hr "정답 5 — 두 토픽 구독 시 불균형"

echo "--- Range (기본) ---"
start_multi s05-ex-range 's05_ex_(orders|payments)'
start_multi s05-ex-range 's05_ex_(orders|payments)'
sleep 3
members s05-ex-range
stop_all

echo "--- RoundRobin ---"
start_multi s05-ex-rr 's05_ex_(orders|payments)' --consumer-property partition.assignment.strategy=$RR
start_multi s05-ex-rr 's05_ex_(orders|payments)' --consumer-property partition.assignment.strategy=$RR
sleep 3
members s05-ex-rr
stop_all

# 정답:
#   Range      → 4 : 2
#   RoundRobin → 3 : 3
#
# 해설:
#   ★ Range 는 "토픽마다 독립적으로" 계산합니다.
#     s05_ex_orders   3파티션 / 2컨슈머 → 2, 1
#     s05_ex_payments 3파티션 / 2컨슈머 → 2, 1
#     둘 다 앞쪽 컨슈머가 2개씩 가져가므로 합쳐서 4 : 2 입니다.
#     ★ 토픽이 10개면 20 : 10 이 됩니다. 토픽이 많을수록 격차가 벌어집니다.
#
#   ★ RoundRobin 은 모든 토픽의 파티션을 한 줄로 세워 배분하므로 균등합니다.
#     대신 리밸런싱 때 기존 할당을 전혀 고려하지 않아 전부 재배치됩니다.
#
#   ★ 그래서 실무의 정답은 대개 CooperativeStickyAssignor 입니다.
#     균등(Sticky) + 기존 할당 유지 + 점진적 리밸런싱을 모두 갖췄습니다.
#     기본값이 여전히 range 인 것은 하위 호환 때문입니다.


# ===========================================================================
# 정답 6. max.poll.interval.ms 초과와 해결
# ===========================================================================
hr "정답 6 — poll timeout"

docker cp "$(dirname "$0")/Practice.java" kafka-1:/tmp/

echo "--- (a) 기본값: max.poll.records=5, 처리 4초 → 20초 > 10초 ---"
set +e
docker exec -e TOPIC=s05_ex_orders kafka-1 sh -c \
  'cd /tmp && timeout 40 java -cp "/opt/kafka/libs/*" Practice.java slow-consumer'
set -e

echo
echo "--- (b) MAX_POLL_RECORDS=1 → 한 루프 4초 < 10초 ---"
set +e
docker exec -e TOPIC=s05_ex_orders -e MAX_POLL_RECORDS=1 kafka-1 sh -c \
  'cd /tmp && timeout 30 java -cp "/opt/kafka/libs/*" Practice.java slow-consumer'
set -e

# 정답:
#   (a) WARN "consumer poll timeout has expired" →
#       "sending LeaveGroup request ... due to consumer poll timeout has expired" →
#       commitSync() 에서 CommitFailedException →
#       재합류 후 같은 5건을 다시 수신 → 무한 반복
#   (b) MAX_POLL_RECORDS=1 이면 한 루프가 4초라 타임아웃이 나지 않고,
#       커밋이 성공하며 오프셋이 정상적으로 전진합니다.
#   (c) max.poll.interval.ms 를 늘리면 rebalance.timeout.ms 가 함께 늘어나
#       Eager 리밸런싱 시 그룹 전체가 멈춰 있는 시간까지 길어지기 때문입니다.
#
# 해설:
#   ★ 왜 하트비트는 멀쩡한데 쫓겨납니까?
#     하트비트는 백그라운드 스레드가 3초마다 따로 보냅니다.
#     처리가 아무리 오래 걸려도 하트비트는 정상입니다.
#     코디네이터 입장에서 이 컨슈머는 살아 있습니다.
#     그런데 poll() 을 안 부르니 컨슈머 자신이 판단해 LeaveGroup 을 보냅니다.
#       session.timeout.ms    → 하트비트 스레드 / 코디네이터가 감지 / 45초
#       max.poll.interval.ms  → 처리 스레드     / 컨슈머가 감지     / 300초
#
#   ★ 이 문제가 무서운 이유는 "에러로 보이지 않는다" 는 것입니다.
#     프로세스는 안 죽고, 처리 로그는 계속 찍히고, 처리량 메트릭도 정상입니다.
#     같은 메시지를 반복 처리하고 있을 뿐입니다.
#     유일한 단서는 LAG 이 줄지 않는 것과 generationId 가 계속 오르는 것입니다.
#
#   ★ 해결 순서 (위에서부터):
#     1) max.poll.records 축소 — 가장 먼저. 500 → 50 이면 한 루프가 1/10.
#     2) 처리 최적화 — 건당 4초가 걸리는 원인(동기 HTTP, N+1 쿼리)을 없앱니다.
#     3) max.poll.interval.ms 증가 — 마지막 수단. (c)의 이유 때문입니다.
#     4) 처리를 스레드 풀로 넘기고 pause()/resume() — 정석이지만 오프셋 관리가 복잡 (Step 06).
#
#   ★ 계산 공식:
#     max.poll.interval.ms >= 건당 처리시간 × max.poll.records × 2
#     건당 3초 × 500건 = 1500초. 기본값 300초를 훌쩍 넘습니다.
#     이때 300 → 1500 으로 늘리는 게 아니라
#     max.poll.records 를 50 으로 줄여 150초로 맞추는 것이 정답입니다.


# ===========================================================================
# 정답 7. 정적 멤버십과 generationId
# ===========================================================================
hr "정답 7 — group.instance.id"

echo "--- 정적 멤버: 두 번 연속 실행 ---"
docker exec -e TOPIC=s05_ex_orders kafka-1 sh -c \
  'cd /tmp && java -cp "/opt/kafka/libs/*" Practice.java static-member' | grep -E 'generationId|assigned'
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s05-static --state
docker exec -e TOPIC=s05_ex_orders kafka-1 sh -c \
  'cd /tmp && java -cp "/opt/kafka/libs/*" Practice.java static-member' | grep -E 'generationId|assigned'

echo "--- 동적 멤버: 두 번 연속 실행 ---"
set +e
docker exec -e TOPIC=s05_ex_orders -e RUN_SECONDS=10 kafka-1 sh -c \
  'cd /tmp && java -cp "/opt/kafka/libs/*" Practice.java consume' | grep -E 'generationId|assigned'
docker exec -e TOPIC=s05_ex_orders -e RUN_SECONDS=10 kafka-1 sh -c \
  'cd /tmp && java -cp "/opt/kafka/libs/*" Practice.java consume' | grep -E 'generationId|assigned'
set -e

# 정답:
#   정적 멤버 → generationId 1, 1   (변화 없음)
#   동적 멤버 → generationId 1, 3   (내려갈 때 +1, 올라올 때 +1)
#
# 해설:
#   ★ group.instance.id 가 있으면 컨슈머는 "정적 멤버" 가 됩니다.
#     - 종료 시 LeaveGroup 을 보내지 않습니다.
#       코디네이터는 "잠깐 자리를 비웠다" 고 간주합니다.
#       그래서 종료 직후에도 --state 가 Stable / #MEMBERS=1 로 나옵니다.
#     - session.timeout.ms 안에 같은 group.instance.id 로 다시 합류하면
#       코디네이터가 이전과 똑같은 파티션을 그대로 돌려줍니다.
#       리밸런싱이 일어나지 않으므로 generationId 가 오르지 않습니다.
#
#   ★ 컨슈머 10개를 롤링 재시작하면
#     동적 멤버십: 리밸런싱 20회 (내려갈 때 10 + 올라올 때 10)
#     정적 멤버십: 리밸런싱 0회
#     Eager 프로토콜이라면 20회 × 전원 정지입니다. 차이가 큽니다.
#
#   ★ 대가:
#     진짜 크래시도 session.timeout.ms 동안 감지되지 않습니다.
#     그동안 그 파티션은 아무도 소비하지 않습니다.
#     session.timeout.ms=600000(10분) 으로 크게 잡으면 롤링은 완벽해지지만
#     진짜 장애 때 10분간 방치됩니다.
#     ★ "재시작 시간 + 여유" 정도(컨테이너 30초면 60000ms)로만 잡으십시오.
#
#   ★ group.instance.id 는 인스턴스마다 고유해야 합니다.
#     쿠버네티스라면 StatefulSet 의 파드 서수를 그대로 쓰면 됩니다.
#     Deployment 는 파드 이름이 매번 바뀌어 궁합이 나쁩니다.
#     같은 값이 동시에 두 개 뜨면:
#       org.apache.kafka.common.errors.FencedInstanceIdException


# ===========================================================================
# 정리
# ===========================================================================
hr "정리"
stop_all
sleep 35   # 정적 멤버가 빠질 때까지 대기

for G in $(K kafka-consumer-groups.sh --bootstrap-server "$BS" --list 2>/dev/null | grep -E '^s05-' || true); do
  K kafka-consumer-groups.sh --bootstrap-server "$BS" --delete --group "$G" 2>/dev/null \
    || echo "  $G 삭제 실패 (활성 멤버 확인)"
done
for T in $(K kafka-topics.sh --bootstrap-server "$BS" --list 2>/dev/null | grep '^s05_' || true); do
  K kafka-topics.sh --bootstrap-server "$BS" --delete --topic "$T" 2>/dev/null || true
done
sleep 3

K kafka-consumer-groups.sh --bootstrap-server "$BS" --list | grep '^s05-' \
  || echo "s05- 그룹 없음 — 정리 완료"
K kafka-topics.sh --bootstrap-server "$BS" --list | grep '^s05_' \
  || echo "s05_ 토픽 없음 — 정리 완료"

docker cp Practice.java kafka-1:/tmp/
docker exec kafka-1 sh -c 'cd /tmp && java -cp "/opt/kafka/libs/*" Practice.java <시나리오>'