#Step 11 — 성능 튜닝

docker exec kafka-1 /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka-1:9092 \
  --create --topic orders-perf --partitions 6 --replication-factor 3 \
  --config segment.bytes=268435456 \
  --config retention.ms=3600000

docker exec kafka-1 /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka-1:9092 \
  --describe --topic orders-perf

docker exec kafka-1 /opt/kafka/bin/kafka-producer-perf-test.sh \
  --topic orders-perf \
  --num-records 1000000 \
  --record-size 1024 \
  --throughput -1 \
  --producer-props bootstrap.servers=kafka-1:9092 acks=1

for A in 0 1 all; do
  docker exec kafka-1 /opt/kafka/bin/kafka-producer-perf-test.sh \
    --topic orders-perf --num-records 1000000 --record-size 1024 --throughput -1 \
    --producer-props bootstrap.servers=kafka-1:9092 acks=$A | tail -1
done

# 현실적인 압축률을 재려면 랜덤 바이트(--record-size)가 아니라 실제 데이터를 써야 합니다
docker exec kafka-1 sh -c 'for i in $(seq 1 1000); do
  echo "{\"order_id\":\"O-$i\",\"customer_id\":\"C00$((i%10))\",\"amount\":39000,\"status\":\"CREATED\"}"
done > /tmp/payload.txt'

docker exec kafka-1 /opt/kafka/bin/kafka-producer-perf-test.sh \
  --topic orders-perf --num-records 1000000 --throughput -1 \
  --payload-file /tmp/payload.txt \
  --producer-props bootstrap.servers=kafka-1:9092 acks=all linger.ms=20 \
                   batch.size=65536 compression.type=lz4 | tail -1

for T in 20000 40000 60000 80000 -1; do
  echo "=== throughput=$T ==="
  docker exec kafka-1 /opt/kafka/bin/kafka-producer-perf-test.sh \
    --topic orders-perf --num-records 600000 --record-size 1024 --throughput $T \
    --producer-props bootstrap.servers=kafka-1:9092 acks=all linger.ms=5 batch.size=65536 \
    | tail -1
done

docker exec kafka-1 /opt/kafka/bin/kafka-consumer-perf-test.sh \
  --bootstrap-server kafka-1:9092 \
  --topic orders-perf \
  --messages 1000000 \
  --group s11-perf \
  --timeout 60000

docker exec kafka-1 sh -c 'cat > /tmp/c.properties <<EOF
fetch.min.bytes=100000
fetch.max.wait.ms=500
max.partition.fetch.bytes=1048576
max.poll.records=500
EOF'

docker exec kafka-1 /opt/kafka/bin/kafka-consumer-perf-test.sh \
  --bootstrap-server kafka-1:9092 --topic orders-perf --messages 1000000 \
  --group s11-perf-2 --consumer.config /tmp/c.properties --timeout 60000

# 한산한 토픽을 흉내냅니다: 1건만 쓰고, fetch.min.bytes=100000 인 컨슈머로 읽습니다
docker exec kafka-1 /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka-1:9092 \
  --create --topic s11_quiet --partitions 1 --replication-factor 3

docker exec kafka-1 sh -c 'cat > /tmp/slow.properties <<EOF
fetch.min.bytes=100000
fetch.max.wait.ms=500
EOF'

# [터미널 B] 컨슈머를 먼저 띄웁니다
docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 --topic s11_quiet \
  --consumer.config /tmp/slow.properties \
  --property print.timestamp=true

# [터미널 A] 한 건 씁니다
echo '{"order_id":"O-1001","status":"CREATED"}' | docker exec -i kafka-1 \
  /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server kafka-1:9092 --topic s11_quiet

# 랙을 10초 간격으로 6번 찍어 추이를 봅니다
for i in $(seq 1 6); do
  echo -n "$(date '+%H:%M:%S')  "
  docker exec kafka-1 /opt/kafka/bin/kafka-consumer-groups.sh \
    --bootstrap-server kafka-1:9092 --describe --group s11-perf 2>/dev/null \
    | awk 'NR>1 && $6 ~ /^[0-9]+$/ {s+=$6} END {print "total-lag=" s}'
  sleep 10
done

docker exec kafka-1 /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 \
  --describe --group s11-perf

docker exec kafka-1 sh -c 'free -m'

docker exec kafka-1 sh -c 'ulimit -n; ls /proc/1/fd | wc -l'

docker exec kafka-1 /opt/kafka/bin/kafka-producer-perf-test.sh \
  --topic orders-perf --num-records 500000 --record-size 1024 --throughput -1 \
  --producer-props bootstrap.servers=kafka-1:9092 acks=all linger.ms=20 batch.size=65536 \
  --print-metrics 2>/dev/null \
  | grep -E 'record-send-rate|batch-size-avg|request-latency-avg|records-per-request-avg|buffer-available-bytes|compression-rate-avg|record-queue-time-avg'

docker exec kafka-1 /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka-1:9092 \
  --delete --topic orders-perf
docker exec kafka-1 /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka-1:9092 \
  --delete --topic s11_quiet

docker exec kafka-1 /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka-1:9092 --list

docker exec kafka-1 /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 \
  --delete --group s11-perf --group s11-perf-2

docker exec kafka-1 sh -c 'du -sh /var/lib/kafka/data | tail -1'

#!/usr/bin/env bash
# ---------------------------------------------------------------------------
# Step 11 — 성능 튜닝 : 본문 실습 스크립트
#
# 실행법:
#   bash step-11-performance/practice.sh          # 통째로 (20~30분)
#   RECORDS=200000 bash step-11-performance/practice.sh   # 빠르게
#   bash -x step-11-performance/practice.sh       # 한 줄씩 확인하며
#
# ⚠️ 이 스크립트는 100만 건 측정을 열 번 넘게 돌립니다. 시간이 없으면
#    RECORDS 를 200000 으로 낮추십시오. 절대값은 달라지지만 배율은 유지됩니다.
# ---------------------------------------------------------------------------
set -euo pipefail

BS=kafka-1:9092
TOPIC=${TOPIC:-orders-perf}
RECORDS=${RECORDS:-1000000}
SIZE=${SIZE:-1024}

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

# 프로듀서 측정 헬퍼: 마지막 요약 줄만 남깁니다.
#   run_producer acks=all linger.ms=20 batch.size=65536
run_producer() {
  K kafka-producer-perf-test.sh \
    --topic "$TOPIC" --num-records "$RECORDS" --record-size "$SIZE" --throughput -1 \
    --producer-props bootstrap.servers=$BS "$@" | tail -1
}

# 목표 처리량을 고정한 측정 (11-9 용)
run_producer_at() {
  local thr="$1"; shift
  K kafka-producer-perf-test.sh \
    --topic "$TOPIC" --num-records $((RECORDS * 6 / 10)) --record-size "$SIZE" --throughput "$thr" \
    --producer-props bootstrap.servers=$BS "$@" | tail -1
}

# ===========================================================================
# [11-1] 측정 전용 토픽 — segment.bytes 를 반드시 덮어씁니다
# ===========================================================================
echo "### [11-1] orders-perf 생성 (6 파티션 / RF 3 / segment 256MiB / retention 1h)"
K kafka-topics.sh --bootstrap-server $BS \
  --create --topic "$TOPIC" --partitions 6 --replication-factor 3 \
  --config segment.bytes=268435456 \
  --config retention.ms=3600000 || true

# 학습용 축소값(log.segment.bytes=1MiB)을 그대로 두면 세그먼트가 수천 개 생겨
# "Kafka 처리량" 이 아니라 "파일 생성 비용" 을 재게 됩니다. 실측 1.4배 차이.
K kafka-topics.sh --bootstrap-server $BS --describe --topic "$TOPIC"

# ===========================================================================
# [11-2] 기본 측정 — 출력 읽는 법
# ===========================================================================
echo "### [11-2] 기본 측정 (acks=1). 첫 진행 줄은 워밍업이라 버립니다"
K kafka-producer-perf-test.sh \
  --topic "$TOPIC" --num-records "$RECORDS" --record-size "$SIZE" --throughput -1 \
  --producer-props bootstrap.servers=$BS acks=1

# ===========================================================================
# [11-3] 매트릭스 ① acks : 0 / 1 / all
#   고정: linger.ms=0, batch.size=16384, compression=none
# ===========================================================================
echo "### [11-3] acks 매트릭스"
for A in 0 1 all; do
  echo -n "acks=$A  → "
  run_producer acks=$A linger.ms=0 batch.size=16384 compression.type=none
done
# 해석: acks=0 이 all 의 약 2.8배. 단 acks=0 은 "사라져도 모르는" 세계입니다(Step 04·08).
#      acks=1 과 all 의 차이는 1.5배뿐이며, linger 를 올리면 거의 사라집니다.

# ===========================================================================
# [11-4] 매트릭스 ② linger.ms : 0 / 5 / 20 / 100
#   고정: acks=all, batch.size=65536, compression=none
# ===========================================================================
echo "### [11-4] linger.ms 매트릭스"
for L in 0 5 20 100; do
  echo -n "linger.ms=$L  → "
  run_producer acks=all linger.ms=$L batch.size=65536 compression.type=none
done
# 해석: 0 → 20 은 처리량 3배 + p99 개선(782→168ms). 트레이드오프가 아니라 그냥 이득.
#      20 → 100 부터가 진짜 선택: 처리량 11% 얻고 p99 3배 손해.

# ===========================================================================
# [11-5] 매트릭스 ③ batch.size : 16K / 64K / 256K
#   고정: acks=all, linger.ms=5
# ===========================================================================
echo "### [11-5] batch.size 매트릭스"
for B in 16384 65536 262144; do
  echo -n "batch.size=$B  → "
  run_producer acks=all linger.ms=5 batch.size=$B compression.type=none
done
# 해석: batch.size 와 linger.ms 는 "둘 중 먼저 도달하는 쪽" 이 발송을 트리거하므로
#      따로 튜닝하면 안 됩니다. batch 를 키워도 linger=0 이면 차기 전에 나갑니다.

# ===========================================================================
# [11-6] 매트릭스 ④ compression.type : none / lz4 / zstd / gzip
#   ⚠️ --record-size 는 랜덤 바이트라 압축률이 1.0 으로 나옵니다.
#      현실적인 압축률을 재려면 --payload-file 로 실제 데이터를 써야 합니다.
# ===========================================================================
echo "### [11-6] 압축 측정용 현실적인 페이로드 생성"
docker exec kafka-1 sh -c 'for i in $(seq 1 1000); do
  echo "{\"order_id\":\"O-$i\",\"customer_id\":\"C00$((i%10))\",\"amount\":39000,\"status\":\"CREATED\"}"
done > /tmp/payload.txt; wc -l /tmp/payload.txt'

echo "### [11-6] compression.type 매트릭스"
for C in none lz4 zstd gzip; do
  echo -n "compression=$C  → "
  K kafka-producer-perf-test.sh \
    --topic "$TOPIC" --num-records "$RECORDS" --throughput -1 \
    --payload-file /tmp/payload.txt \
    --producer-props bootstrap.servers=$BS acks=all linger.ms=20 \
                     batch.size=65536 compression.type=$C | tail -1
done
# 해석: lz4 는 처리량까지 1.33배 올립니다(네트워크 바이트가 1/3 로 줄어서).
#      gzip 은 CPU 3배 쓰고 처리량 절반. 쓸 이유가 거의 없습니다.

echo "### [11-6] 토픽의 compression.type 확인 — 'producer' 여야 재압축이 안 일어납니다"
K kafka-configs.sh --bootstrap-server $BS \
  --describe --entity-type topics --entity-name "$TOPIC"

# ===========================================================================
# [11-7] 매트릭스 ⑤ 레코드 크기 : 100B / 1KB / 10KB (총 전송량 고정)
# ===========================================================================
echo "### [11-7] 레코드 크기 매트릭스 — msg/s 와 MB/s 가 반대로 움직입니다"
for pair in "100 $((RECORDS * 10))" "1024 $RECORDS" "10240 $((RECORDS / 10))"; do
  set -- $pair
  echo -n "record-size=$1 (n=$2)  → "
  K kafka-producer-perf-test.sh \
    --topic "$TOPIC" --num-records "$2" --record-size "$1" --throughput -1 \
    --producer-props bootstrap.servers=$BS acks=all linger.ms=20 \
                     batch.size=65536 compression.type=lz4 | tail -1
done
# 해석: "초당 40만 건" 은 레코드 크기 없이는 아무 의미가 없습니다.
#      용량 산정은 반드시 MB/s 로 환산해서 네트워크·디스크 한계와 비교하십시오.

# ===========================================================================
# [11-9] 함정 A — 목표 처리량을 고정해야 지연이 지연이다
#   이 스텝에서 가장 중요한 측정입니다. 시간이 없어도 여기만은 돌리십시오.
# ===========================================================================
echo "### [11-9] 목표 처리량을 단계적으로 올려 '무릎 지점' 찾기"
for T in 20000 40000 60000 80000 -1; do
  echo "=== throughput=$T ==="
  run_producer_at "$T" acks=all linger.ms=5 batch.size=65536
done
# 해석: 목표 20000 에서 p99 11ms, 무제한에서 168ms. 같은 클러스터인데 15배 차이.
#      무제한의 지연은 대부분 "버퍼에서 순서를 기다린 시간" 입니다.
#      절차: ① -1 로 상한을 찾고 ② 상한의 50~70% 로 지연을 재고
#            ③ SLO 를 만족하는지 보고 ④ 무릎 지점을 찾아 그 절반을 용량으로.

# ===========================================================================
# [11-10] kafka-consumer-perf-test.sh — CSV 출력
# ===========================================================================
echo "### [11-10] 컨슈머 기본 측정"
K kafka-consumer-perf-test.sh \
  --bootstrap-server $BS --topic "$TOPIC" --messages "$RECORDS" \
  --group s11-perf --timeout 60000
# 해석: MB.sec 이 아니라 fetch.MB.sec 을 보십시오.
#      앞의 것은 rebalance.time.ms 가 섞여 있어 짧은 측정일수록 왜곡됩니다.

echo "### [11-10] fetch 설정 조합"
docker exec kafka-1 sh -c 'cat > /tmp/c1.properties <<EOF
fetch.min.bytes=100000
fetch.max.wait.ms=500
EOF
cat > /tmp/c2.properties <<EOF
fetch.min.bytes=100000
fetch.max.wait.ms=50
EOF
cat > /tmp/c3.properties <<EOF
max.partition.fetch.bytes=1048576
EOF
cat > /tmp/c4.properties <<EOF
max.poll.records=50
EOF'

i=0
for CFG in /tmp/c1.properties /tmp/c2.properties /tmp/c3.properties /tmp/c4.properties; do
  i=$((i+1))
  echo "=== $CFG ==="
  K kafka-consumer-perf-test.sh \
    --bootstrap-server $BS --topic "$TOPIC" --messages "$RECORDS" \
    --group "s11-perf-c$i" --consumer.config "$CFG" --timeout 60000 | tail -1
done
# 해석: fetch.min.bytes 1 → 100000 로 소비 속도 약 1.45배.
#      max.poll.records 500 → 50 은 오히려 느려집니다(poll 호출 오버헤드).

# ===========================================================================
# [11-11] 함정 B — fetch.min.bytes 가 만드는 "가끔 느린" 시스템
# ===========================================================================
echo "### [11-11] 한산한 토픽 준비"
K kafka-topics.sh --bootstrap-server $BS \
  --create --topic s11_quiet --partitions 1 --replication-factor 3 || true

# --- 터미널 2개 버전 ---------------------------------------------------------
# [터미널 B] 이 컨슈머를 먼저 띄우십시오:
#   docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
#     --bootstrap-server kafka-1:9092 --topic s11_quiet \
#     --consumer.config /tmp/c1.properties --property print.timestamp=true
# [터미널 A] 그다음 한 건 쓰고, 터미널 B 에 줄이 뜨기까지의 체감 시간을 보십시오:
#   echo '{"order_id":"O-1001","status":"CREATED"}' | docker exec -i kafka-1 \
#     /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server kafka-1:9092 --topic s11_quiet
# ---------------------------------------------------------------------------

# --- 단일 터미널 자동 측정 버전 ---------------------------------------------
measure_e2e() {
  local cfg="$1" label="$2"
  echo '{"order_id":"O-1001","status":"CREATED"}' | docker exec -i kafka-1 \
    /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server $BS --topic s11_quiet
  local t0 t1
  t0=$(date +%s%3N)
  docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
    --bootstrap-server $BS --topic s11_quiet --from-beginning --max-messages 1 \
    ${cfg:+--consumer.config "$cfg"} >/dev/null 2>&1 || true
  t1=$(date +%s%3N)
  echo "  $label : $((t1 - t0)) ms"
}

echo "### [11-11] 한 건이 도착하기까지의 시간"
measure_e2e ""                   "fetch.min.bytes=1 (기본값)"
measure_e2e /tmp/c1.properties   "fetch.min.bytes=100000, fetch.max.wait.ms=500"
measure_e2e /tmp/c2.properties   "fetch.min.bytes=100000, fetch.max.wait.ms=50"
# 해석: 2ms → 503ms → 51ms. 250배 차이인데 에러는 한 줄도 안 납니다.
#      낮에는 정상, 새벽에만 느립니다. 부하 테스트로는 절대 안 잡힙니다.
#      fetch.min.bytes 를 올릴 때는 fetch.max.wait.ms 를 반드시 함께 낮추십시오.

# ===========================================================================
# [11-12] 컨슈머 랙을 시간순으로 재기
# ===========================================================================
echo "### [11-12] 랙 추이 (10초 × 6회)"
for i in $(seq 1 6); do
  echo -n "$(date '+%H:%M:%S')  "
  K kafka-consumer-groups.sh --bootstrap-server $BS --describe --group s11-perf 2>/dev/null \
    | awk 'NR>1 && $6 ~ /^[0-9]+$/ {s+=$6} END {print "total-lag=" s+0}'
  sleep 10
done
# 해석: 줄고 있으면 감소율로 소진 시각을 계산할 수 있습니다.
#      늘고 있으면 대응해야 합니다. 그대로 두면 retention.ms 에 걸려
#      "읽지 못한 메시지가 삭제" 됩니다(Step 09). 랙이 무서운 진짜 이유입니다.

# ===========================================================================
# [11-13] 함정 C — 컨슈머 병렬성의 상한은 파티션 수
# ===========================================================================
echo "### [11-13] 현재 그룹의 파티션 할당 — 줄 수를 세십시오"
K kafka-consumer-groups.sh --bootstrap-server $BS --describe --group s11-perf || true
# 해석: 출력 줄 수 == 파티션 수 이고, 서로 다른 CONSUMER-ID 수도 그와 같다면
#      이미 상한입니다. 컨슈머를 더 띄워도 어느 줄에도 나타나지 않고 놉니다.
#      게다가 띄울 때마다 리밸런싱이 일어나 잠시 더 느려집니다.

# ===========================================================================
# [11-14] 브로커 측 — 페이지 캐시와 파일 디스크립터
# ===========================================================================
echo "### [11-14] 페이지 캐시 (buff/cache 가 클수록 컨슈머가 디스크에 안 갑니다)"
docker exec kafka-1 sh -c 'free -m' || true

echo "### [11-14] 파일 디스크립터 한도와 현재 사용량"
docker exec kafka-1 sh -c 'ulimit -n; ls /proc/1/fd | wc -l'
# 해석: 운영 권장 최소 100000. 부족하면 'Too many open files' 로 브로커가
#      새 커넥션도 못 받고 세그먼트도 못 만듭니다. 파티션 대량 증설 직후에 자주 터집니다.

# ===========================================================================
# [11-15] --print-metrics 해독
# ===========================================================================
echo "### [11-15] 주요 프로듀서 메트릭"
K kafka-producer-perf-test.sh \
  --topic "$TOPIC" --num-records $((RECORDS / 2)) --record-size "$SIZE" --throughput -1 \
  --producer-props bootstrap.servers=$BS acks=all linger.ms=20 batch.size=65536 \
  --print-metrics 2>/dev/null \
  | grep -E 'record-send-rate|batch-size-avg|request-latency-avg|records-per-request-avg|buffer-available-bytes|compression-rate-avg|record-queue-time-avg' || true
# 해석의 핵심:
#   request-latency-avg  = 브로커가 응답하는 데 걸린 순수 시간
#   record-queue-time-avg = 클라이언트 버퍼에서 대기한 시간
#   → 이 둘의 비율을 보십시오. 뒤쪽이 크면 브로커를 증설해도 효과가 없습니다.
#   batch-size-avg 가 batch.size 의 20% 미만이면 linger.ms 를 올릴 여지가 있습니다.
#   records-per-request-avg 가 1에 가까우면 배치가 전혀 안 되고 있다는 뜻입니다.
#   buffer-available-bytes 가 0 에 가까우면 send() 가 블로킹 중 = 11-9 의 함정 상태.

# ===========================================================================
# [11-17] 정리 (실습 마무리)
#   ⚠️ 이 정리를 건너뛰면 Step 12 실습 중 디스크가 부족할 수 있습니다.
# ===========================================================================
echo "### [11-17] 토픽 삭제"
K kafka-topics.sh --bootstrap-server $BS --delete --topic "$TOPIC"  || true
K kafka-topics.sh --bootstrap-server $BS --delete --topic s11_quiet || true

echo "### [11-17] 컨슈머 그룹 삭제"
K kafka-consumer-groups.sh --bootstrap-server $BS \
  --delete --group s11-perf --group s11-perf-c1 --group s11-perf-c2 \
           --group s11-perf-c3 --group s11-perf-c4 || true

echo "### [11-17] 남은 토픽 (공용 4개만 남아야 정상)"
K kafka-topics.sh --bootstrap-server $BS --list

echo "### [11-17] 디스크 회수 확인"
# 삭제 직후에는 -delete 접미사 디렉터리가 남아 있습니다.
# file.delete.delay.ms (기본 60초) 뒤에 실제로 지워집니다.
docker exec kafka-1 sh -c 'du -sh /var/lib/kafka/data | tail -1'
docker exec kafka-1 sh -c 'ls /var/lib/kafka/data | grep -- "-delete" || echo "  (대기 중인 삭제 없음)"'

echo "=== Step 11 practice 완료 ==="

#!/usr/bin/env bash
# ---------------------------------------------------------------------------
# Step 11 — 성능 튜닝 : 연습문제 (7문제)
#
# 실행법:
#   bash step-11-performance/exercise.sh
#   (권장) 편집기로 열어 `# 여기에 작성:` 을 채우고 한 문제씩 실행
#
# ⚠️ 이 스텝의 문제는 "정답 명령" 뿐 아니라 **여러분 환경의 숫자를 표에 적는 것**
#    까지가 문제입니다. 적지 않으면 문제 3 의 무릎 지점 계산을 할 수 없습니다.
#    교재의 숫자와 다른 것이 정상입니다.
#
# 정답: solution.sh
# ---------------------------------------------------------------------------
set -uo pipefail

BS=kafka-1:9092
TOPIC=s11_ex
RECORDS=${RECORDS:-300000}
SIZE=1024

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

# ===========================================================================
# 준비 — 문제지가 만들어 주는 것
#   본문의 orders-perf 는 6 파티션이지만, 여기는 3 파티션입니다.
#   문제 6 에서 "컨슈머 > 파티션" 상황을 컨슈머 4개만으로 만들기 위해서입니다.
# ===========================================================================
echo "### 준비: s11_ex 토픽 (3 파티션 / RF 3 / segment 256MiB / retention 1h)"
K kafka-topics.sh --bootstrap-server $BS \
  --create --topic $TOPIC --partitions 3 --replication-factor 3 \
  --config segment.bytes=268435456 --config retention.ms=3600000 2>/dev/null || true


# ===========================================================================
# 문제 1.
#   같은 조건에서 acks=1 과 acks=all 의 처리량과 p99 를 각각 측정하고,
#   몇 배 차이인지 계산하십시오.
#   고정 조건: --record-size 1024, --throughput -1, linger.ms=5, batch.size=65536
# ===========================================================================
echo "### 문제 1"
# (a) acks=1
# 여기에 작성:


# (b) acks=all
# 여기에 작성:


# 측정값 기록:
#   acks=1   : ______ msg/s , p99 ______ ms
#   acks=all : ______ msg/s , p99 ______ ms
#   배율     : ______ 배
#   (1.3~1.6배가 정상 범위입니다. 3배 이상이면 다른 문제를 의심하십시오)


# ===========================================================================
# 문제 2.
#   linger.ms 를 0 / 10 / 50 으로 바꿔가며 --print-metrics 로
#   records-per-request-avg 와 batch-size-avg 가 어떻게 변하는지 확인하십시오.
#
#   힌트: --print-metrics 는 200줄 넘게 쏟아냅니다. grep 필수.
#         grep -E 'records-per-request-avg|batch-size-avg'
# ===========================================================================
echo "### 문제 2"
for L in 0 10 50; do
  echo "=== linger.ms=$L ==="
  # 여기에 작성:

done

# 측정값 기록:
#   linger.ms=0  : records-per-request-avg ______ , batch-size-avg ______
#   linger.ms=10 : records-per-request-avg ______ , batch-size-avg ______
#   linger.ms=50 : records-per-request-avg ______ , batch-size-avg ______
#   → records-per-request-avg 가 몇 이하면 "배치가 사실상 없다" 고 봐야 합니까?
#     답:


# ===========================================================================
# 문제 3.
#   --throughput 을 단계적으로 올려 **무릎 지점**을 찾으십시오.
#   무릎 = "목표 처리량을 더 이상 달성하지 못하고 p99 가 급격히 꺾이는 지점"
#   찾은 뒤, 운영 용량으로 얼마를 제안하겠습니까?
#
#   힌트: 값의 범위는 여러분이 정합니다. 아래 배열을 채우십시오.
# ===========================================================================
echo "### 문제 3"
for T in 10000 ; do   # ← 여기에 단계를 추가하십시오 (예: 10000 20000 40000 60000 -1)
  echo "=== throughput=$T ==="
  # 여기에 작성:

done

# 측정값 기록:
#   목표 10000 : 달성 ______ , p99 ______ ms
#   목표 _____ : 달성 ______ , p99 ______ ms
#   목표 _____ : 달성 ______ , p99 ______ ms
#   목표 -1    : 달성 ______ , p99 ______ ms   ← 상한
#   무릎 지점      : ______ msg/s
#   제안 운영 용량 : ______ msg/s   (근거는?)


# ===========================================================================
# 문제 4.
#   compression.type 을 none / lz4 / gzip 으로 바꿔 처리량과
#   **브로커 CPU** 를 함께 재십시오.
#
#   ⚠️ 랜덤 바이트(--record-size)로 압축을 재면 압축률이 1.0 으로 나옵니다.
#      아래에서 문제지가 /tmp/ex_payload.txt 를 만들어 둡니다. --payload-file 을 쓰십시오.
#
#   # [터미널 B] 측정이 도는 동안 다른 창에서 계속 찍으십시오:
#   #   watch -n 1 'docker stats --no-stream --format "{{.Name}} {{.CPUPerc}}" kafka-1'
# ===========================================================================
echo "### 문제 4"
docker exec kafka-1 sh -c 'for i in $(seq 1 1000); do
  echo "{\"order_id\":\"O-$i\",\"customer_id\":\"C00$((i%10))\",\"amount\":39000,\"status\":\"CREATED\"}"
done > /tmp/ex_payload.txt'

for C in none lz4 gzip; do
  echo "=== compression=$C ==="
  # 여기에 작성:

done

# 측정값 기록:
#   none : ______ msg/s , 브로커 CPU ______ %
#   lz4  : ______ msg/s , 브로커 CPU ______ %
#   gzip : ______ msg/s , 브로커 CPU ______ %
#   → gzip 을 써야 할 이유가 하나라도 있습니까?
#     답:


# ===========================================================================
# 문제 5.
#   한산한 토픽에서 fetch.min.bytes=100000 컨슈머가
#   **한 건을 받기까지 몇 ms 가 걸리는지** 측정하고,
#   fetch.max.wait.ms 를 낮춰 개선하십시오.
#
#   힌트: date +%s%3N 으로 앞뒤 시각을 찍어 차이를 냅니다.
#         --max-messages 1 로 한 건만 읽고 종료하게 하면 측정이 쉽습니다.
# ===========================================================================
echo "### 문제 5"
K kafka-topics.sh --bootstrap-server $BS \
  --create --topic s11_ex_quiet --partitions 1 --replication-factor 3 2>/dev/null || true

docker exec kafka-1 sh -c 'cat > /tmp/ex_slow.properties <<EOF
fetch.min.bytes=100000
fetch.max.wait.ms=500
EOF
cat > /tmp/ex_fast.properties <<EOF
fetch.min.bytes=100000
fetch.max.wait.ms=50
EOF'

# (a) 메시지 1건을 쓰고, 기본 설정 컨슈머로 읽기까지의 ms 측정
echo '{"order_id":"O-9001","status":"CREATED"}' | docker exec -i kafka-1 \
  /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server $BS --topic s11_ex_quiet
T0=$(date +%s%3N)
# 여기에 작성:  (기본 설정으로 --max-messages 1 읽기)

T1=$(date +%s%3N)
echo "  기본값       : $((T1 - T0)) ms"

# (b) /tmp/ex_slow.properties (fetch.max.wait.ms=500) 로 같은 측정
# 여기에 작성:


# (c) /tmp/ex_fast.properties (fetch.max.wait.ms=50) 로 같은 측정
# 여기에 작성:


# 측정값 기록:
#   기본값(fetch.min.bytes=1)          : ______ ms
#   fetch.min.bytes=100000, wait=500  : ______ ms
#   fetch.min.bytes=100000, wait=50   : ______ ms
#   → 두 번째 숫자가 무엇과 정확히 일치합니까? 이게 왜 진단의 열쇠입니까?
#     답:


# ===========================================================================
# 문제 6.
#   s11_ex 는 파티션이 3개입니다. 컨슈머를 4개 띄우고
#   --describe 출력에서 **노는 컨슈머**를 찾아내십시오.
#
#   문제지가 백그라운드로 4개를 띄웁니다. 여러분은 --describe 를 채웁니다.
# ===========================================================================
echo "### 문제 6"
K kafka-producer-perf-test.sh --topic $TOPIC --num-records 200000 \
  --record-size 512 --throughput -1 \
  --producer-props bootstrap.servers=$BS acks=1 >/dev/null 2>&1

for n in 1 2 3 4; do
  docker exec -d kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
    --bootstrap-server $BS --topic $TOPIC --group s11-ex-many \
    --consumer-property client.id=ex-consumer-$n
done
sleep 12   # 리밸런싱이 끝나기를 기다립니다

# (a) 파티션 할당 확인
# 여기에 작성:


# 관찰 기록:
#   출력 줄 수                : ______
#   서로 다른 CONSUMER-ID 수  : ______
#   4번째 컨슈머는 어디에 나옵니까?
#     답:


# ===========================================================================
# 문제 7.
#   컨슈머 그룹 s11-ex-many 의 총 랙을 10초 간격으로 6회 찍고,
#   감소율로 소진 예상 시각을 계산하십시오.
#
#   힌트: --describe 출력의 6번째 컬럼이 LAG 입니다.
#         '-' 가 들어간 줄이 섞이므로 awk 정규식으로 숫자만 골라야 합니다.
# ===========================================================================
echo "### 문제 7"
for i in $(seq 1 6); do
  echo -n "$(date '+%H:%M:%S')  "
  # 여기에 작성:  (총 랙을 한 줄로 출력)

  sleep 10
done

# 측정값 기록:
#   t=0s   total-lag ______
#   t=50s  total-lag ______
#   감소율        : ______ 건/초
#   소진 예상까지 : ______ 초
#   → 만약 랙이 **늘고** 있었다면 이 계산이 성립합니까? 그때의 진짜 데드라인은?
#     답:


# ===========================================================================
# 정리 — 문제를 다 못 풀었어도 이 블록만 실행하면 환경이 깨끗해집니다
#   ⚠️ 문제 6 의 백그라운드 컨슈머를 안 죽이면 계속 돌면서 CPU 를 씁니다.
# ===========================================================================
echo "### 정리"
docker exec kafka-1 sh -c 'pkill -f kafka-console-consumer || true' 2>/dev/null || true
sleep 3
K kafka-topics.sh --bootstrap-server $BS --delete --topic $TOPIC        2>/dev/null || true
K kafka-topics.sh --bootstrap-server $BS --delete --topic s11_ex_quiet  2>/dev/null || true
K kafka-consumer-groups.sh --bootstrap-server $BS --delete --group s11-ex-many 2>/dev/null || true
echo "남은 토픽:"; K kafka-topics.sh --bootstrap-server $BS --list

#!/usr/bin/env bash
# ---------------------------------------------------------------------------
# Step 11 — 성능 튜닝 : 연습문제 정답 + 해설
#
# 실행법:
#   bash step-11-performance/solution.sh
#
# exercise.sh 를 직접 풀어 본 뒤에 여십시오.
# 숫자는 환경마다 다릅니다. 해설은 "배율" 과 "패턴" 에 초점을 맞춥니다.
# ---------------------------------------------------------------------------
set -uo pipefail

BS=kafka-1:9092
TOPIC=s11_ex
RECORDS=${RECORDS:-300000}
SIZE=1024

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

echo "### 준비"
K kafka-topics.sh --bootstrap-server $BS \
  --create --topic $TOPIC --partitions 3 --replication-factor 3 \
  --config segment.bytes=268435456 --config retention.ms=3600000 2>/dev/null || true


# ===========================================================================
# 정답 1 — acks=1 vs acks=all
# ===========================================================================
echo "### 정답 1"
for A in 1 all; do
  echo -n "acks=$A  → "
  K kafka-producer-perf-test.sh \
    --topic $TOPIC --num-records $RECORDS --record-size $SIZE --throughput -1 \
    --producer-props bootstrap.servers=$BS acks=$A linger.ms=5 batch.size=65536 | tail -1
done

# 해설:
#   예시 출력:
#     acks=1   → 300000 records sent, 74812.1 records/sec (73.06 MB/sec), ... 118 ms 99th, ...
#     acks=all → 300000 records sent, 51204.7 records/sec (50.00 MB/sec), ... 186 ms 99th, ...
#     배율: 74812 / 51205 = 약 1.46배
#
#   **1.3~1.6배가 정상 범위입니다.** acks=all 은 리더가 ISR 전원의 복제를 확인한 뒤
#   응답하므로, 브로커 간 왕복 한 번이 추가로 붙습니다. 로컬 도커 3대는 네트워크
#   지연이 거의 0 이라 이 정도로 끝납니다. 리전이 갈린 클러스터라면 훨씬 큽니다.
#
#   ⚠️ 3배 이상 차이가 난다면 acks 때문이 아닙니다. 다음을 의심하십시오:
#     ① ISR 이 축소되어 있다 (팔로워가 못 따라오는 중)
#     ② min.insync.replicas 를 못 채워 프로듀서가 재시도하고 있다
#     ③ 특정 브로커의 디스크가 느리다
#   확인:
#     K kafka-topics.sh --bootstrap-server $BS --describe --topic s11_ex | grep Isr
#   Isr 이 Replicas 보다 짧으면 ①·② 입니다. Step 08 로 돌아가십시오.
#
#   운영에서라면: 이 1.5배는 "유실 없음" 의 가격입니다. 결제·정산 파이프라인에서
#   이걸 아끼려고 acks=1 로 내리는 것은 거의 항상 잘못된 선택입니다.
K kafka-topics.sh --bootstrap-server $BS --describe --topic $TOPIC | grep Isr || true


# ===========================================================================
# 정답 2 — linger.ms 와 배치 메트릭
# ===========================================================================
echo "### 정답 2"
for L in 0 10 50; do
  echo "=== linger.ms=$L ==="
  K kafka-producer-perf-test.sh \
    --topic $TOPIC --num-records $RECORDS --record-size $SIZE --throughput -1 \
    --producer-props bootstrap.servers=$BS acks=all linger.ms=$L batch.size=65536 \
    --print-metrics 2>/dev/null \
    | grep -E 'records-per-request-avg|batch-size-avg|record-send-rate' || true
done

# 해설:
#   예시:
#     linger.ms=0  : records-per-request-avg 3.412  , batch-size-avg 4128.7
#     linger.ms=10 : records-per-request-avg 41.208 , batch-size-avg 42904.1
#     linger.ms=50 : records-per-request-avg 63.874 , batch-size-avg 65193.6
#
#   **records-per-request-avg 가 10 미만이면 배치가 사실상 없는 상태입니다.**
#   linger.ms=0 에서 3.4 라는 것은 요청 하나에 레코드 3~4개만 실려 나간다는 뜻이고,
#   요청 헤더와 왕복 비용을 레코드 3개가 나눠 지고 있다는 뜻입니다.
#
#   batch-size-avg 를 batch.size(65536) 와 나눠 보십시오.
#     linger=0  : 4128 / 65536 = 6%   → 배치가 거의 안 참. linger 를 올릴 여지 큼
#     linger=10 : 42904 / 65536 = 65% → 잘 차고 있음
#     linger=50 : 65193 / 65536 = 99% → 포화. 더 올려도 batch.size 가 상한
#   **이 한 숫자만 보면 "linger 를 올릴 여지가 있는가" 를 즉시 판단할 수 있습니다.**
#   99% 에 도달했다면 그다음은 linger 가 아니라 batch.size 를 키울 차례입니다.
#
#   운영에서라면: 이 메트릭들은 JMX 로도 나옵니다(kafka.producer:type=producer-metrics).
#   대시보드에 batch-size-avg 하나만 걸어 두어도 프로듀서 튜닝의 8할이 보입니다.


# ===========================================================================
# 정답 3 — 무릎 지점 찾기
# ===========================================================================
echo "### 정답 3"
for T in 10000 20000 40000 60000 -1; do
  echo -n "throughput=$T  → "
  K kafka-producer-perf-test.sh \
    --topic $TOPIC --num-records $RECORDS --record-size $SIZE --throughput $T \
    --producer-props bootstrap.servers=$BS acks=all linger.ms=5 batch.size=65536 | tail -1
done

# 해설:
#   예시:
#     10000 → 달성 9999.4  , p99 8 ms     ← 여유
#     20000 → 달성 19998.1 , p99 13 ms    ← 여유
#     40000 → 달성 39994.7 , p99 34 ms    ← 슬슬
#     60000 → 달성 52108.3 , p99 241 ms   ← 목표 미달 + p99 급등 = **무릎**
#     -1    → 달성 51204.7 , p99 186 ms   ← 상한
#
#   무릎을 판정하는 두 신호가 **동시에** 나타납니다.
#     ① 목표 처리량을 달성하지 못한다 (60000 요청 → 52108 달성)
#     ② p99 가 이전 단계 대비 한 자리 수 배로 뛴다 (34ms → 241ms)
#   둘 중 하나만 있으면 아직 무릎이 아닙니다.
#
#   제안 운영 용량: **무릎의 절반 이하**, 여기서는 25,000 msg/s 정도.
#   근거는 여유율이 아니라 **비선형성** 입니다. 무릎 근처에서 운영하면
#   트래픽이 20% 튀는 순간 p99 가 10배가 됩니다. 30% 여유는 여유가 아닙니다.
#   무릎 이전 구간은 지연이 완만하게 늘지만, 무릎 이후는 절벽입니다.
#
#   운영에서라면: 이 측정을 **분기마다 반복**하십시오. 파티션이 늘고 토픽이 늘고
#   다른 팀의 트래픽이 붙으면서 무릎은 계속 왼쪽으로 이동합니다.


# ===========================================================================
# 정답 4 — 압축과 브로커 CPU
# ===========================================================================
echo "### 정답 4"
docker exec kafka-1 sh -c 'for i in $(seq 1 1000); do
  echo "{\"order_id\":\"O-$i\",\"customer_id\":\"C00$((i%10))\",\"amount\":39000,\"status\":\"CREATED\"}"
done > /tmp/ex_payload.txt'

for C in none lz4 gzip; do
  echo "=== compression=$C ==="
  K kafka-producer-perf-test.sh \
    --topic $TOPIC --num-records $RECORDS --throughput -1 \
    --payload-file /tmp/ex_payload.txt \
    --producer-props bootstrap.servers=$BS acks=all linger.ms=20 \
                     batch.size=65536 compression.type=$C | tail -1
  docker stats --no-stream --format '  broker CPU: {{.CPUPerc}}' kafka-1
done

# 해설:
#   예시:
#     none : 126903 msg/s , 브로커 CPU 18%
#     lz4  : 168412 msg/s , 브로커 CPU 26%
#     gzip :  74208 msg/s , 브로커 CPU 61%
#
#   압축을 켰는데 **처리량이 올라간 것**이 이 문제의 핵심입니다. CPU 를 더 쓰는데도
#   빨라지는 이유는 네트워크로 흘려보낼 바이트가 1/3 로 줄었기 때문입니다.
#   Kafka 에서 병목은 대개 CPU 가 아니라 네트워크와 디스크입니다.
#
#   gzip 만 예외입니다. CPU 를 lz4 의 2배 이상 쓰면서 처리량은 절반이고,
#   압축률조차 zstd 보다 나쁩니다.
#   **결론 한 줄: lz4 아니면 zstd. gzip 은 레거시 호환 외에는 쓸 이유가 없습니다.**
#     - 지연/처리량 우선 → lz4
#     - 저장 공간 우선   → zstd
#
#   ⚠️ 랜덤 바이트(--record-size)로 이 측정을 하면 세 값이 거의 같게 나옵니다.
#      랜덤 데이터는 압축이 안 되기 때문입니다. 압축 벤치마크에서 가장 흔한 실수이며,
#      그래서 이 문제는 --payload-file 을 강제합니다.
#
#   운영에서라면: 토픽의 compression.type 은 'producer' 로 두십시오.
#   구체적 코덱을 지정해 두고 프로듀서가 다른 코덱을 쓰면 브로커가 **재압축**을 하고,
#   제로카피가 깨지면서 CPU 가 튑니다. 그런데 에러는 안 납니다.


# ===========================================================================
# 정답 5 — fetch.min.bytes 가 만드는 지연
# ===========================================================================
echo "### 정답 5"
K kafka-topics.sh --bootstrap-server $BS \
  --create --topic s11_ex_quiet --partitions 1 --replication-factor 3 2>/dev/null || true

docker exec kafka-1 sh -c 'cat > /tmp/ex_slow.properties <<EOF
fetch.min.bytes=100000
fetch.max.wait.ms=500
EOF
cat > /tmp/ex_fast.properties <<EOF
fetch.min.bytes=100000
fetch.max.wait.ms=50
EOF'

measure() {
  local cfg="$1" label="$2" t0 t1
  echo '{"order_id":"O-9001","status":"CREATED"}' | docker exec -i kafka-1 \
    /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server $BS --topic s11_ex_quiet
  t0=$(date +%s%3N)
  docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
    --bootstrap-server $BS --topic s11_ex_quiet --from-beginning --max-messages 1 \
    ${cfg:+--consumer.config "$cfg"} >/dev/null 2>&1 || true
  t1=$(date +%s%3N)
  echo "  $label : $((t1 - t0)) ms"
}

measure ""                      "기본값 (fetch.min.bytes=1)"
measure /tmp/ex_slow.properties "fetch.min.bytes=100000, wait=500"
measure /tmp/ex_fast.properties "fetch.min.bytes=100000, wait=50"

# 해설:
#   예시:
#     기본값                           :   2 ms  (+ JVM 기동 오버헤드)
#     fetch.min.bytes=100000, wait=500 : 503 ms
#     fetch.min.bytes=100000, wait=50  :  51 ms
#
#   두 번째 숫자가 **fetch.max.wait.ms 와 정확히 일치**합니다. 이것이 진단의 열쇠입니다.
#   브로커는 100 KB 가 모일 때까지 요청을 붙잡고 있다가, 안 모이면 wait 시간이
#   다 됐을 때 있는 만큼만 돌려줍니다. 한산한 토픽에서는 **항상** wait 시간을 다 씁니다.
#
#   현장에서 이렇게 판단하십시오:
#     "지연이 딱 500ms 근처에서 **일정하게** 나온다"
#     → 무작위 변동이 아니라 상수. 상수는 설정값입니다. fetch.max.wait.ms 를 보십시오.
#   지연이 들쭉날쭉하면 부하나 GC 이고, 상수에 붙어 있으면 타임아웃 설정입니다.
#
#   ⚠️ 이 함정이 지독한 이유는 **부하에 따라 증상이 나타났다 사라진다**는 것입니다.
#      낮(트래픽 많음) 정상 → 새벽(트래픽 적음) 느림 → 부하 테스트(트래픽 많음) 정상.
#      재현이 안 되고, 로그에 아무것도 없고, 알림도 안 울립니다. 500ms 는 에러가 아니니까요.
#
#   운영에서라면: fetch.min.bytes 와 fetch.max.wait.ms 는 **항상 세트로** 관리하십시오.
#   하나만 바꾸는 것은 절반만 설정한 것입니다. 최악 지연은 정확히 fetch.max.wait.ms 입니다.


# ===========================================================================
# 정답 6 — 컨슈머 > 파티션
# ===========================================================================
echo "### 정답 6"
K kafka-producer-perf-test.sh --topic $TOPIC --num-records 200000 \
  --record-size 512 --throughput -1 \
  --producer-props bootstrap.servers=$BS acks=1 >/dev/null 2>&1

for n in 1 2 3 4; do
  docker exec -d kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
    --bootstrap-server $BS --topic $TOPIC --group s11-ex-many \
    --consumer-property client.id=ex-consumer-$n
done
sleep 12

K kafka-consumer-groups.sh --bootstrap-server $BS --describe --group s11-ex-many || true

# 해설:
#   예시:
#     GROUP        TOPIC    PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG    CONSUMER-ID              ...
#     s11-ex-many  s11_ex   0          21044           66801           45757  consumer-s11-ex-many-1-...
#     s11-ex-many  s11_ex   1          20988           66740           45752  consumer-s11-ex-many-2-...
#     s11-ex-many  s11_ex   2          21107           66852           45745  consumer-s11-ex-many-3-...
#
#   **출력 줄이 3개뿐이고, 서로 다른 CONSUMER-ID 도 3개입니다.**
#   4번째 컨슈머는 어느 줄에도 나타나지 않습니다. 파티션을 하나도 할당받지
#   못했기 때문입니다. 프로세스는 살아 있고, 하트비트를 보내고,
#   로그에는 'Successfully joined group' 까지 찍혀 있습니다. 아무 일도 안 할 뿐입니다.
#
#   진단 절차 두 줄:
#     ① --describe 출력의 **줄 수**를 센다 → 파티션 수와 같으면 상한
#     ② 서로 다른 CONSUMER-ID 수를 센다   → 줄 수보다 적으면 아직 여유
#
#   장애 대응 중에 이걸 모르면 "파드를 계속 늘리는데 랙이 안 줄어든다" 로 30분을 씁니다.
#   게다가 파드를 늘릴 때마다 리밸런싱이 일어나 **오히려 잠시 더 느려집니다.**
#
#   해결은 파티션 증설이지만 대가가 있습니다. 같은 customer_id 가 다른 파티션으로
#   가면서 **키 기반 순서 보장이 그 순간 깨집니다**(Step 03). 그래서 파티션 수는
#   처음 설계할 때 여유 있게 잡는 것이 정석입니다.
#   임시 대응으로 컨슈머 내부 워커 풀 병렬화가 있지만, 오프셋 커밋 관리가
#   어려워지고 순서가 깨집니다(Step 06·07).


# ===========================================================================
# 정답 7 — 랙 추이와 소진 예상
# ===========================================================================
echo "### 정답 7"
for i in $(seq 1 6); do
  echo -n "$(date '+%H:%M:%S')  "
  K kafka-consumer-groups.sh --bootstrap-server $BS --describe --group s11-ex-many 2>/dev/null \
    | awk 'NR>1 && $6 ~ /^[0-9]+$/ {s+=$6} END {print "total-lag=" s+0}'
  sleep 10
done

# 해설:
#   예시:
#     10:22:01  total-lag=137254
#     10:22:11  total-lag=112038
#     10:22:21  total-lag=86901
#     10:22:31  total-lag=61744
#     10:22:41  total-lag=36612
#     10:22:51  total-lag=11488
#
#   계산:
#     감소량   = 137254 - 11488 = 125766 건
#     경과     = 50 초
#     감소율   = 125766 / 50 = 약 2515 건/초
#     소진까지 = 11488 / 2515 = 약 4.6 초
#
#   awk 정규식 $6 ~ /^[0-9]+$/ 이 중요합니다. LAG 컬럼에 '-' 가 들어간 줄
#   (컨슈머가 없는 파티션)이 섞이면 합계가 어긋납니다.
#
#   ⚠️ 랙이 **늘고** 있으면 이 계산은 성립하지 않습니다. 생산 > 소비이므로
#      소진 시각이라는 것 자체가 없습니다. 그때의 진짜 데드라인은 **retention.ms** 입니다.
#      가장 오래된 미처리 메시지가 보존 기간을 넘기는 순간, 그 메시지는
#      **읽히지 못한 채 삭제됩니다**(Step 09). 컨슈머는 조용히 다음 오프셋으로
#      건너뛰고, 에러는 나지 않습니다. 랙이 무서운 진짜 이유가 이것입니다.
#
#   운영에서라면: 랙 알림은 **절대값이 아니라 추세**로 거십시오.
#   "랙 10만 초과" 는 배치 시작 직후마다 울려서 곧 무시하게 됩니다.
#   "랙이 5분 연속 증가" 가 훨씬 유용한 알림입니다.


# ===========================================================================
# 정리
# ===========================================================================
echo "### 정리"
docker exec kafka-1 sh -c 'pkill -f kafka-console-consumer || true' 2>/dev/null || true
sleep 3
K kafka-topics.sh --bootstrap-server $BS --delete --topic $TOPIC       2>/dev/null || true
K kafka-topics.sh --bootstrap-server $BS --delete --topic s11_ex_quiet 2>/dev/null || true
K kafka-consumer-groups.sh --bootstrap-server $BS --delete --group s11-ex-many 2>/dev/null || true
echo "남은 토픽:"; K kafka-topics.sh --bootstrap-server $BS --list
echo "=== Step 11 solution 완료 ==="