#Step 06 — 오프셋 관리

docker exec kafka-1 /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka-1:9092 \
  --create --topic s06_orders --partitions 1 --replication-factor 3

docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka-1:9092 --topic s06_orders \
  --property parse.key=true --property key.separator=: <<'EOF'
$(seq 1 100)
EOF

for i in $(seq 1001 1100); do
  c=$(printf "C%03d" $(( (i % 10) + 1 )))
  echo "${c}:{\"order_id\":\"O-${i}\",\"customer_id\":\"${c}\",\"amount\":39000,\"status\":\"CREATED\"}"
done | docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka-1:9092 --topic s06_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 s06_orders

docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 --topic s06_orders \
  --group s06-basic --from-beginning --max-messages 60

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

docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 --topic s06_orders \
  --group s06-latest --timeout-ms 5000

docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 --topic s06_orders \
  --group s06-earliest --consumer-property auto.offset.reset=earliest \
  --timeout-ms 5000 | tail -1

docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 --topic s06_orders \
  --group s06-none --consumer-property auto.offset.reset=none \
  --timeout-ms 5000

docker exec kafka-1 /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka-1:9092 \
  --describe --topic __consumer_offsets | head -4

docker exec kafka-1 sh -c 'cat > /tmp/H.java <<EOF
public class H { public static void main(String[] a) {
  System.out.println(Math.abs(a[0].hashCode()) % 50); } }
EOF
cd /tmp && java H.java s06-basic'

docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 \
  --topic __consumer_offsets \
  --formatter "kafka.coordinator.group.GroupMetadataMessageFormatter" \
  --from-beginning --timeout-ms 10000 | grep s06

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

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

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

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

kcg --describe --group s06-dup

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

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

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

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

docker exec kafka-1 /opt/kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server kafka-1:9092 \
  --group s06-basic --topic s06_orders \
  --reset-offsets --to-earliest --dry-run

docker exec kafka-1 /opt/kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server kafka-1:9092 \
  --group s06-basic --topic s06_orders \
  --reset-offsets --to-earliest --execute

kcg --describe --group s06-basic

# [터미널 B]
docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 --topic s06_orders --group s06-basic

# [터미널 A]
kcg --group s06-basic --topic s06_orders --reset-offsets --to-earliest --execute

kcg --group s06-basic --topic s06_orders --reset-offsets --by-duration PT1H --dry-run

kcg --group s06-basic --topic s06_orders \
  --reset-offsets --to-datetime 2024-03-11T10:22:00.000 --dry-run

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

for g in s06-basic s06-latest s06-earliest s06-loss s06-dup s06-manual s06-async s06-per-record s06-seek; do
  docker exec kafka-1 /opt/kafka/bin/kafka-consumer-groups.sh \
    --bootstrap-server kafka-1:9092 --delete --group "$g" 2>/dev/null || true
done

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

kcg --list | grep s06 || echo "남은 s06 그룹 없음"
kt --list | grep s06 || echo "남은 s06 토픽 없음"

#!/usr/bin/env bash
#
# Step 06 — 오프셋 관리 : 본문 예제 모음
#
# 실행법:
#   bash practice.sh              # 통째로 실행
#   bash -x practice.sh           # 한 줄씩 확인하며 실행 (권장)
#
# 사전 조건:
#   docker compose up -d  로 kafka-1/2/3 이 모두 (healthy) 여야 합니다.
#
# 주의:
#   - 이 스크립트는 s06_orders 토픽과 s06-* 컨슈머 그룹만 건드립니다.
#   - 6-4, 6-5 구간은 Practice.java 가 같은 디렉터리에 있어야 합니다.
#   - [터미널 B] 주석이 붙은 구간은 별도 창에서 실행해야 관찰됩니다.

set -euo pipefail

BS=kafka-1:9092
K()  { docker exec    kafka-1 /opt/kafka/bin/"$@"; }   # 일반 실행
KI() { docker exec -i kafka-1 /opt/kafka/bin/"$@"; }   # stdin 을 넘겨야 할 때 (프로듀서)

TOPIC=s06_orders

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

# ---------------------------------------------------------------------------
# [6-0] 실습 준비 — 토픽 생성과 메시지 100건
# ---------------------------------------------------------------------------
hr "[6-0] 토픽 생성"

K kafka-topics.sh --bootstrap-server "$BS" --delete --topic "$TOPIC" 2>/dev/null || true
sleep 2

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

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

hr "[6-0] 메시지 100건 발행 (O-1001 ~ O-1100, 키는 C001~C010 순환)"

# 이 100 이라는 숫자가 이후 모든 절의 기준값입니다. 바꾸지 마세요.
for i in $(seq 1001 1100); do
  c=$(printf "C%03d" $(( (i % 10) + 1 )))
  echo "${c}:{\"order_id\":\"O-${i}\",\"customer_id\":\"${c}\",\"amount\":39000,\"status\":\"CREATED\"}"
done | KI kafka-console-producer.sh --bootstrap-server "$BS" --topic "$TOPIC" \
        --property parse.key=true --property key.separator=:

# 총 건수 확인 — 결과는 s06_orders:0:100
K kafka-get-offsets.sh --bootstrap-server "$BS" --topic "$TOPIC"

# ---------------------------------------------------------------------------
# [6-1] 오프셋 4종 — 60건만 읽고 커밋한 뒤 --describe
# ---------------------------------------------------------------------------
hr "[6-1] s06-basic 그룹으로 60건만 읽습니다"

K kafka-console-consumer.sh --bootstrap-server "$BS" --topic "$TOPIC" \
  --group s06-basic --from-beginning --max-messages 60 | tail -1

# CURRENT-OFFSET 60 / LOG-END-OFFSET 100 / LAG 40 이 나와야 합니다.
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s06-basic

# ---------------------------------------------------------------------------
# [6-2] auto.offset.reset — latest / earliest / none
#       ※ 반드시 서로 다른 새 그룹으로 붙어야 차이가 보입니다.
#          기존 그룹은 커밋된 오프셋이 있어 이 설정이 무시됩니다.
# ---------------------------------------------------------------------------
hr "[6-2-1] latest (기본값) — 0건을 읽습니다"

K kafka-console-consumer.sh --bootstrap-server "$BS" --topic "$TOPIC" \
  --group s06-latest --timeout-ms 5000 || true

# LAG 0 인데 처리 건수는 0 입니다. 이것이 함정 A 입니다.
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s06-latest

hr "[6-2-2] earliest — 100건을 읽습니다"

K kafka-console-consumer.sh --bootstrap-server "$BS" --topic "$TOPIC" \
  --group s06-earliest --consumer-property auto.offset.reset=earliest \
  --timeout-ms 5000 2>&1 | tail -2 || true

K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s06-earliest

hr "[6-2-3] none — NoOffsetForPartitionException 으로 죽습니다"

# 의도적으로 실패하는 명령이므로 || true 로 감쌉니다. set -e 가 걸려 있습니다.
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic "$TOPIC" \
  --group s06-none --consumer-property auto.offset.reset=none \
  --timeout-ms 5000 2>&1 | grep -E 'NoOffsetForPartition|Processed' || true

# ---------------------------------------------------------------------------
# [6-3] __consumer_offsets 직접 조회
# ---------------------------------------------------------------------------
hr "[6-3] __consumer_offsets 토픽 구조"

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

hr "[6-3] 그룹이 어느 파티션에 저장되는지 계산 — abs(groupId.hashCode()) % 50"

docker exec kafka-1 sh -c 'cat > /tmp/H.java <<EOF
public class H { public static void main(String[] a) {
  for (String s : a) System.out.println(s + " -> partition " + (Math.abs(s.hashCode()) % 50)); } }
EOF
cd /tmp && java H.java s06-basic s06-latest s06-earliest'

hr "[6-3] 실제 커밋 레코드 조회"

# GroupMetadataMessageFormatter 가 오프셋 커밋과 그룹 메타데이터를 모두 디코딩합니다.
K kafka-console-consumer.sh --bootstrap-server "$BS" \
  --topic __consumer_offsets \
  --formatter "kafka.coordinator.group.GroupMetadataMessageFormatter" \
  --from-beginning --timeout-ms 10000 2>/dev/null | grep s06 || true

hr "[6-3] offsets.retention.minutes 기본값 확인 (10080 = 7일)"

K kafka-configs.sh --bootstrap-server "$BS" \
  --describe --entity-type brokers --entity-name 1 --all 2>/dev/null \
  | grep offsets.retention || true

# ---------------------------------------------------------------------------
# [6-4] 핵심 함정 B — enable.auto.commit=true 가 만드는 유실
# ---------------------------------------------------------------------------
hr "[6-4] Practice.java 를 컨테이너로 복사"

docker cp Practice.java kafka-1:/tmp/

hr "[6-4] autocommit-loss — 60건 처리 후 halt(). 커밋은 100까지 됩니다"

docker exec kafka-1 sh -c 'cd /tmp && java -cp "/opt/kafka/libs/*" Practice.java autocommit-loss' || true

# LAG 0 인데 실제 처리는 60건뿐입니다.
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s06-loss

hr "[6-4] autocommit-loss-resume — 남은 것을 읽어 유실 건수를 집계"

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

# ---------------------------------------------------------------------------
# [6-5] 핵심 함정 C — 같은 설정이 만드는 중복
# ---------------------------------------------------------------------------
hr "[6-5] autocommit-dup — 100건 처리, 커밋은 40 에서 멈춘 채 halt()"

docker exec kafka-1 sh -c 'cd /tmp && java -cp "/opt/kafka/libs/*" Practice.java autocommit-dup' || true

# CURRENT-OFFSET 40 / LAG 60 이 나옵니다.
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s06-dup

hr "[6-5] autocommit-dup-resume — 60건이 재처리됩니다"

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

# ---------------------------------------------------------------------------
# [6-6] 수동 커밋 — commitSync / commitAsync / 특정 오프셋
# ---------------------------------------------------------------------------
hr "[6-6-1] manual-sync — 유실 0, 중복 0"

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

hr "[6-6-2] manual-async — 같은 결과, 더 빠름"

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

hr "[6-6-3] manual-per-record — 10건마다 offset+1 을 명시 커밋, metadata 도 함께"

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

# metadata=processed-by-worker-3 이 보여야 합니다.
K kafka-console-consumer.sh --bootstrap-server "$BS" \
  --topic __consumer_offsets \
  --formatter "kafka.coordinator.group.GroupMetadataMessageFormatter" \
  --from-beginning --timeout-ms 10000 2>/dev/null | grep s06-per-record || true

# ---------------------------------------------------------------------------
# [6-7] 오프셋 리셋
# ---------------------------------------------------------------------------
hr "[6-7] --dry-run (기본) — 아직 아무것도 바뀌지 않습니다"

K kafka-consumer-groups.sh --bootstrap-server "$BS" \
  --group s06-basic --topic "$TOPIC" --reset-offsets --to-earliest --dry-run

hr "[6-7] --execute — 실제로 적용"

K kafka-consumer-groups.sh --bootstrap-server "$BS" \
  --group s06-basic --topic "$TOPIC" --reset-offsets --to-earliest --execute

# CURRENT-OFFSET 0 / LAG 100 으로 되감겼는지 확인합니다.
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s06-basic

hr "[6-7] 다양한 리셋 방식 (전부 dry-run 으로만)"

K kafka-consumer-groups.sh --bootstrap-server "$BS" \
  --group s06-basic --topic "$TOPIC" --reset-offsets --to-latest --dry-run

K kafka-consumer-groups.sh --bootstrap-server "$BS" \
  --group s06-basic --topic "$TOPIC" --reset-offsets --to-offset 30 --dry-run

K kafka-consumer-groups.sh --bootstrap-server "$BS" \
  --group s06-basic --topic "$TOPIC" --reset-offsets --shift-by -20 --dry-run

K kafka-consumer-groups.sh --bootstrap-server "$BS" \
  --group s06-basic --topic "$TOPIC" --reset-offsets --by-duration PT1H --dry-run

hr "[6-7] 그룹 상태 확인 — Empty 여야 리셋할 수 있습니다"

K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s06-basic --state

# ---------------------------------------------------------------------------
# [6-7] 활성 그룹 리셋 에러 재현 — 터미널 2개가 필요합니다
# ---------------------------------------------------------------------------
# [터미널 B] 아래를 별도 창에서 실행해 그룹을 Stable 로 만들어 두십시오.
#
#   docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
#     --bootstrap-server kafka-1:9092 --topic s06_orders --group s06-basic
#
# [터미널 A] 그 상태에서 아래를 실행하면 다음 에러가 납니다.
#
#   Error: Assignments can only be reset if the group 's06-basic' is inactive,
#          but the current state is Stable.
#
# K kafka-consumer-groups.sh --bootstrap-server "$BS" \
#   --group s06-basic --topic "$TOPIC" --reset-offsets --to-earliest --execute

# ---------------------------------------------------------------------------
# [6-8] seek / seekToBeginning / seekToEnd
# ---------------------------------------------------------------------------
hr "[6-8] seek — 애플리케이션 안에서 위치를 바꿉니다 (커밋하지 않음)"

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

# seek 은 커밋하지 않으므로 --describe 는 변하지 않습니다.
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s06-seek || true

# ---------------------------------------------------------------------------
# [6-9] 정리
# ---------------------------------------------------------------------------
hr "[6-9] 정리 — 그룹과 토픽 삭제"

for g in s06-basic s06-latest s06-earliest s06-none s06-loss s06-dup \
         s06-manual s06-async s06-per-record s06-seek; do
  K kafka-consumer-groups.sh --bootstrap-server "$BS" --delete --group "$g" 2>/dev/null || true
done

K kafka-topics.sh --bootstrap-server "$BS" --delete --topic "$TOPIC" 2>/dev/null || true

docker exec kafka-1 rm -f /tmp/Practice.java /tmp/H.java /tmp/s06-count.txt 2>/dev/null || true

hr "[6-9] 확인"

K kafka-consumer-groups.sh --bootstrap-server "$BS" --list | grep s06 || echo "남은 s06 그룹 없음"
K kafka-topics.sh --bootstrap-server "$BS" --list | grep s06 || echo "남은 s06 토픽 없음"

echo
echo "Step 06 실습 완료."

#!/usr/bin/env bash
#
# Step 06 — 오프셋 관리 : 연습문제 (7문제)
#
# 실행법:
#   bash exercise.sh              # setup 만 돌고 문제는 직접 채워야 합니다
#
# 규칙:
#   - "# 여기에 작성:" 아래를 직접 채우세요.
#   - 본문(index.md)의 숫자를 그대로 옮겨 적으면 틀립니다. 여기 데이터는 50건입니다.
#   - 정답은 solution.sh 에 있습니다. 먼저 풀어 보세요.

set -euo pipefail

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

TOPIC=s06x_orders

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

setup() {
  hr "setup — s06x_orders 토픽(파티션 1개)과 메시지 50건"

  K kafka-topics.sh --bootstrap-server "$BS" --delete --topic "$TOPIC" 2>/dev/null || true
  sleep 2
  K kafka-topics.sh --bootstrap-server "$BS" \
    --create --topic "$TOPIC" --partitions 1 --replication-factor 3

  for i in $(seq 2001 2050); do
    c=$(printf "C%03d" $(( (i % 10) + 1 )))
    echo "${c}:{\"order_id\":\"O-${i}\",\"customer_id\":\"${c}\",\"amount\":21000,\"status\":\"CREATED\"}"
  done | KI kafka-console-producer.sh --bootstrap-server "$BS" --topic "$TOPIC" \
          --property parse.key=true --property key.separator=:

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

  # 문제 2 에서 쓸 해시 계산기
  docker exec kafka-1 sh -c 'cat > /tmp/H.java <<EOF
public class H { public static void main(String[] a) {
  for (String s : a) System.out.println(s + " -> partition " + (Math.abs(s.hashCode()) % 50)); } }
EOF'

  docker cp Practice.java kafka-1:/tmp/ 2>/dev/null || \
    echo "경고: Practice.java 를 찾지 못했습니다. 문제 3·4·5 를 풀 수 없습니다."
}

cleanup() {
  hr "cleanup — s06x 그룹과 토픽 삭제"
  for g in s06x-latest s06x-earliest s06x-a s06x-loss s06x-fix s06x-bug s06x-reset s06x-seek; do
    K kafka-consumer-groups.sh --bootstrap-server "$BS" --delete --group "$g" 2>/dev/null || true
  done
  K kafka-topics.sh --bootstrap-server "$BS" --delete --topic "$TOPIC" 2>/dev/null || true
  docker exec kafka-1 rm -f /tmp/H.java /tmp/PracticeBug.java /tmp/s06-count.txt 2>/dev/null || true
}

setup

# ===========================================================================
# 문제 1. auto.offset.reset 의 latest 와 earliest 비교
#
#   새 그룹 s06x-latest 를 auto.offset.reset=latest(기본) 로,
#   새 그룹 s06x-earliest 를 earliest 로 붙여 각각 몇 건을 읽는지 확인하고,
#   두 그룹의 --describe LAG 을 비교하십시오.
#
#   질문: 두 그룹의 LAG 이 같습니까? 처리한 건수는 같습니까?
#         같지 않다면 대시보드에서 LAG 만 보는 운영자는 무엇을 놓칩니까?
# ===========================================================================
hr "문제 1"

# 여기에 작성: s06x-latest 로 5초 동안 읽기


# 여기에 작성: s06x-latest 의 --describe


# 여기에 작성: s06x-earliest 로 earliest 설정해서 읽기


# 여기에 작성: s06x-earliest 의 --describe


# ===========================================================================
# 문제 2. __consumer_offsets 에서 특정 그룹의 커밋만 뽑기
#
#   (a) 그룹 s06x-a 를 만들어 20건만 읽고 커밋하십시오.
#   (b) s06x-a 의 커밋이 __consumer_offsets 의 몇 번 파티션에 저장되는지
#       /tmp/H.java 로 계산하십시오.
#   (c) 그 파티션만 읽어서 s06x-a 의 OffsetAndMetadata 레코드를 출력하십시오.
#       (힌트: --partition 옵션. 50개 파티션을 전부 훑을 필요가 없습니다)
# ===========================================================================
hr "문제 2"

# 여기에 작성 (a): s06x-a 로 20건 읽기


# 여기에 작성 (b): 파티션 번호 계산


# 여기에 작성 (c): 해당 파티션만 조회


# ===========================================================================
# 문제 3. 자동 커밋 유실을 숫자로 증명하기
#
#   Practice.java 의 autocommit-loss 시나리오를 그룹 이름만 s06x-loss 로,
#   토픽을 s06x_orders 로 바꿔 돌리고(환경변수 S06_TOPIC / S06_GROUP 지원),
#   아래 두 숫자를 각각 어디서 얻는지 명령으로 보이십시오.
#
#     A. 애플리케이션이 실제로 처리한 건수
#     B. 브로커에 커밋된 오프셋
#
#   질문: A 와 B 중 --describe 로 알 수 있는 것은 무엇입니까?
#         나머지 하나는 어디서 얻어야 합니까?
# ===========================================================================
hr "문제 3"

# 여기에 작성: Practice.java autocommit-loss 실행 (S06_TOPIC / S06_GROUP 지정)


# 여기에 작성: 커밋된 오프셋 확인


# ===========================================================================
# 문제 4. 자동 커밋을 유지하면서 유실 창 없애기
#
#   enable.auto.commit=true 를 유지한 채,
#   max.poll.records 와 auto.commit.interval.ms 를 조정해
#   "배치 하나의 처리 시간 < auto.commit.interval.ms" 가 되도록 만드십시오.
#   건당 처리 시간은 50ms 로 고정입니다 (Practice.java 의 sleep).
#
#   (a) 어떤 조합을 골랐습니까? 계산 근거를 주석으로 쓰십시오.
#   (b) 그 조합으로 돌려 유실이 0 인지 확인하십시오.
#   (c) 이 해법의 한계는 무엇입니까? (해설에서 다시 다룹니다)
# ===========================================================================
hr "문제 4"

# 여기에 작성 (a): 계산 근거
#   배치당 처리 시간 = max.poll.records × 50ms =
#   auto.commit.interval.ms =
#   따라서 고른 값:

# 여기에 작성 (b): 실행


# ===========================================================================
# 문제 5. 커밋 오프셋에서 "+1" 을 빼먹으면?
#
#   Practice.java 를 /tmp/PracticeBug.java 로 복사한 뒤,
#   manual-per-record 시나리오의 커밋 부분을
#       new OffsetAndMetadata(r.offset() + 1, ...)
#   에서
#       new OffsetAndMetadata(r.offset(), ...)
#   로 고쳐서 그룹 s06x-bug 로 돌리십시오. (클래스명도 PracticeBug 로 바꿔야 합니다)
#
#   질문: --describe 의 LAG 이 얼마입니까? 왜 그 값입니까?
#         파티션이 3개인 토픽이었다면 LAG 이 얼마로 나오겠습니까?
# ===========================================================================
hr "문제 5"

# 여기에 작성: PracticeBug.java 만들기 (sed 로 두 곳을 치환하면 됩니다)


# 여기에 작성: 실행


# 여기에 작성: --describe 로 LAG 확인


# ===========================================================================
# 문제 6. 활성 그룹을 리셋하려고 하면?      [터미널 2개 필요]
#
#   ⚠️ 이 문제는 반드시 창을 두 개 열어야 합니다.
#      한 창에서 순차 실행하면 이미 컨슈머가 종료된 뒤라 에러가 재현되지 않습니다.
#
#   [터미널 B] 컨슈머를 띄워 그룹 s06x-reset 을 Stable 로 만드십시오.
#   [터미널 A] 그 상태에서 --reset-offsets --to-earliest --execute 를 시도해
#             에러 메시지를 확인하십시오.
#
#   그다음, 리셋이 성공하도록 절차를 고치십시오.
#   힌트: 컨슈머를 내린 직후에도 실패할 수 있습니다. --state 를 확인하세요.
# ===========================================================================
hr "문제 6"

# [터미널 B] 여기에 작성: 컨슈머 띄우기


# [터미널 A] 여기에 작성: 리셋 시도 (에러 확인)


# 여기에 작성: 성공하는 절차 (상태가 Empty 가 될 때까지 대기하는 루프 포함)


# ===========================================================================
# 문제 7. --reset-offsets --shift-by 와 seek() 의 차이
#
#   (a) 그룹 s06x-seek 로 50건을 전부 읽어 커밋하십시오. (CURRENT-OFFSET 50)
#   (b) --reset-offsets --shift-by -30 --execute 를 하고 --describe 를 보십시오.
#   (c) 다시 50 으로 되돌린 뒤, Practice.java 의 seek 시나리오로
#       같은 30 만큼 되감고 --describe 를 보십시오.
#
#   질문: (b) 와 (c) 에서 --describe 의 CURRENT-OFFSET 이 어떻게 다릅니까?
#         (c) 의 결과를 (b) 와 같게 만들려면 무엇을 추가해야 합니까?
# ===========================================================================
hr "문제 7"

# 여기에 작성 (a)


# 여기에 작성 (b)


# 여기에 작성 (c)


# ===========================================================================
cleanup

echo
echo "연습문제 종료. solution.sh 로 채점하세요."

#!/usr/bin/env bash
#
# Step 06 — 오프셋 관리 : 연습문제 정답과 해설
#
# 실행법:
#   bash solution.sh
#
# 이 파일은 exercise.sh 와 같은 setup(50건) 을 쓰며,
# 각 정답 뒤에 "왜 그 답인가" 를 설명하는 # 해설: 블록이 붙어 있습니다.
# 문제를 풀어 본 뒤에 여세요.

set -euo pipefail

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

TOPIC=s06x_orders

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

setup() {
  hr "setup — s06x_orders(1 파티션) + 메시지 50건"
  K kafka-topics.sh --bootstrap-server "$BS" --delete --topic "$TOPIC" 2>/dev/null || true
  sleep 2
  K kafka-topics.sh --bootstrap-server "$BS" \
    --create --topic "$TOPIC" --partitions 1 --replication-factor 3
  for i in $(seq 2001 2050); do
    c=$(printf "C%03d" $(( (i % 10) + 1 )))
    echo "${c}:{\"order_id\":\"O-${i}\",\"customer_id\":\"${c}\",\"amount\":21000,\"status\":\"CREATED\"}"
  done | KI kafka-console-producer.sh --bootstrap-server "$BS" --topic "$TOPIC" \
          --property parse.key=true --property key.separator=:
  K kafka-get-offsets.sh --bootstrap-server "$BS" --topic "$TOPIC"

  docker exec kafka-1 sh -c 'cat > /tmp/H.java <<EOF
public class H { public static void main(String[] a) {
  for (String s : a) System.out.println(s + " -> partition " + (Math.abs(s.hashCode()) % 50)); } }
EOF'
  docker cp Practice.java kafka-1:/tmp/ 2>/dev/null || true
}

cleanup() {
  hr "cleanup"
  for g in s06x-latest s06x-earliest s06x-a s06x-loss s06x-fix s06x-bug s06x-reset s06x-seek; do
    K kafka-consumer-groups.sh --bootstrap-server "$BS" --delete --group "$g" 2>/dev/null || true
  done
  K kafka-topics.sh --bootstrap-server "$BS" --delete --topic "$TOPIC" 2>/dev/null || true
  docker exec kafka-1 rm -f /tmp/H.java /tmp/PracticeBug.java /tmp/s06-count.txt 2>/dev/null || true
}

setup

# ===========================================================================
# 정답 1
# ===========================================================================
hr "정답 1 — latest 는 0건, earliest 는 50건. 그런데 LAG 은 둘 다 0"

K kafka-console-consumer.sh --bootstrap-server "$BS" --topic "$TOPIC" \
  --group s06x-latest --timeout-ms 5000 2>&1 | grep Processed || true
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s06x-latest

K kafka-console-consumer.sh --bootstrap-server "$BS" --topic "$TOPIC" \
  --group s06x-earliest --consumer-property auto.offset.reset=earliest \
  --timeout-ms 5000 2>&1 | grep Processed || true
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s06x-earliest

# 해설:
#   s06x-latest    → Processed a total of 0 messages   / CURRENT-OFFSET 50, LAG 0
#   s06x-earliest  → Processed a total of 50 messages  / CURRENT-OFFSET 50, LAG 0
#
#   두 그룹의 --describe 출력은 완전히 동일합니다. CURRENT-OFFSET 50, LAG 0.
#   그런데 실제로 처리한 건수는 0 과 50 으로 정반대입니다.
#
#   왜 이렇게 되느냐면, latest 그룹은 "파티션의 끝(50)에서 시작" 했고
#   그 상태 그대로 커밋했기 때문입니다. 읽은 게 없는데 오프셋은 50 입니다.
#
#   즉 LAG 은 "처리했는가" 를 말해 주지 않습니다.
#   LAG 은 오직 "커밋된 오프셋이 로그 끝에 얼마나 가까운가" 만 말합니다.
#   대시보드에서 LAG 만 감시하는 운영자는 이 사고를 절대 발견하지 못합니다.
#
#   발견하려면 애플리케이션 쪽 지표(처리 건수 카운터)가 반드시 있어야 하고,
#   그 값과 브로커의 오프셋 증가분을 대조해야 합니다. 이것이 문제 3 의 주제입니다.

# ===========================================================================
# 정답 2
# ===========================================================================
hr "정답 2 — 그룹 → 파티션 계산 후 그 파티션만 조회"

# (a) 20건 읽고 커밋
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic "$TOPIC" \
  --group s06x-a --from-beginning --max-messages 20 >/dev/null

# (b) 파티션 번호 계산
docker exec kafka-1 sh -c 'cd /tmp && java H.java s06x-a'

# (c) 계산된 파티션만 읽기
P=$(docker exec kafka-1 sh -c 'cd /tmp && java H.java s06x-a' | grep -oE '[0-9]+$')
echo "s06x-a 는 __consumer_offsets 파티션 ${P} 에 저장됩니다."

K kafka-console-consumer.sh --bootstrap-server "$BS" \
  --topic __consumer_offsets --partition "$P" \
  --formatter "kafka.coordinator.group.GroupMetadataMessageFormatter" \
  --from-beginning --timeout-ms 8000 2>/dev/null | grep s06x-a || true

# 해설:
#   출력 예:
#     [s06x-a,s06x_orders,0]::OffsetAndMetadata(offset=20, leaderEpoch=Optional[0],
#       metadata=, commitTimestamp=1710120233117, expireTimestamp=None)
#
#   키가 [그룹, 토픽, 파티션] 이고 값이 OffsetAndMetadata 입니다.
#   offset=20 은 --describe 의 CURRENT-OFFSET 20 과 같은 숫자입니다.
#   CLI 가 특별한 API 를 쓰는 게 아니라, 이 토픽을 읽어서 보여줄 뿐입니다.
#
#   --partition N 을 지정하는 것이 이 문제의 실전 포인트입니다.
#   __consumer_offsets 는 파티션이 50개이고 운영 클러스터에서는 수백만 건이 쌓여 있습니다.
#   --from-beginning 으로 전부 훑으면 수 분이 걸리고 브로커에 부하도 갑니다.
#   abs(groupId.hashCode()) % 50 으로 목표 파티션을 특정하면 한 파티션만 읽으면 됩니다.
#
#   보너스: 그 파티션의 리더 브로커가 곧 이 그룹의 코디네이터입니다.
#     kafka-consumer-groups.sh --describe --group s06x-a --state
#   의 COORDINATOR (ID) 열과 일치하는지 확인해 보십시오.

# ===========================================================================
# 정답 3
# ===========================================================================
hr "정답 3 — 처리 건수와 커밋 오프셋은 출처가 다릅니다"

docker exec -e S06_TOPIC="$TOPIC" -e S06_GROUP=s06x-loss kafka-1 \
  sh -c 'cd /tmp && java -cp "/opt/kafka/libs/*" Practice.java autocommit-loss' || true

K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s06x-loss

# 해설:
#   A. 애플리케이션이 실제로 처리한 건수
#      → Practice.java 의 표준 출력에서만 얻을 수 있습니다.
#        "처리 30건, offset=29" 같은 로그의 마지막 값, 또는 halt 직전의 집계 출력입니다.
#        브로커는 이 숫자를 모릅니다. 알 방법이 없습니다.
#
#   B. 브로커에 커밋된 오프셋
#      → --describe 의 CURRENT-OFFSET, 또는 __consumer_offsets 직접 조회입니다.
#
#   즉 --describe 로 알 수 있는 것은 B 뿐입니다.
#   A 는 애플리케이션이 스스로 노출하지 않으면 아무도 모릅니다.
#
#   이것이 이 스텝 전체의 결론입니다.
#   유실은 "브로커가 아는 숫자"와 "애플리케이션이 아는 숫자"의 차이로만 드러나며,
#   둘 중 하나만 보고 있으면 영원히 보이지 않습니다.
#
#   실무에서는 컨슈머가 처리 건수를 메트릭으로 내보내고(Micrometer 등),
#   그 증가분과 kafka.consumer:type=consumer-fetch-manager-metrics 의
#   records-consumed-total 을 함께 그래프로 겹쳐 봅니다.
#   두 선이 벌어지는 순간이 유실입니다.

# ===========================================================================
# 정답 4
# ===========================================================================
hr "정답 4 — max.poll.records 를 줄여 배치를 커밋 주기 안에 끝냅니다"

# (a) 계산 근거
#     건당 처리 시간 = 50ms (고정)
#     배치당 처리 시간 = max.poll.records × 50ms
#     이 값이 auto.commit.interval.ms 보다 작아야 합니다.
#
#     max.poll.records = 5      → 배치당 250ms
#     auto.commit.interval.ms = 5000 (기본)
#     250ms << 5000ms 이므로 배치를 처리하는 동안 커밋이 끼어들 여지가 없습니다.
#
#     반대로 본문의 유실 조합은 max.poll.records=10 × 50ms = 500ms 인데
#     auto.commit.interval.ms 를 1000 으로 줄여서 창을 강제로 열었던 것입니다.

# (b) 실행
docker exec -e S06_TOPIC="$TOPIC" -e S06_GROUP=s06x-fix \
            -e S06_MAX_POLL=5 -e S06_COMMIT_INTERVAL=5000 kafka-1 \
  sh -c 'cd /tmp && java -cp "/opt/kafka/libs/*" Practice.java autocommit-loss' || true

K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s06x-fix

# 해설 (c) — 이 해법의 한계:
#   이건 미봉책입니다. 세 가지 이유입니다.
#
#   1. 처리 시간이 고정이라는 전제가 틀립니다.
#      외부 API 호출이 평소 50ms 여도 상대 서버가 느려지면 5초가 됩니다.
#      그 순간 배치당 처리 시간이 25초가 되어 창이 활짝 열립니다.
#      "평소에는 괜찮다가 장애 때만 유실된다" 는 최악의 형태입니다.
#
#   2. max.poll.records 를 줄이면 처리량이 떨어집니다.
#      poll 왕복 횟수가 배치 크기에 반비례해서 늘어납니다.
#      500 → 5 로 줄이면 네트워크 왕복이 100배가 됩니다.
#
#   3. 계산이 맞는지 아무도 검증해 주지 않습니다.
#      코드 어디에도 "이 두 값의 관계가 중요하다" 는 흔적이 남지 않습니다.
#      다음 사람이 성능 튜닝한다고 max.poll.records 를 500 으로 올리는 순간
#      아무 경고 없이 유실이 시작됩니다.
#
#   근본 해법은 하나뿐입니다: enable.auto.commit=false + 처리 후 명시적 커밋.
#   이때는 "처리했으니 커밋한다" 는 인과관계가 코드에 직접 드러나므로
#   타이밍 계산이 필요 없고, 처리 시간이 얼마가 되든 안전합니다.

# ===========================================================================
# 정답 5
# ===========================================================================
hr "정답 5 — +1 을 빼먹으면 LAG 이 파티션 수만큼 고정됩니다"

docker exec kafka-1 sh -c \
  'cd /tmp && sed -e "s/class Practice/class PracticeBug/" \
                  -e "s/r.offset() + 1/r.offset()/" \
                  Practice.java > PracticeBug.java'

docker exec -e S06_TOPIC="$TOPIC" -e S06_GROUP=s06x-bug kafka-1 \
  sh -c 'cd /tmp && java -cp "/opt/kafka/libs/*" PracticeBug.java manual-per-record'

K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s06x-bug

# 해설:
#   결과:
#     GROUP      TOPIC        PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
#     s06x-bug   s06x_orders  0          49              50              1
#
#   LAG 이 정확히 1 입니다.
#
#   왜냐하면 커밋된 오프셋의 의미가 "마지막으로 처리한 레코드" 가 아니라
#   "다음에 읽어야 할 위치" 이기 때문입니다.
#   마지막 레코드의 오프셋이 49 이므로 커밋해야 할 값은 50 인데 49 를 커밋했습니다.
#   브로커 입장에서는 "이 그룹은 아직 49번 레코드를 안 읽었다" 로 보입니다.
#
#   그래서 재시작할 때마다 마지막 한 건을 항상 다시 처리합니다.
#   50건 중 1건이라 로그를 훑어봐서는 절대 눈치채지 못합니다.
#
#   파티션이 3개인 토픽이었다면 LAG 이 3 으로 나옵니다.
#   파티션마다 각각 1씩 밀리기 때문입니다. 파티션 12개면 LAG 12 입니다.
#   "LAG 이 항상 파티션 수와 같은 값에서 안 내려간다" 는 이 버그의 진단 신호입니다.
#
#   인자 없는 commitSync() 는 클라이언트가 알아서 +1 을 계산해 주므로
#   이 실수가 나지 않습니다. 오프셋을 직접 지정할 때만 조심하면 됩니다.

# ===========================================================================
# 정답 6
# ===========================================================================
hr "정답 6 — 컨슈머 종료 후 상태가 Empty 로 굳을 때까지 기다립니다"

# 먼저 그룹을 만들어 둡니다
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic "$TOPIC" \
  --group s06x-reset --from-beginning --max-messages 50 >/dev/null

# [터미널 B] 아래를 별도 창에서 실행해 그룹을 Stable 로 만듭니다.
#   docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
#     --bootstrap-server kafka-1:9092 --topic s06x_orders --group s06x-reset
#
# [터미널 A] 그 상태에서 리셋을 시도하면:
#   Error: Assignments can only be reset if the group 's06x-reset' is inactive,
#          but the current state is Stable.

# 성공하는 절차 — 상태가 Empty 가 될 때까지 대기하는 루프
wait_until_empty() {
  local g="$1" i
  for i in $(seq 1 60); do
    if K kafka-consumer-groups.sh --bootstrap-server "$BS" \
         --describe --group "$g" --state 2>/dev/null | grep -qE '\bEmpty\b'; then
      echo "그룹 ${g} 가 Empty 상태가 되었습니다 (${i}초 경과)"
      return 0
    fi
    sleep 1
  done
  echo "타임아웃: ${g} 가 여전히 활성 상태입니다."
  return 1
}

wait_until_empty s06x-reset
K kafka-consumer-groups.sh --bootstrap-server "$BS" \
  --group s06x-reset --topic "$TOPIC" --reset-offsets --to-earliest --execute
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s06x-reset

# 해설:
#   에러 메시지의 "inactive" 는 그룹 상태가 Empty 또는 Dead 라는 뜻입니다.
#   Stable / PreparingRebalance / CompletingRebalance 는 전부 거부됩니다.
#
#   이건 안전장치입니다. 살아 있는 컨슈머가 계속 커밋하는 중에 밖에서 되감으면
#   되감은 값이 즉시 덮어씌워지거나 같은 구간을 두 번 읽는 등
#   예측 불가능한 상태가 되기 때문입니다.
#
#   중요한 것은 "컨슈머 프로세스를 죽였다 = 즉시 Empty" 가 아니라는 점입니다.
#   컨슈머가 close() 를 정상 호출하고 종료하면 LeaveGroup 요청을 보내 바로 빠지지만,
#   kill -9 로 죽으면 브로커는 session.timeout.ms(기본 45000) 동안
#   그 멤버가 살아 있다고 간주합니다. 그래서 종료 직후 리셋은 실패합니다.
#
#   그래서 운영 절차는 반드시 3단계입니다:
#     1) 컨슈머를 전부 내린다 (파드 replicas=0)
#     2) --describe --state 가 Empty 가 될 때까지 폴링한다  ← 위 wait_until_empty
#     3) --reset-offsets --execute
#     4) 컨슈머를 다시 올린다
#
#   이 절차 동안 서비스가 멈춘다는 것이 --reset-offsets 의 실질적 비용입니다.
#   무중단으로 되감아야 한다면 애플리케이션 안에서 seek() 를 써야 합니다 (문제 7).

# ===========================================================================
# 정답 7
# ===========================================================================
hr "정답 7 — --reset-offsets 는 즉시 커밋, seek() 는 커밋하지 않음"

# (a) 50건을 전부 읽어 커밋
K kafka-consumer-groups.sh --bootstrap-server "$BS" --delete --group s06x-seek 2>/dev/null || true
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic "$TOPIC" \
  --group s06x-seek --from-beginning --max-messages 50 >/dev/null
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s06x-seek

# (b) shift-by -30
wait_until_empty s06x-seek
K kafka-consumer-groups.sh --bootstrap-server "$BS" \
  --group s06x-seek --topic "$TOPIC" --reset-offsets --shift-by -30 --execute
echo "--- (b) 이후 ---"
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s06x-seek

# 50 으로 되돌린 뒤 (c)
K kafka-consumer-groups.sh --bootstrap-server "$BS" \
  --group s06x-seek --topic "$TOPIC" --reset-offsets --to-latest --execute >/dev/null

# (c) seek 시나리오 (커밋하지 않음)
docker exec -e S06_TOPIC="$TOPIC" -e S06_GROUP=s06x-seek kafka-1 \
  sh -c 'cd /tmp && java -cp "/opt/kafka/libs/*" Practice.java seek'
echo "--- (c) 이후 ---"
K kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group s06x-seek

# 해설:
#   (b) --reset-offsets --shift-by -30 --execute
#       → CURRENT-OFFSET 50 → 20. --describe 에 즉시 반영됩니다.
#         이 명령은 __consumer_offsets 에 새 레코드를 직접 씁니다.
#         컨슈머가 없어도 동작하며, 되돌릴 수 없습니다.
#
#   (c) seek(tp, position - 30)
#       → CURRENT-OFFSET 은 50 그대로입니다. 변하지 않습니다.
#         seek() 는 컨슈머 프로세스의 메모리상 position 만 바꿉니다.
#         그 위치에서 읽더라도, 커밋하지 않고 프로세스가 죽으면
#         다음 기동 때는 원래 커밋 지점(50)으로 돌아갑니다.
#
#   (c) 를 (b) 와 같게 만들려면 seek 직후 또는 재처리 후에
#       consumer.commitSync();
#   를 호출해야 합니다. 그때 비로소 __consumer_offsets 가 갱신됩니다.
#
#   실무에서 어느 쪽을 쓰느냐:
#     - 일회성 운영 개입, 전체 재처리     → --reset-offsets (단, 서비스 중단 필요)
#     - 무중단 되감기, 조건부 재처리 로직 → seek() (서비스 안 내려도 됨)
#     - "특정 시각부터" 가 필요하면 seek 쪽은 offsetsForTimes() 로 오프셋을 먼저 구합니다.
#       이때 반환값이 null 일 수 있으니(그 시각 이후 메시지가 없는 경우) 반드시 처리하세요.

# ===========================================================================
cleanup

echo
echo "Step 06 정답 확인 완료."

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