#Step 09 — 보존과 로그 압축

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

alias kt='docker exec kafka-1 /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka-1:9092'
alias kconf='docker exec kafka-1 /opt/kafka/bin/kafka-configs.sh --bootstrap-server kafka-1:9092'

kt --create --topic s09_ret --partitions 1 --replication-factor 3 \
   --config retention.ms=10000

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

kt --describe --topic s09_ret

docker exec kafka-1 ls -la /var/lib/kafka/data/s09_ret-0

sleep 90
docker exec kafka-1 ls -la /var/lib/kafka/data/s09_ret-0

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

kconf --alter --entity-type topics --entity-name s09_ret \
      --add-config segment.ms=10000

docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka-1:9092 --topic s09_ret <<'EOF'
{"trigger":"roll"}
EOF

sleep 30
docker exec kafka-1 ls -la /var/lib/kafka/data/s09_ret-0

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

docker logs kafka-1 --since 3m 2>&1 | grep -E 'Rolled|Deleting segment|deleted'

kt --create --topic s09_bytes --partitions 6 --replication-factor 3 \
   --config retention.bytes=1073741824
kt --describe --topic s09_bytes

kt --create --topic s09_compact --partitions 1 --replication-factor 3 \
   --config cleanup.policy=compact \
   --config min.cleanable.dirty.ratio=0.01 \
   --config segment.ms=10000 \
   --config min.compaction.lag.ms=0

docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka-1:9092 --topic s09_compact \
  --property parse.key=true --property key.separator=: <<'EOF'
O-1001:{"order_id":"O-1001","status":"CREATED","at":"2024-03-11T10:00:00Z"}
O-1002:{"order_id":"O-1002","status":"CREATED","at":"2024-03-11T10:01:00Z"}
O-1001:{"order_id":"O-1001","status":"PAID","at":"2024-03-11T10:02:00Z"}
O-1003:{"order_id":"O-1003","status":"CREATED","at":"2024-03-11T10:03:00Z"}
O-1002:{"order_id":"O-1002","status":"PAID","at":"2024-03-11T10:04:00Z"}
O-1001:{"order_id":"O-1001","status":"SHIPPED","at":"2024-03-11T10:05:00Z"}
O-1004:{"order_id":"O-1004","status":"CREATED","at":"2024-03-11T10:06:00Z"}
O-1003:{"order_id":"O-1003","status":"PAID","at":"2024-03-11T10:07:00Z"}
O-1002:{"order_id":"O-1002","status":"SHIPPED","at":"2024-03-11T10:08:00Z"}
O-1001:{"order_id":"O-1001","status":"DELIVERED","at":"2024-03-11T10:09:00Z"}
O-1004:{"order_id":"O-1004","status":"PAID","at":"2024-03-11T10:10:00Z"}
O-1003:{"order_id":"O-1003","status":"SHIPPED","at":"2024-03-11T10:11:00Z"}
EOF

docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 --topic s09_compact --from-beginning \
  --property print.key=true --property print.offset=true --timeout-ms 5000

docker exec kafka-1 /opt/kafka/bin/kafka-dump-log.sh \
  --files /var/lib/kafka/data/s09_compact-0/00000000000000000000.log \
  --print-data-log | grep -E '^\| offset'

docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka-1:9092 --topic s09_compact \
  --property parse.key=true --property key.separator=: <<'EOF'
O-9999:{"order_id":"O-9999","status":"TRIGGER","at":"2024-03-11T10:12:00Z"}
EOF

sleep 40

docker logs kafka-1 --since 2m 2>&1 | grep -iE 'Cleaner|cleaned log'

docker exec kafka-1 /opt/kafka/bin/kafka-dump-log.sh \
  --files /var/lib/kafka/data/s09_compact-0/00000000000000000000.log \
  --print-data-log | grep -E '^\| offset'

docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 --topic s09_compact --from-beginning \
  --property print.key=true --property print.offset=true --timeout-ms 5000

docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka-1:9092 --topic s09_compact \
  --property parse.key=true --property key.separator=: \
  --property null.marker=NULL <<'EOF'
O-1001:NULL
EOF

docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka-1:9092 --topic s09_compact \
  --property parse.key=true --property key.separator=: <<'EOF'
O-8888:{"order_id":"O-8888","status":"TRIGGER2","at":"2024-03-11T10:14:00Z"}
EOF

sleep 15
docker exec kafka-1 /opt/kafka/bin/kafka-dump-log.sh \
  --files /var/lib/kafka/data/s09_compact-0/00000000000000000012.log \
  --print-data-log | grep -E '^\| offset'

kconf --alter --entity-type topics --entity-name s09_compact \
      --add-config delete.retention.ms=10000

docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka-1:9092 --topic s09_compact \
  --property parse.key=true --property key.separator=: <<'EOF'
O-7777:{"order_id":"O-7777","status":"TRIGGER3","at":"2024-03-11T10:16:00Z"}
EOF

sleep 45
docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 --topic s09_compact --from-beginning \
  --property print.key=true --property print.offset=true --timeout-ms 5000

kt --create --topic s09_both --partitions 1 --replication-factor 3 \
   --config cleanup.policy=compact,delete \
   --config retention.ms=604800000 \
   --config segment.ms=3600000
kt --describe --topic s09_both

kt --describe --topic __consumer_offsets | head -2

kt --delete --topic s09_ret
kt --delete --topic s09_bytes
kt --delete --topic s09_compact
kt --delete --topic s09_both
kt --list | grep '^s09_' || echo 'OK: s09_ 토픽 없음'

docker exec kafka-1 ls /var/lib/kafka/data/ | grep 's09_' || echo '디스크에서도 정리됨'

docker compose ps --format 'table {{.Name}}\t{{.Status}}'

#!/usr/bin/env bash
#
# Step 09 — 보존과 로그 압축 : 실습 스크립트
#
# 실행법:
#   cd kafka/docker && bash ../step-09-retention-compaction/practice.sh
#   (한 단계씩 보려면)  bash -x ../step-09-retention-compaction/practice.sh
#
# 이 스크립트는 브로커를 죽이지 않습니다. 대신 sleep 이 많습니다.
# 보존 삭제와 로그 압축은 브로커의 백그라운드 스레드가 하는 일이라
# "명령을 치면 즉시 결과" 가 나오지 않기 때문입니다.
# 통째로 돌리면 5~8분쯤 걸립니다. sleep 을 줄이면 관찰 대상이 아직 안 일어납니다.
#
set -euo pipefail

# ── 공통 헬퍼 ────────────────────────────────────────────────────────────
K() { docker exec kafka-1 /opt/kafka/bin/"$@"; }
BS=kafka-1:9092
DATA=/var/lib/kafka/data

# 대기 시간. 노트북이 느려 관찰이 안 되면 1.5배씩 올리세요.
WAIT_RET=90     # 보존 삭제 관찰용
WAIT_CLEAN=40   # 로그 클리너 관찰용

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

# ── [9-0] 실습 준비 ─────────────────────────────────────────────────────
hr "[9-0] 실습 준비 — Step 08 에서 죽인 브로커가 다 살아 있는지 확인"

ALIVE=$(K kafka-broker-api-versions.sh --bootstrap-server "$BS" | grep -cE '^kafka-[0-9]')
echo "살아 있는 브로커: ${ALIVE} (3 이어야 정상)"
[ "$ALIVE" -eq 3 ] || { echo 'FAIL: docker compose up -d 로 복구하세요'; exit 1; }

# ── [9-1] 대전제 ────────────────────────────────────────────────────────
hr "[9-1] Kafka 는 컨슈머가 읽어도 지우지 않는다"
echo "컨슈머가 100개든 0개든 디스크 사용량은 같습니다."
echo "지우는 것은 오직 cleanup.policy = delete | compact | compact,delete 입니다."

# ── [9-3] 액티브 세그먼트는 절대 안 지운다 ─────────────────────────────
hr "[9-3] 핵심 — 액티브 세그먼트는 절대 삭제되지 않는다"

K kafka-topics.sh --bootstrap-server "$BS" --create --topic s09_ret \
  --partitions 1 --replication-factor 3 --config retention.ms=10000

echo "-- 512건 (1KB × 512 = 약 512KB. segment.bytes=1MiB 에 못 미쳐 롤링 안 됨)"
K kafka-producer-perf-test.sh --topic s09_ret \
  --num-records 512 --record-size 1024 --throughput -1 \
  --producer-props bootstrap.servers="$BS" acks=all

echo
echo "-- 리더 확인 (리더 브로커의 디스크를 봐야 합니다)"
K kafka-topics.sh --bootstrap-server "$BS" --describe --topic s09_ret

echo
echo "########## ① BEFORE — 세그먼트 하나뿐 = 액티브 ##########"
docker exec kafka-1 ls -la "$DATA/s09_ret-0"

echo
echo "-- retention.ms=10000 인데 ${WAIT_RET}초를 기다립니다 (보존 시간의 9배)"
sleep "$WAIT_RET"

echo
echo "########## ② AFTER(${WAIT_RET}초) — 하나도 안 바뀝니다 ##########"
docker exec kafka-1 ls -la "$DATA/s09_ret-0"
echo "-- earliest 오프셋 (0 이면 한 건도 안 지워진 것)"
K kafka-get-offsets.sh --bootstrap-server "$BS" --topic s09_ret --time -2

echo
echo "-- segment.ms=10000 을 추가해 롤링을 유도합니다"
K kafka-configs.sh --bootstrap-server "$BS" --alter \
  --entity-type topics --entity-name s09_ret --add-config segment.ms=10000

echo "-- ★ 설정만으로는 롤링이 안 됩니다. 새 레코드가 들어와야 롤링 판정이 일어납니다."
docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server "$BS" --topic s09_ret <<'EOF'
{"trigger":"roll"}
EOF

sleep 30

echo
echo "########## ③ AFTER(segment.ms 적용) — 00000000000000000000.log 소멸 ##########"
docker exec kafka-1 ls -la "$DATA/s09_ret-0"

echo
echo "-- earliest / latest"
K kafka-get-offsets.sh --bootstrap-server "$BS" --topic s09_ret --time -2
K kafka-get-offsets.sh --bootstrap-server "$BS" --topic s09_ret --time -1

echo
echo "-- 브로커 로그: ① 롤링 → ② 보존 위반 판정 → ③ 파일 삭제"
docker logs kafka-1 --since 4m 2>&1 | grep -E 'Rolled|Deleting segment|Deleted log' || \
  echo '(로그를 못 찾았습니다 — --since 를 늘려 보세요)'

# ── [9-4] retention.bytes 는 파티션당 ──────────────────────────────────
hr "[9-4] retention.bytes 는 파티션당 크기다"

K kafka-topics.sh --bootstrap-server "$BS" --create --topic s09_bytes \
  --partitions 6 --replication-factor 3 --config retention.bytes=1073741824
K kafka-topics.sh --bootstrap-server "$BS" --describe --topic s09_bytes

echo
echo "  파티션당 상한 : 1 GiB"
echo "  파티션 수     : 6"
echo "  토픽 총량     : 6 GiB"
echo "  복제 계수     : 3"
echo "  클러스터 총량 : 18 GiB   ← 설정값의 18배"

# ── [9-5] cleanup.policy=compact ───────────────────────────────────────
hr "[9-5] cleanup.policy=compact — 같은 키의 옛 값이 사라진다"

K kafka-topics.sh --bootstrap-server "$BS" --create --topic s09_compact \
  --partitions 1 --replication-factor 3 \
  --config cleanup.policy=compact \
  --config min.cleanable.dirty.ratio=0.01 \
  --config segment.ms=10000 \
  --config min.compaction.lag.ms=0

echo "-- 키 4개(O-1001 ~ O-1004)로 12건"
docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server "$BS" --topic s09_compact \
  --property parse.key=true --property key.separator=: <<'EOF'
O-1001:{"order_id":"O-1001","status":"CREATED","at":"2024-03-11T10:00:00Z"}
O-1002:{"order_id":"O-1002","status":"CREATED","at":"2024-03-11T10:01:00Z"}
O-1001:{"order_id":"O-1001","status":"PAID","at":"2024-03-11T10:02:00Z"}
O-1003:{"order_id":"O-1003","status":"CREATED","at":"2024-03-11T10:03:00Z"}
O-1002:{"order_id":"O-1002","status":"PAID","at":"2024-03-11T10:04:00Z"}
O-1001:{"order_id":"O-1001","status":"SHIPPED","at":"2024-03-11T10:05:00Z"}
O-1004:{"order_id":"O-1004","status":"CREATED","at":"2024-03-11T10:06:00Z"}
O-1003:{"order_id":"O-1003","status":"PAID","at":"2024-03-11T10:07:00Z"}
O-1002:{"order_id":"O-1002","status":"SHIPPED","at":"2024-03-11T10:08:00Z"}
O-1001:{"order_id":"O-1001","status":"DELIVERED","at":"2024-03-11T10:09:00Z"}
O-1004:{"order_id":"O-1004","status":"PAID","at":"2024-03-11T10:10:00Z"}
O-1003:{"order_id":"O-1003","status":"SHIPPED","at":"2024-03-11T10:11:00Z"}
EOF

echo
echo "########## 압축 전 — 컨슈머 (12건) ##########"
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic s09_compact \
  --from-beginning --property print.key=true --property print.offset=true \
  --timeout-ms 5000 2>&1 || true

echo
echo "########## 압축 전 — dump-log ##########"
docker exec kafka-1 /opt/kafka/bin/kafka-dump-log.sh \
  --files "$DATA/s09_compact-0/00000000000000000000.log" \
  --print-data-log | grep -E '^\| offset' || true

echo
echo "-- 클리너도 액티브 세그먼트는 안 건드립니다. 트리거 1건으로 롤링 유도"
docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server "$BS" --topic s09_compact \
  --property parse.key=true --property key.separator=: <<'EOF'
O-9999:{"order_id":"O-9999","status":"TRIGGER","at":"2024-03-11T10:12:00Z"}
EOF

sleep "$WAIT_CLEAN"

echo
echo "-- 클리너 스레드 로그"
docker logs kafka-1 --since 2m 2>&1 | grep -iE 'Cleaner|cleaned log|size reduction' || \
  echo '(클리너 로그를 못 찾았습니다 — WAIT_CLEAN 을 늘려 보세요)'

echo
echo "########## 압축 후 — dump-log (오프셋 0~7 이 사라졌는지 보세요) ##########"
docker exec kafka-1 /opt/kafka/bin/kafka-dump-log.sh \
  --files "$DATA/s09_compact-0/00000000000000000000.log" \
  --print-data-log | grep -E '^\| offset' || true

echo
echo "########## 압축 후 — 컨슈머 (12건 → 5건) ##########"
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic s09_compact \
  --from-beginning --property print.key=true --property print.offset=true \
  --timeout-ms 5000 2>&1 || true

# ── [9-6] tombstone ────────────────────────────────────────────────────
hr "[9-6] tombstone — 값이 null 인 메시지 = 삭제 표식"

echo "-- 3.7 의 console-producer 는 --property null.marker=<문자열> 을 지원합니다"
docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server "$BS" --topic s09_compact \
  --property parse.key=true --property key.separator=: \
  --property null.marker=NULL <<'EOF'
O-1001:NULL
EOF

echo "-- 트리거로 롤링 유도"
docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server "$BS" --topic s09_compact \
  --property parse.key=true --property key.separator=: <<'EOF'
O-8888:{"order_id":"O-8888","status":"TRIGGER2","at":"2024-03-11T10:14:00Z"}
EOF
sleep 15

echo
echo "-- tombstone 은 valueSize: -1 이어야 합니다 (빈 문자열은 0)"
SEG=$(docker exec kafka-1 sh -c "ls $DATA/s09_compact-0/*.log | sort | tail -2 | head -1")
docker exec kafka-1 /opt/kafka/bin/kafka-dump-log.sh \
  --files "$SEG" --print-data-log | grep -E '^\| offset' || true

if docker exec kafka-1 /opt/kafka/bin/kafka-dump-log.sh --files "$SEG" \
     --print-data-log | grep -q 'valueSize: -1'; then
  echo "OK: valueSize: -1 확인 — 진짜 tombstone 입니다"
else
  echo "WARN: valueSize: -1 을 못 찾았습니다. null.marker 가 안 먹었을 수 있습니다."
fi

echo
echo "-- delete.retention.ms 를 10초로 줄여 tombstone 자체가 사라지는 것까지 봅니다"
K kafka-configs.sh --bootstrap-server "$BS" --alter \
  --entity-type topics --entity-name s09_compact --add-config delete.retention.ms=10000

docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server "$BS" --topic s09_compact \
  --property parse.key=true --property key.separator=: <<'EOF'
O-7777:{"order_id":"O-7777","status":"TRIGGER3","at":"2024-03-11T10:16:00Z"}
EOF

sleep 45

echo
echo "########## O-1001 이 완전히 사라졌는지 확인 ##########"
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic s09_compact \
  --from-beginning --property print.key=true --property print.offset=true \
  --timeout-ms 5000 2>&1 || true

echo "-- O-1001 이 위 출력에 없어야 정상입니다 (값도, tombstone 도 없음)"

# ── [9-8] compact,delete 조합 ─────────────────────────────────────────
hr "[9-8] cleanup.policy=compact,delete 조합"

K kafka-topics.sh --bootstrap-server "$BS" --create --topic s09_both \
  --partitions 1 --replication-factor 3 \
  --config cleanup.policy=compact,delete \
  --config retention.ms=604800000 \
  --config segment.ms=3600000
K kafka-topics.sh --bootstrap-server "$BS" --describe --topic s09_both

# ── [9-9] 압축 토픽 설계 기준 ─────────────────────────────────────────
hr "[9-9] 대표 사례 — __consumer_offsets 는 교과서적 압축 토픽"

K kafka-topics.sh --bootstrap-server "$BS" --describe --topic __consumer_offsets | head -2

echo
echo "  ① 키 필수                 : (그룹, 토픽, 파티션)      ✅"
echo "  ② 카디널리티 유한          : 그룹 × 파티션 수          ✅"
echo "  ③ 값이 전체 상태(델타 금지) : 커밋된 오프셋 절대값       ✅"
echo "  ④ 순서 의존 금지           : 마지막 커밋만 필요         ✅"

# ── [9-10] 정리 ──────────────────────────────────────────────────────
hr "[9-10] 정리 — s09_ 토픽 삭제"

for t in s09_ret s09_bytes s09_compact s09_both; do
  K kafka-topics.sh --bootstrap-server "$BS" --delete --topic "$t" || true
done
sleep 3

K kafka-topics.sh --bootstrap-server "$BS" --list | grep '^s09_' \
  && { echo 'FAIL: s09_ 토픽이 남아 있습니다'; exit 1; } || echo 'OK: s09_ 토픽 없음'

echo
echo "-- 토픽 삭제는 비동기입니다. 디스크에는 -delete 접미사 디렉터리가 잠시 남습니다"
docker exec kafka-1 ls "$DATA/" | grep 's09_' || echo '디스크에서도 정리됨'
echo "  (file.delete.delay.ms 기본 60초 뒤 자동 삭제됩니다)"

echo
docker compose ps --format 'table {{.Name}}\t{{.Status}}'
echo
echo "완료. → Step 10"

#!/usr/bin/env bash
#
# Step 09 — 보존과 로그 압축 : 연습문제 (7문제)
#
# 실행법:
#   cd kafka/docker && bash ../step-09-retention-compaction/exercise.sh
#   각 문제의 "# 여기에 작성:" 아래를 직접 채우고 다시 실행하세요.
#
# 브로커를 죽이지 않으므로 안전하지만 sleep 이 많습니다.
# 파일 끝의 정리 블록은 꼭 실행하세요 (압축 토픽을 남기면 클리너가 계속 돕니다).
#
set -euo pipefail

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

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


# ══════════════════════════════════════════════════════════════════════
hr "문제 1 — retention.ms 만 걸면 왜 안 지워지는가"
# ══════════════════════════════════════════════════════════════════════
K kafka-topics.sh --bootstrap-server "$BS" --create --if-not-exists \
  --topic s09ex_ret --partitions 1 --replication-factor 3 --config retention.ms=5000

K kafka-producer-perf-test.sh --topic s09ex_ret \
  --num-records 300 --record-size 1024 --throughput -1 \
  --producer-props bootstrap.servers="$BS" acks=all

echo "-- BEFORE"
docker exec kafka-1 ls -la "$DATA/s09ex_ret-0"

# Q1-a. 90초를 기다린 뒤 같은 ls -la 를 다시 찍고, earliest 오프셋도 확인하세요.
#       파일이 지워졌습니까?
#
# 여기에 작성:


# Q1-b. 지워지지 않았다면 그 이유를 한 줄로 설명하는 echo 를 작성하세요.
#       힌트: 세그먼트가 몇 개입니까? 그 세그먼트의 상태는?
#
# 여기에 작성:



# ══════════════════════════════════════════════════════════════════════
hr "문제 2 — 실제로 지워지게 만드는 최소한의 추가 설정"
# ══════════════════════════════════════════════════════════════════════
# Q2-a. s09ex_ret 이 실제로 삭제되도록 설정을 하나만 추가하세요.
#       (kafka-configs.sh --alter --add-config ...)
#
# 여기에 작성:


# Q2-b. 설정을 바꾼 직후 ls -la 를 찍어 보세요. 변화가 있습니까?
#       없다면 무엇이 더 필요합니까? 그것을 실행하세요.
#
# 여기에 작성:


# Q2-c. 30초 대기 후 ls -la 와 earliest 오프셋을 확인하고,
#       브로커 로그에서 롤링/삭제 메시지를 grep 하세요.
#
# 여기에 작성:



# ══════════════════════════════════════════════════════════════════════
hr "문제 3 — retention.bytes 로 클러스터 디스크 사용량 계산"
# ══════════════════════════════════════════════════════════════════════
K kafka-topics.sh --bootstrap-server "$BS" --create --if-not-exists \
  --topic s09ex_bytes --partitions 12 --replication-factor 3 \
  --config retention.bytes=536870912          # 512 MiB

K kafka-topics.sh --bootstrap-server "$BS" --describe --topic s09ex_bytes | head -1

# Q3. 이 토픽이 클러스터 전체에서 최대 몇 GiB 를 쓸 수 있습니까?
#     계산 과정을 echo 로 출력하세요.
#     힌트: retention.bytes 의 단위는 무엇입니까?
#
# 여기에 작성:



# ══════════════════════════════════════════════════════════════════════
hr "문제 4 — 압축으로 오프셋에 구멍이 생기는 것 확인"
# ══════════════════════════════════════════════════════════════════════
K kafka-topics.sh --bootstrap-server "$BS" --create --if-not-exists \
  --topic s09ex_compact --partitions 1 --replication-factor 3 \
  --config cleanup.policy=compact \
  --config min.cleanable.dirty.ratio=0.01 \
  --config segment.ms=10000 \
  --config min.compaction.lag.ms=0

# Q4-a. 키 5개(K-1 ~ K-5)로 총 20건을 쓰세요.
#       각 키가 4번씩 나오도록, 값은 {"k":"K-N","v":M} 형태로 하세요.
#       (--property parse.key=true --property key.separator=: 를 씁니다)
#
# 여기에 작성:


# Q4-b. 압축 전 레코드 목록을 kafka-dump-log.sh 로 출력하세요.
#       세그먼트 파일 이름은 "첫 오프셋을 20자리로 채운 것" 입니다.
#
# 여기에 작성:


# Q4-c. 트리거 메시지 1건으로 롤링을 유도하고 40초 기다린 뒤,
#       압축 후 레코드 목록을 다시 출력하세요.
#       몇 건이 남았고 오프셋은 몇 번부터입니까?
#
# 여기에 작성:


# Q4-d. 클리너 로그에서 "size reduction" 이 포함된 줄을 찾아 출력하세요.
#
# 여기에 작성:



# ══════════════════════════════════════════════════════════════════════
hr "문제 5 — tombstone(valueSize: -1) 과 빈 문자열(valueSize: 0) 구별"
# ══════════════════════════════════════════════════════════════════════
# Q5-a. s09ex_compact 에 다음 두 건을 쓰세요.
#         - K-1 에 대한 진짜 tombstone (값이 null)
#         - K-2 에 대한 빈 문자열 값
#       한 번의 console-producer 호출로 둘 다 보내려면
#       --property null.marker=NULL 을 쓰고, 빈 문자열은 그냥 비워 둡니다.
#
# 여기에 작성:


# Q5-b. 롤링을 유도한 뒤 dump-log 로 두 레코드를 확인하세요.
#       valueSize: -1 과 valueSize: 0 이 나란히 보여야 정답입니다.
#
# 여기에 작성:


# Q5-c. 압축 후 컨슈머로 읽었을 때 K-1 과 K-2 중 무엇이 남습니까?
#       확인하고, 왜 그런지 echo 로 설명하세요.
#
# 여기에 작성:



# ══════════════════════════════════════════════════════════════════════
hr "문제 6 — 압축 토픽에 키 없는 레코드가 들어가면"
# ══════════════════════════════════════════════════════════════════════
K kafka-topics.sh --bootstrap-server "$BS" --create --if-not-exists \
  --topic s09ex_nullkey --partitions 1 --replication-factor 3 \
  --config cleanup.policy=compact \
  --config min.cleanable.dirty.ratio=0.01 \
  --config segment.ms=10000

# Q6-a. 키 있는 레코드 5건을 먼저 넣으세요. (키 A, B 를 섞어서)
#
# 여기에 작성:


# Q6-b. 이번에는 parse.key 를 쓰지 않고(= 키 없이) 3건을 넣으세요.
#
# 여기에 작성:


# Q6-c. 롤링을 유도하고 40초 기다린 뒤, 브로커 로그에서
#       클리너의 경고 메시지를 찾아 출력하세요.
#       힌트: grep -i 'skipping compaction'
#       이 경고는 에러입니까, 경고입니까? 어떤 결과를 낳습니까?
#
# 여기에 작성:



# ══════════════════════════════════════════════════════════════════════
hr "문제 7 — 압축 적합성 판정 (명령이 아니라 판정 문제)"
# ══════════════════════════════════════════════════════════════════════
# 설계 기준 4가지:
#   ① 키가 반드시 있어야 한다
#   ② 키 카디널리티가 유한해야 한다
#   ③ 값이 "전체 상태" 여야 한다 (델타 금지)
#   ④ 순서 의존 처리를 하면 안 된다
#
# 아래 네 토픽이 cleanup.policy=compact 에 적합한지 판정하고,
# 부적합하면 어느 조건을 위반하는지 주석으로 채우세요.
#
# (A) user-profile-snapshot
#     키 = user_id, 값 = 사용자 프로필 전체 JSON.
#     "현재 프로필" 을 조회하는 캐시 구축용.
#     판정:  # 여기에 작성:
#
# (B) page-view-log
#     키 = 없음, 값 = {"url":..., "at":...}.
#     시간대별 페이지뷰 집계에 사용.
#     판정:  # 여기에 작성:
#
# (C) account-balance-delta
#     키 = account_id, 값 = {"delta": +10000} 형태의 증감액.
#     컨슈머가 delta 를 누적해 잔액을 계산.
#     판정:  # 여기에 작성:
#
# (D) session-heartbeat
#     키 = session_id, 값 = {"last_seen": ...} 전체 상태.
#     세션은 계속 새로 생기고 만료된다.
#     판정:  # 여기에 작성:
#
# Q7. 위 네 판정을 echo 로 출력하세요.
#
# 여기에 작성:



# ══════════════════════════════════════════════════════════════════════
hr "뒷정리 — s09ex_ 토픽 삭제"
# ══════════════════════════════════════════════════════════════════════
for t in s09ex_ret s09ex_bytes s09ex_compact s09ex_nullkey; 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" --list | grep '^s09ex_' \
  && { echo 'FAIL: s09ex_ 토픽이 남아 있습니다'; exit 1; } || echo 'OK: s09ex_ 토픽 없음'

docker compose ps --format 'table {{.Name}}\t{{.Status}}'
echo "뒷정리 완료."

#!/usr/bin/env bash
#
# Step 09 — 보존과 로그 압축 : 연습문제 정답과 해설
#
# 실행법:
#   cd kafka/docker && bash ../step-09-retention-compaction/solution.sh
#
# exercise.sh 를 먼저 풀어 본 뒤에 여세요.
#
set -euo pipefail

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

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


# ══════════════════════════════════════════════════════════════════════
hr "정답 1 — retention.ms 만 걸면 안 지워진다"
# ══════════════════════════════════════════════════════════════════════
K kafka-topics.sh --bootstrap-server "$BS" --create --if-not-exists \
  --topic s09ex_ret --partitions 1 --replication-factor 3 --config retention.ms=5000

K kafka-producer-perf-test.sh --topic s09ex_ret \
  --num-records 300 --record-size 1024 --throughput -1 \
  --producer-props bootstrap.servers="$BS" acks=all

echo "-- BEFORE"
docker exec kafka-1 ls -la "$DATA/s09ex_ret-0"

echo "-- 90초 대기 (보존 시간 5초의 18배)"
sleep 90

echo "-- AFTER: 하나도 안 바뀝니다"
docker exec kafka-1 ls -la "$DATA/s09ex_ret-0"
K kafka-get-offsets.sh --bootstrap-server "$BS" --topic s09ex_ret --time -2

echo
echo "정답: 안 지워집니다. 세그먼트가 하나뿐이고 그것이 액티브 세그먼트이기 때문입니다."

# 해설:
#   보존 삭제는 "세그먼트 파일 단위" 로만 일어나고,
#   ★ 액티브 세그먼트(지금 쓰고 있는 파일)는 절대 삭제 대상이 아닙니다.
#
#   300건 × 1KB = 약 300KB 이고 segment.bytes 는 1MiB 이므로 롤링되지 않았습니다.
#   segment.ms 기본값은 7일이므로 시간으로도 롤링되지 않습니다.
#   → 닫힌 세그먼트가 0개 → 삭제 후보가 0개 → 아무 일도 안 일어납니다.
#
#   ★ 브로커 로그가 조용한 것이 정상입니다.
#     log.retention.check.interval.ms=10000 이므로 브로커는 10초마다 검사하고 있습니다.
#     다만 매번 "닫힌 세그먼트 중 보존 위반" 을 찾는데 닫힌 세그먼트가 없어서
#     조용히 아무것도 안 하고 돌아갑니다. 에러도 경고도 없습니다.
#
#   이것이 실무에서 "개인정보 보존기간 1일로 설정했다" 고 보고했는데
#   실제로는 8일 넘게 남아 있는 상황의 정체입니다.
#   트래픽이 적은 토픽일수록 이 격차가 큽니다.


# ══════════════════════════════════════════════════════════════════════
hr "정답 2 — segment.ms 하나를 추가한다 (그리고 레코드를 하나 더 넣는다)"
# ══════════════════════════════════════════════════════════════════════
K kafka-configs.sh --bootstrap-server "$BS" --alter \
  --entity-type topics --entity-name s09ex_ret --add-config segment.ms=5000

echo "-- 설정 변경 직후 ls -la : 아직 아무 변화가 없습니다"
docker exec kafka-1 ls -la "$DATA/s09ex_ret-0"

echo
echo "-- ★ 새 레코드를 하나 넣어야 롤링 판정이 일어납니다"
docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server "$BS" --topic s09ex_ret <<'EOF'
{"trigger":"roll"}
EOF

sleep 30

echo "-- AFTER: 00000000000000000000.log 가 사라졌습니다"
docker exec kafka-1 ls -la "$DATA/s09ex_ret-0"
K kafka-get-offsets.sh --bootstrap-server "$BS" --topic s09ex_ret --time -2

echo
docker logs kafka-1 --since 3m 2>&1 | grep -E 'Rolled|Deleting segment|Deleted log' || true

# 해설:
#   정답은 segment.ms 하나입니다. retention.ms 는 이미 걸려 있으므로
#   부족한 것은 "닫힌 세그먼트를 만들어 주는 것" 뿐입니다.
#
#   ★ 그런데 설정만 바꿔서는 아무 일도 안 일어납니다. 이것이 문제 2의 진짜 함정입니다.
#     롤링 판정은 브로커가 append 경로에서 합니다. 즉 "새 레코드가 들어올 때"
#     "이 세그먼트가 segment.bytes 를 넘었나? segment.ms 가 지났나?" 를 검사합니다.
#     레코드가 안 들어오면 그 검사 자체가 실행되지 않습니다.
#
#   그래서 트래픽이 완전히 멈춘 토픽은 segment.ms 를 걸어도 롤링되지 않고,
#   결과적으로 보존 삭제도 되지 않습니다. 마지막 세그먼트는 영원히 남습니다.
#
#   ★ 브로커 로그의 세 줄이 순서를 정확히 말해 줍니다:
#     ① Rolled new log segment at offset 301
#     ② Deleting segment ... due to retention time 5000ms breach
#        based on the largest record timestamp in the segment
#     ③ Deleted log .../00000000000000000000.log.deleted
#     ①이 없으면 ②도 ③도 없습니다.
#
#   ②의 "based on the largest record timestamp" 도 중요합니다.
#   삭제 판정은 파일 수정 시각이 아니라 세그먼트 안의 최대 레코드 타임스탬프 기준입니다.
#   프로듀서가 CreateTime 을 과거로 조작하면(마이그레이션에서 흔합니다)
#   넣자마자 보존 위반 상태가 되고, 미래로 조작하면 영원히 안 지워집니다.
#   message.timestamp.type=LogAppendTime 으로 바꾸면 브로커가 수신 시각으로 덮어씁니다.
#
#   ★ 트레이드오프: segment.ms 를 짧게 하면 세그먼트 파일 수가 늘어
#     파일 핸들과 인덱스 메모리를 더 씁니다. .index 와 .timeindex 는
#     세그먼트마다 최대 10MiB 씩 미리 할당되므로 오버헤드가 작지 않습니다.
#     "정확한 보존 기간" 과 "파일 수" 의 트레이드오프입니다.


# ══════════════════════════════════════════════════════════════════════
hr "정답 3 — retention.bytes 는 파티션당 크기다"
# ══════════════════════════════════════════════════════════════════════
K kafka-topics.sh --bootstrap-server "$BS" --create --if-not-exists \
  --topic s09ex_bytes --partitions 12 --replication-factor 3 \
  --config retention.bytes=536870912

cat <<'CALC'
  retention.bytes = 536870912 = 512 MiB   ← ★ 파티션당 (토픽 전체가 아님)
  파티션 수       = 12
  ────────────────────────────────────────
  토픽 데이터 총량 = 512 MiB × 12 = 6 GiB
  복제 계수 RF     = 3
  ────────────────────────────────────────
  클러스터 전체    = 6 GiB × 3 = 18 GiB

  여기에 실무 여유율을 더하면:
    + 액티브 세그먼트 (파티션·리플리카마다 최대 1MiB)
    + 인덱스 파일 (.index + .timeindex = 세그먼트당 최대 20MiB 사전 할당)
    + 삭제 지연분 (file.delete.delay.ms)
  → 18 GiB × 1.3 ≈ 23.4 GiB 를 산정하는 것이 안전합니다.

  산정식:  필요 디스크 = retention.bytes × 파티션 수 × RF × 1.3
CALC

# 해설:
#   가장 흔한 오답은 "512 MiB" 또는 "1.5 GiB" 입니다.
#   retention.bytes 를 토픽 전체 상한으로 오해했기 때문입니다.
#
#   ★ 더 위험한 것은 파티션 수를 나중에 늘릴 때입니다.
#     12개 → 24개로 늘리면 아무도 retention.bytes 를 안 건드렸는데
#     디스크 사용량 상한이 그 순간 36 GiB 로 두 배가 됩니다.
#     파티션 증설은 Step 03 에서 봤듯 순서 보장도 깨뜨리는데, 디스크까지 건드립니다.
#
#   ★ 그리고 retention.ms 와 retention.bytes 는 OR 조건입니다.
#     "둘 다 만족해야 삭제" 가 아니라 "둘 중 하나만 걸려도 삭제" 입니다.
#     retention.ms=7일 + retention.bytes=512MiB 인 토픽에 트래픽이 몰려
#     512 MiB 를 하루 만에 채우면 하루치만 남습니다.
#     "7일치 보관" 이라고 믿고 있던 재처리 계획이 그날 조용히 무너집니다.
#     시간 기준 보존이 계약 사항이면 retention.bytes 는 -1 로 두고
#     디스크는 용량 알람으로 관리하세요.


# ══════════════════════════════════════════════════════════════════════
hr "정답 4 — 압축은 오프셋에 구멍을 낸다"
# ══════════════════════════════════════════════════════════════════════
K kafka-topics.sh --bootstrap-server "$BS" --create --if-not-exists \
  --topic s09ex_compact --partitions 1 --replication-factor 3 \
  --config cleanup.policy=compact \
  --config min.cleanable.dirty.ratio=0.01 \
  --config segment.ms=10000 \
  --config min.compaction.lag.ms=0

# 4-a: 키 5개 × 4회 = 20건
{
  for v in 1 2 3 4; do
    for k in 1 2 3 4 5; do
      echo "K-${k}:{\"k\":\"K-${k}\",\"v\":${v}}"
    done
  done
} | docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
      --bootstrap-server "$BS" --topic s09ex_compact \
      --property parse.key=true --property key.separator=:

# 4-b: 압축 전
echo
echo "########## 압축 전 (20건) ##########"
docker exec kafka-1 /opt/kafka/bin/kafka-dump-log.sh \
  --files "$DATA/s09ex_compact-0/00000000000000000000.log" \
  --print-data-log | grep -E '^\| offset' || true

# 4-c: 롤링 유도 + 대기
docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server "$BS" --topic s09ex_compact \
  --property parse.key=true --property key.separator=: <<'EOF'
K-TRIGGER:{"k":"K-TRIGGER","v":0}
EOF
sleep 40

echo
echo "########## 압축 후 (5건, 오프셋 15~19) ##########"
docker exec kafka-1 /opt/kafka/bin/kafka-dump-log.sh \
  --files "$DATA/s09ex_compact-0/00000000000000000000.log" \
  --print-data-log | grep -E '^\| offset' || true

# 4-d
echo
docker logs kafka-1 --since 2m 2>&1 | grep -iE 'size reduction|Start size|End size|dirty section' || \
  echo '(클리너 로그를 못 찾았습니다)'

# 해설:
#   압축 후 5건이 남고 오프셋은 15, 16, 17, 18, 19 입니다.
#   각 키의 마지막 쓰기(v=4 라운드)가 오프셋 15~19 에 있었기 때문입니다.
#   클리너 로그는 이렇게 나옵니다:
#     Start size: 20 messages / End size: 5 messages / 75.0% size reduction
#     dirty section = [0, 20]
#
#   ★ 핵심은 오프셋 번호가 재부여되지 않는다는 것입니다.
#     0~14 는 통째로 사라지고, 살아남은 레코드는 원래 번호를 그대로 씁니다.
#     압축은 "로그를 다시 쓰는" 것이 아니라 "안 쓰는 레코드를 빼고 옮겨 담는" 것입니다.
#
#   ★ 그래서 압축 토픽에서 "오프셋 + 1 = 다음 레코드" 를 가정하면 안 됩니다.
#     - seek(partition, 3) 은 에러가 아니라 "3 이상의 첫 레코드" 인 오프셋 15 로 갑니다.
#     - LOG-END-OFFSET 과 실제 레코드 수가 다르므로,
#       "오프셋 개수 = 메시지 개수" 를 가정한 랙 계산이 부정확해집니다.
#       압축 토픽의 랙은 "몇 건 남았나" 가 아니라 "얼마나 뒤처졌나" 로만 읽어야 합니다.
#
#   ★ 그리고 압축은 "최신 값만 보인다" 를 보장하지 않습니다.
#     위 출력에도 액티브 세그먼트의 K-TRIGGER 는 압축되지 않은 채 있습니다.
#     트래픽이 있는 실제 토픽이라면 액티브 세그먼트에 같은 키가 수십 개 있는 게 정상입니다.
#     컨슈머는 반드시 오프셋 순서대로 끝까지 읽고 같은 키를 만나면 덮어써야 합니다:
#       for (record : poll()) cache.put(record.key(), record.value());   // 나중 것이 이긴다
#     첫 번째로 만난 값을 쓰고 멈추면 옛 상태를 잡습니다. Kafka Streams 의 KTable 이
#     정확히 이 덮어쓰기 방식으로 동작합니다(Step 13).


# ══════════════════════════════════════════════════════════════════════
hr "정답 5 — tombstone(-1) 과 빈 문자열(0)"
# ══════════════════════════════════════════════════════════════════════
docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server "$BS" --topic s09ex_compact \
  --property parse.key=true --property key.separator=: \
  --property null.marker=NULL <<'EOF'
K-1:NULL
K-2:
EOF

docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server "$BS" --topic s09ex_compact \
  --property parse.key=true --property key.separator=: <<'EOF'
K-TRIGGER2:{"k":"K-TRIGGER2","v":0}
EOF
sleep 20

echo
echo "-- valueSize: -1 (tombstone) 과 valueSize: 0 (빈 문자열) 이 나란히 보여야 합니다"
SEG=$(docker exec kafka-1 sh -c "ls $DATA/s09ex_compact-0/*.log | sort | tail -2 | head -1")
docker exec kafka-1 /opt/kafka/bin/kafka-dump-log.sh --files "$SEG" \
  --print-data-log | grep -E '^\| offset' || true

sleep 40
echo
echo "-- 압축 후 컨슈머: K-1 은 사라지고 K-2 는 빈 값으로 남습니다"
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic s09ex_compact \
  --from-beginning --property print.key=true --property print.offset=true \
  --timeout-ms 5000 2>&1 || true

# 해설:
#   K-1 은 사라지고, K-2 는 "빈 값" 으로 남습니다.
#
#   ★ 값이 null 인 레코드만 tombstone 입니다. 빈 문자열은 유효한 값입니다.
#     dump-log 에서:
#       valueSize: -1  → tombstone. 압축 시 그 키를 통째로 제거
#       valueSize:  0  → 길이 0 인 문자열. 그 키의 "최신 값" 으로 살아남음
#
#   실무 사고 시나리오가 정확히 여기서 나옵니다.
#   "삭제 이벤트를 보냈는데 왜 안 지워지지?" → 값을 빈 문자열로 보냈기 때문입니다.
#   에러는 없습니다. 컨슈머는 빈 값을 "유효한 상태" 로 받아 캐시에 넣습니다.
#   그러면 삭제된 줄 알았던 레코드가 빈 껍데기로 살아 있게 됩니다.
#   JSON 파서가 빈 문자열에서 예외를 던지면 그나마 다행이고,
#   null 체크만 하는 코드라면 아무도 눈치채지 못합니다.
#
#   ★ 3.7 의 kafka-console-producer.sh 는 --property null.marker=<문자열> 을 지원합니다.
#     지정한 문자열이 값 자리에 오면 그 레코드를 null 값으로 보냅니다.
#     옛 버전에서는 kafkacat -Z 를 쓰거나 자바로
#     new ProducerRecord<>(topic, key, null) 을 보내야 했습니다.
#
#   ★ delete.retention.ms (기본 24시간) 가 왜 필요한가:
#     tombstone 은 두 단계로 처리됩니다.
#       ① 1차 압축: 그 키의 옛 값들이 사라지고 tombstone 만 남는다
#          → 이 상태를 delete.retention.ms 동안 유지해
#            뒤늦게 붙는 컨슈머도 "삭제됐다" 를 읽을 수 있게 한다
#       ② delete.retention.ms 경과 후: tombstone 자신도 삭제
#
#     ★ 그래서 delete.retention.ms 보다 오래 멈춰 있던 컨슈머는
#       삭제 사실을 영영 못 봅니다. 이미 캐시를 갖고 있다가 30시간 만에 재기동하면
#       마지막 커밋 오프셋부터 이어 읽는데, 그 사이 tombstone 이 사라졌으므로
#       캐시에 삭제된 키가 그대로 남습니다.
#       "이미 취소한 주문이 목록에 계속 보인다" 의 정체입니다. 에러는 없습니다.
#
#       대응 세 가지:
#         ① 압축 토픽 컨슈머의 "마지막 커밋 시각" 에 알람 (랙이 아니라 시각)
#         ② 오래 멈췄던 컨슈머는 오프셋을 이어받지 말고 처음부터 재구축
#            (압축 토픽은 키 수 = 레코드 수라 이게 쌉니다)
#         ③ delete.retention.ms 를 컨슈머 최대 중단 시간보다 넉넉하게


# ══════════════════════════════════════════════════════════════════════
hr "정답 6 — 키가 null 이면 클리너가 압축을 거부한다"
# ══════════════════════════════════════════════════════════════════════
K kafka-topics.sh --bootstrap-server "$BS" --create --if-not-exists \
  --topic s09ex_nullkey --partitions 1 --replication-factor 3 \
  --config cleanup.policy=compact \
  --config min.cleanable.dirty.ratio=0.01 \
  --config segment.ms=10000

docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server "$BS" --topic s09ex_nullkey \
  --property parse.key=true --property key.separator=: <<'EOF'
A:{"k":"A","v":1}
B:{"k":"B","v":1}
A:{"k":"A","v":2}
B:{"k":"B","v":2}
A:{"k":"A","v":3}
EOF

echo "-- 이번에는 키 없이 3건 (parse.key 를 쓰지 않습니다)"
docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server "$BS" --topic s09ex_nullkey <<'EOF'
{"no":"key1"}
{"no":"key2"}
{"no":"key3"}
EOF

docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server "$BS" --topic s09ex_nullkey \
  --property parse.key=true --property key.separator=: <<'EOF'
TRIGGER:{"k":"TRIGGER","v":0}
EOF
sleep 40

echo
docker logs kafka-1 --since 2m 2>&1 | grep -i 'skipping compaction' || \
  echo '(경고를 못 찾았습니다 — 대기 시간을 늘려 보세요)'

echo
echo "-- 레코드가 하나도 안 줄었는지 확인"
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic s09ex_nullkey \
  --from-beginning --timeout-ms 5000 2>&1 | tail -1 || true

# 해설:
#   로그에 이런 경고가 남습니다:
#     WARN Cleaner 0: Skipping compaction of s09ex_nullkey-0
#          due to record with null key at offset 5 (kafka.log.LogCleaner)
#
#   ★ 에러가 아니라 경고입니다. 그래서 더 위험합니다.
#     - 프로듀서는 정상입니다. 계속 성공합니다.
#     - 컨슈머도 정상입니다. 모든 레코드를 다 받습니다.
#     - 브로커는 살아 있고, under-replicated 도 0 이고, 랙도 정상입니다.
#     - 다만 그 파티션의 압축만 조용히 멈춥니다.
#       → 토픽이 계속 커집니다. 한 달 뒤 디스크가 찹니다.
#       → 그때 원인을 찾으려면 한 달 전 로그의 WARN 한 줄을 찾아야 합니다.
#
#   ★ 압축 토픽에 키 없는 레코드를 넣을 수 있다는 것 자체가 함정입니다.
#     브로커는 그것을 거부하지 않습니다. 압축이 불가능해질 뿐입니다.
#
#   이 지뢰를 가장 자주 밟는 경로는 "delete 토픽을 나중에 compact 로 바꾸는 것" 입니다.
#     kconf --alter --add-config cleanup.policy=compact
#   이 명령은 즉시 적용되고 재시작도 필요 없습니다.
#   그 토픽에 키 없는 레코드가 하나라도 섞여 있으면 압축이 그 지점에서 멈춥니다.
#   그리고 키가 있는 레코드들은 옛 값이 사라져 이력이 손상됩니다. 되돌릴 수 없습니다.
#
#   ★ 원칙: cleanup.policy 변경은 새 토픽을 만들어 마이그레이션하는 것입니다.
#     운영 중인 토픽의 정책을 바꾸지 마세요.


# ══════════════════════════════════════════════════════════════════════
hr "정답 7 — 압축 적합성 판정"
# ══════════════════════════════════════════════════════════════════════
cat <<'JUDGE'
  설계 기준:  ① 키 필수  ② 카디널리티 유한  ③ 값이 전체 상태(델타 금지)  ④ 순서 의존 금지

  (A) user-profile-snapshot        ✅ 적합
      ① 키 = user_id                                    OK
      ② 사용자 수는 유한                                 OK
      ③ 값 = 프로필 전체 JSON (델타 아님)                 OK
      ④ "현재 프로필" 만 필요, 변경 이력 불필요            OK
      → 교과서적인 압축 토픽입니다. 탈퇴 시 tombstone 으로 키를 지웁니다.

  (B) page-view-log                ❌ 부적합 (① ④ 위반)
      ① 키가 없다 → 클리너가 압축을 아예 거부 (정답 6 참조)
      ④ 페이지뷰는 "몇 번 봤는가" 가 데이터의 본질이라
         같은 키의 옛 레코드를 지우면 집계가 불가능
      → cleanup.policy=delete + 적절한 retention.ms 가 맞습니다.

  (C) account-balance-delta        ❌ 부적합 (③ 위반 — 가장 위험한 케이스)
      ③ 값이 델타(+10000)입니다.
         압축 후 마지막 delta 하나만 남으면 컨슈머가 계산한 잔액이 틀립니다.
           +10000, +5000, -3000  →  압축 후 -3000 만 남음
           컨슈머 계산: -3000원.  실제: 12000원.  에러 없음, 숫자만 틀림.
      → 값을 "잔액 전체" 로 바꾸면(account-balance-snapshot) 적합해집니다.
         또는 이벤트 스트림으로 두고 cleanup.policy=delete 를 씁니다.

  (D) session-heartbeat            ⚠️ 조건부 적합 (② 위반)
      ① 키 = session_id                                  OK
      ③ 값 = last_seen 전체 상태                          OK
      ④ 마지막 상태만 필요                                OK
      ② 세션은 계속 새로 생기므로 키 카디널리티가 무한히 증가
      → cleanup.policy=compact 만 걸면 토픽이 무한 성장합니다.
         compact,delete + retention.ms 로 상한을 줘야 합니다.
         (Kafka Streams 의 윈도우 상태 저장소가 정확히 이 조합을 씁니다 — Step 13)
JUDGE

# 해설:
#   ★ 이 스텝의 결론은 한 문장으로 정리됩니다.
#     압축은 "이벤트 스트림" 이 아니라 "상태 스냅샷" 을 위한 도구입니다.
#
#   (B) 와 (C) 는 이벤트 스트림입니다. 각 레코드가 "무슨 일이 있었는가" 를 말하고
#   그 이력 자체가 데이터입니다. 압축은 이력을 지우므로 근본적으로 상충합니다.
#
#   (A) 와 (D) 는 상태 스냅샷입니다. 각 레코드가 "지금 어떤 상태인가" 를 말하고
#   옛 레코드는 버려도 됩니다. 압축이 정확히 이것을 위해 만들어졌습니다.
#
#   ★ 판정이 가장 어려운 것은 (C) 입니다. 키도 있고 카디널리티도 유한해서
#     ①②④ 만 보면 통과합니다. ③ 하나가 걸리는데, 그 결과가
#     "에러 없이 숫자만 틀리는" 형태라 발견이 가장 늦습니다.
#     압축 토픽 설계 리뷰에서 ③ 을 가장 먼저 물어야 하는 이유입니다.
#
#   ★ 이 코스의 order-events 토픽은 네 조건을 모두 만족합니다.
#     키 = order_id, 주문 수 유한, 값 = 주문의 현재 상태 전체, 최종 상태만 필요.
#     Kafka 자신의 __consumer_offsets 도 마찬가지입니다.
#     키 = (그룹, 토픽, 파티션), 유한, 값 = 커밋 오프셋 절대값, 마지막 커밋만 필요.


# ══════════════════════════════════════════════════════════════════════
hr "뒷정리 — s09ex_ 토픽 삭제"
# ══════════════════════════════════════════════════════════════════════
for t in s09ex_ret s09ex_bytes s09ex_compact s09ex_nullkey; 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" --list | grep '^s09ex_' \
  && { echo 'FAIL: s09ex_ 토픽이 남아 있습니다'; exit 1; } || echo 'OK: s09ex_ 토픽 없음'

docker compose ps --format 'table {{.Name}}\t{{.Status}}'
echo "완료. → Step 10"