#Step 10 — 직렬화와 스키마

cd kafka/docker
docker compose --profile registry up -d
docker compose ps

curl -s localhost:8081/subjects

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

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

printf 'O-1001\nO-1002\nO-1003\n' | docker exec -i kafka-1 \
  /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server kafka-1:9092 --topic s10_types

docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka-1:9092 --topic s10_types --from-beginning \
  --value-deserializer org.apache.kafka.common.serialization.LongDeserializer

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

docker exec kafka-1 /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 \
  --group s10-poison --topic s10_types:0 --reset-offsets --shift-by 1 --execute

printf '%s' '{"order_id":"O-1001","customer_id":"C001","amount":39000,"status":"CREATED"}' | wc -c

curl -s -X POST http://localhost:8081/subjects/orders-value/versions \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d '{"schema":"{\"type\":\"record\",\"name\":\"Order\",\"namespace\":\"shop.order\",\"fields\":[{\"name\":\"order_id\",\"type\":\"string\"},{\"name\":\"customer_id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"int\"},{\"name\":\"status\",\"type\":\"string\"}]}"}'

curl -s localhost:8081/subjects

curl -s localhost:8081/subjects/orders-value/versions

curl -s localhost:8081/subjects/orders-value/versions/latest

curl -s localhost:8081/schemas/ids/1

curl -s -X POST http://localhost:8081/compatibility/subjects/orders-value/versions/latest \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d '{"schema":"{\"type\":\"record\",\"name\":\"Order\",\"namespace\":\"shop.order\",\"fields\":[{\"name\":\"order_id\",\"type\":\"string\"},{\"name\":\"customer_id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"int\"},{\"name\":\"status\",\"type\":\"string\"},{\"name\":\"channel\",\"type\":\"string\",\"default\":\"WEB\"}]}"}'

curl -s localhost:8081/config

curl -s -X PUT http://localhost:8081/config/orders-value \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d '{"compatibility":"FULL"}'

curl -s -X DELETE http://localhost:8081/config/orders-value

docker exec schema-registry kafka-avro-console-producer \
  --bootstrap-server kafka-1:9092 --topic s10_avro \
  --property schema.registry.url=http://schema-registry:8081 \
  --property value.schema='{"type":"record","name":"Order","namespace":"shop.order","fields":[{"name":"order_id","type":"string"},{"name":"customer_id","type":"string"},{"name":"amount","type":"int"},{"name":"status","type":"string"}]}' <<'EOF'
{"order_id":"O-1001","customer_id":"C001","amount":39000,"status":"CREATED"}
EOF

docker exec kafka-1 /opt/kafka/bin/kafka-dump-log.sh \
  --files /var/lib/kafka/data/s10_avro-0/00000000000000000000.log --print-data-log

docker exec kafka-1 sh -c \
  'tail -c 60 /var/lib/kafka/data/s10_avro-0/00000000000000000000.log | xxd | tail -4'

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

docker exec schema-registry kafka-avro-console-consumer \
  --bootstrap-server kafka-1:9092 --topic s10_avro --from-beginning \
  --property schema.registry.url=http://schema-registry:8081

curl -s -X PUT http://localhost:8081/config/orders-value \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d '{"compatibility":"BACKWARD"}'

curl -s -X POST http://localhost:8081/subjects/orders-value/versions \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d '{"schema":"{\"type\":\"record\",\"name\":\"Order\",\"namespace\":\"shop.order\",\"fields\":[{\"name\":\"order_id\",\"type\":\"string\"},{\"name\":\"customer_id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"int\"},{\"name\":\"status\",\"type\":\"string\"},{\"name\":\"coupon_id\",\"type\":\"string\"}]}"}'

curl -s -X POST http://localhost:8081/subjects/orders-value/versions \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d '{"schema":"{\"type\":\"record\",\"name\":\"Order\",\"namespace\":\"shop.order\",\"fields\":[{\"name\":\"order_id\",\"type\":\"string\"},{\"name\":\"customer_id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"int\"},{\"name\":\"status\",\"type\":\"string\"},{\"name\":\"coupon_id\",\"type\":[\"null\",\"string\"],\"default\":null}]}"}'

curl -s -X PUT http://localhost:8081/config/orders-value \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d '{"compatibility":"NONE"}'

curl -s -X POST http://localhost:8081/subjects/orders-value/versions \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d '{"schema":"{\"type\":\"record\",\"name\":\"Order\",\"namespace\":\"shop.order\",\"fields\":[{\"name\":\"order_id\",\"type\":\"string\"},{\"name\":\"customer_id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"int\"},{\"name\":\"status\",\"type\":\"string\"},{\"name\":\"coupon_id\",\"type\":\"string\"}]}"}'

curl -s -X PUT http://localhost:8081/config/orders-value \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d '{"compatibility":"BACKWARD"}'

curl -s -X PUT http://localhost:8081/config/orders-value \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d '{"compatibility":"FORWARD"}'

# FORWARD 에서는 기본값 없는 필드 추가가 허용됩니다
curl -s -X POST http://localhost:8081/compatibility/subjects/orders-value/versions/latest \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d '{"schema":"{\"type\":\"record\",\"name\":\"Order\",\"namespace\":\"shop.order\",\"fields\":[{\"name\":\"order_id\",\"type\":\"string\"},{\"name\":\"customer_id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"int\"},{\"name\":\"status\",\"type\":\"string\"},{\"name\":\"coupon_id\",\"type\":\"string\"},{\"name\":\"channel\",\"type\":\"string\"}]}"}'

curl -s -X POST http://localhost:8081/compatibility/subjects/orders-value/versions/latest \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d '{"schema":"{\"type\":\"record\",\"name\":\"Order\",\"namespace\":\"shop.order\",\"fields\":[{\"name\":\"order_id\",\"type\":\"string\"},{\"name\":\"customer_id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"int\"}]}"}'

# amount 를 int → long 으로 (BACKWARD 관점에서 허용되는 승격)
curl -s -X PUT localhost:8081/config/orders-value \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' -d '{"compatibility":"BACKWARD"}' > /dev/null

curl -s -X POST http://localhost:8081/compatibility/subjects/orders-value/versions/latest \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d '{"schema":"{\"type\":\"record\",\"name\":\"Order\",\"namespace\":\"shop.order\",\"fields\":[{\"name\":\"order_id\",\"type\":\"string\"},{\"name\":\"customer_id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"long\"},{\"name\":\"status\",\"type\":\"string\"}]}"}'

# subject 소프트 삭제
curl -s -X DELETE localhost:8081/subjects/orders-value

# 완전 삭제 (소프트 삭제 후에만 가능)
curl -s -X DELETE 'localhost:8081/subjects/orders-value?permanent=true'
curl -s -X DELETE 'localhost:8081/subjects/s10_avro-value?permanent=true'
curl -s localhost:8081/subjects

# 실습 토픽 삭제
docker exec kafka-1 /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka-1:9092 \
  --delete --topic s10_types
docker exec kafka-1 /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka-1:9092 \
  --delete --topic s10_avro

# 컨슈머 그룹 삭제
docker exec kafka-1 /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 \
  --delete --group s10-poison

docker compose stop schema-registry

#!/usr/bin/env bash
# ---------------------------------------------------------------------------
# Step 10 — 직렬화와 스키마 : 본문 실습 스크립트
#
# 실행법:
#   bash step-10-serialization/practice.sh          # 통째로
#   bash -x step-10-serialization/practice.sh       # 한 줄씩 확인하며 (권장)
#
# 전제:
#   - kafka/docker 에서 `docker compose --profile registry up -d` 가 되어 있어야 합니다.
#     (스크립트 [10-0] 이 직접 띄우고 준비될 때까지 기다립니다)
#   - jq 가 있으면 씁니다. 없으면 python3 폴백을 씁니다.
# ---------------------------------------------------------------------------
set -euo pipefail

BS=kafka-1:9092
SR_URL=http://localhost:8081

# 컨테이너 안 CLI 호출 헬퍼 : K kafka-topics.sh --list ...
K() { docker exec kafka-1 /opt/kafka/bin/"$@"; }
# Schema Registry 안에 들어 있는 avro 콘솔 도구 헬퍼
SRX() { docker exec schema-registry "$@"; }

# JSON 문자열 이스케이프 (Avro 스키마를 {"schema":"..."} 안에 넣기 위해)
json_escape() {
  if command -v jq >/dev/null 2>&1; then
    jq -Rs .
  else
    python3 -c 'import json,sys; print(json.dumps(sys.stdin.read()))'
  fi
}

# heredoc 으로 받은 Avro 스키마를 subject 에 등록
#   register_schema orders-value <<'EOF' ... EOF
register_schema() {
  local subject="$1"
  local body
  body=$(cat | tr -d '\n' | json_escape)
  curl -s -X POST "$SR_URL/subjects/$subject/versions" \
    -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
    -d "{\"schema\": $body}"
  echo
}

# 등록하지 않고 호환성만 검사
check_compat() {
  local subject="$1"
  local body
  body=$(cat | tr -d '\n' | json_escape)
  curl -s -X POST "$SR_URL/compatibility/subjects/$subject/versions/latest" \
    -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
    -d "{\"schema\": $body}"
  echo
}

set_compat() {
  curl -s -X PUT "$SR_URL/config/$1" \
    -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
    -d "{\"compatibility\":\"$2\"}"
  echo
}

# ===========================================================================
# [10-0] 실습 준비 — Schema Registry 기동과 헬스체크
# ===========================================================================
echo "### [10-0] Schema Registry 기동"
( cd "$(dirname "$0")/../docker" && docker compose --profile registry up -d )

echo "### [10-0] /subjects 가 응답할 때까지 최대 60초 대기"
for i in $(seq 1 60); do
  if curl -sf "$SR_URL/subjects" >/dev/null 2>&1; then
    echo "  ready after ${i}s"
    break
  fi
  sleep 1
done
curl -s "$SR_URL/subjects"; echo

# Registry 가 스키마를 저장하는 내부 토픽. cleanup.policy=compact 임을 확인합니다.
echo "### [10-0] _schemas 토픽 — 파티션 1개, compact"
K kafka-topics.sh --bootstrap-server $BS --describe --topic _schemas

# ===========================================================================
# [10-2] 함정 A — Deserializer 불일치 (poison pill)
# ===========================================================================
echo "### [10-2] 실습 토픽 생성 후 String 으로 3건 투입"
K kafka-topics.sh --bootstrap-server $BS \
  --create --topic s10_types --partitions 1 --replication-factor 3 || true

printf 'O-1001\nO-1002\nO-1003\n' | docker exec -i kafka-1 \
  /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server $BS --topic s10_types

# ↓ 여기는 "일부러 실패하는" 구간입니다. set -e 로 죽지 않도록 || true 를 붙입니다.
#   기대 출력:
#   org.apache.kafka.common.errors.SerializationException:
#     Size of data received by LongDeserializer is not 8
echo "### [10-2] LongDeserializer 로 읽기 — 실패가 정상입니다"
timeout 20 docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server $BS --topic s10_types --from-beginning \
  --group s10-poison \
  --value-deserializer org.apache.kafka.common.serialization.LongDeserializer || true

# 오프셋이 0 에서 멈춰 있고 LAG 가 3 인 것을 확인합니다. 이것이 poison pill 입니다.
echo "### [10-2] 랙 확인 — CURRENT-OFFSET 0 / LAG 3"
K kafka-consumer-groups.sh --bootstrap-server $BS --describe --group s10-poison || true

# 응급 복구: 오프셋을 한 칸 건너뜁니다. (컨슈머가 떠 있으면 거부됩니다)
echo "### [10-2] --shift-by 1 로 poison 메시지 건너뛰기"
K kafka-consumer-groups.sh --bootstrap-server $BS \
  --group s10-poison --topic s10_types:0 --reset-offsets --shift-by 1 --execute || true

# ===========================================================================
# [10-3] JSON 크기 실측
# ===========================================================================
echo "### [10-3] JSON 한 건의 바이트 수"
printf '%s' '{"order_id":"O-1001","customer_id":"C001","amount":39000,"status":"CREATED"}' | wc -c

echo "### [10-3] 필드 7개짜리 완전한 레코드"
printf '%s' '{"order_id":"O-1001","customer_id":"C001","amount":39000,"status":"CREATED","created_at":"2024-03-11T10:22:00Z","channel":"WEB"}' | wc -c

# ===========================================================================
# [10-4] Schema Registry REST API 전체
# ===========================================================================
echo "### [10-4] orders-value 스키마 v1 등록 → {\"id\":1}"
register_schema orders-value <<'AVRO'
{
  "type": "record",
  "name": "Order",
  "namespace": "shop.order",
  "fields": [
    {"name": "order_id",    "type": "string"},
    {"name": "customer_id", "type": "string"},
    {"name": "amount",      "type": "int"},
    {"name": "status",      "type": "string"}
  ]
}
AVRO

echo "### [10-4] subject 목록"
curl -s "$SR_URL/subjects"; echo

echo "### [10-4] 버전 목록"
curl -s "$SR_URL/subjects/orders-value/versions"; echo

echo "### [10-4] 최신 버전 전체"
curl -s "$SR_URL/subjects/orders-value/versions/latest"; echo

echo "### [10-4] ID 로 조회 — 컨슈머가 실제로 쓰는 엔드포인트"
curl -s "$SR_URL/schemas/ids/1"; echo

echo "### [10-4] 호환성 검사만 (등록하지 않음) — default 있는 필드 추가 → true"
check_compat orders-value <<'AVRO'
{
  "type": "record", "name": "Order", "namespace": "shop.order",
  "fields": [
    {"name": "order_id",    "type": "string"},
    {"name": "customer_id", "type": "string"},
    {"name": "amount",      "type": "int"},
    {"name": "status",      "type": "string"},
    {"name": "channel",     "type": "string", "default": "WEB"}
  ]
}
AVRO

echo "### [10-4] 전역 호환성 수준"
curl -s "$SR_URL/config"; echo

echo "### [10-4] subject 단위 호환성 변경 → FULL"
set_compat orders-value FULL
curl -s "$SR_URL/config/orders-value"; echo

echo "### [10-4] subject 단위 설정 해제 → 전역값으로 복귀"
curl -s -X DELETE "$SR_URL/config/orders-value"; echo

# ===========================================================================
# [10-5] 와이어 포맷 — 매직바이트 0x00 + 스키마 ID 4바이트
# ===========================================================================
echo "### [10-5] Avro 토픽 생성"
K kafka-topics.sh --bootstrap-server $BS \
  --create --topic s10_avro --partitions 1 --replication-factor 3 || true

echo "### [10-5] kafka-avro-console-producer 로 1건 쓰기"
SRX kafka-avro-console-producer \
  --bootstrap-server $BS --topic s10_avro \
  --property schema.registry.url=http://schema-registry:8081 \
  --property value.schema='{"type":"record","name":"Order","namespace":"shop.order","fields":[{"name":"order_id","type":"string"},{"name":"customer_id","type":"string"},{"name":"amount","type":"int"},{"name":"status","type":"string"}]}' <<'EOF'
{"order_id":"O-1001","customer_id":"C001","amount":39000,"status":"CREATED"}
EOF

echo "### [10-5] 로그 세그먼트 덤프 — valueSize 가 39(=34+5) 인지 확인"
K kafka-dump-log.sh \
  --files /var/lib/kafka/data/s10_avro-0/00000000000000000000.log --print-data-log

# 전체를 xxd 하면 배치 헤더까지 쏟아지므로 끝부분 60바이트만 봅니다.
echo "### [10-5] xxd 로 실제 바이트 — 00 | 00 00 00 NN | Avro 페이로드"
docker exec kafka-1 sh -c \
  'tail -c 60 /var/lib/kafka/data/s10_avro-0/00000000000000000000.log | xxd | tail -4'

echo "### [10-5] 일반 console-consumer 로 읽으면 깨져 보입니다 (도구가 틀린 것)"
# timeout 은 셸 함수에 못 걸리므로 여기서는 K() 를 쓰지 않고 전체 명령을 씁니다.
docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server $BS --topic s10_avro --from-beginning --timeout-ms 5000 || true

echo "### [10-5] kafka-avro-console-consumer 로 읽으면 정상"
docker exec schema-registry kafka-avro-console-consumer \
  --bootstrap-server $BS --topic s10_avro --from-beginning \
  --property schema.registry.url=http://schema-registry:8081 \
  --timeout-ms 5000 || true

echo "### [10-5] 이 토픽의 subject 는 TopicNameStrategy 로 자동 생성됩니다"
curl -s "$SR_URL/subjects"; echo

# ===========================================================================
# [10-8] 함정 B — 스키마 변경이 컨슈머를 깨뜨린다
# ===========================================================================
echo "### [10-8] (1) BACKWARD 로 되돌리고, 기본값 없는 필드 추가 → 409 거부"
set_compat orders-value BACKWARD

register_schema orders-value <<'AVRO'
{
  "type": "record", "name": "Order", "namespace": "shop.order",
  "fields": [
    {"name": "order_id",    "type": "string"},
    {"name": "customer_id", "type": "string"},
    {"name": "amount",      "type": "int"},
    {"name": "status",      "type": "string"},
    {"name": "coupon_id",   "type": "string"}
  ]
}
AVRO
# 기대: {"error_code":409,... errorType:'READER_FIELD_MISSING_DEFAULT_VALUE' ...}

echo "### [10-8] (1') 기본값을 붙이면 통과 — nullable 은 [\"null\",\"T\"] + default null"
register_schema orders-value <<'AVRO'
{
  "type": "record", "name": "Order", "namespace": "shop.order",
  "fields": [
    {"name": "order_id",    "type": "string"},
    {"name": "customer_id", "type": "string"},
    {"name": "amount",      "type": "int"},
    {"name": "status",      "type": "string"},
    {"name": "coupon_id",   "type": ["null","string"], "default": null}
  ]
}
AVRO

echo "### [10-8] (2) 호환성을 NONE 으로 — 검사 장치를 스스로 꺼 봅니다"
set_compat orders-value NONE

echo "### [10-8] (2') 방금 거부됐던 그 변경을 그대로 다시 — 이번엔 성공합니다"
register_schema orders-value <<'AVRO'
{
  "type": "record", "name": "Order", "namespace": "shop.order",
  "fields": [
    {"name": "order_id",    "type": "string"},
    {"name": "customer_id", "type": "string"},
    {"name": "amount",      "type": "int"},
    {"name": "status",      "type": "string"},
    {"name": "coupon_id",   "type": "string"}
  ]
}
AVRO
# 등록은 성공하지만, v1 스키마를 가진 옛 컨슈머는 런타임에 이렇게 죽습니다:
#   org.apache.avro.AvroTypeException: ... missing required field coupon_id
# 그리고 이 예외는 [10-2] 의 poison pill 과 완전히 같은 형태입니다.

echo "### [10-8] (2'') 호환성 즉시 복구 — NONE 을 켜 둔 채로 넘어가면 안 됩니다"
set_compat orders-value BACKWARD

echo "### [10-8] (3) FORWARD 에서는 기본값 없는 필드 추가가 허용됩니다"
set_compat orders-value FORWARD
check_compat orders-value <<'AVRO'
{
  "type": "record", "name": "Order", "namespace": "shop.order",
  "fields": [
    {"name": "order_id",    "type": "string"},
    {"name": "customer_id", "type": "string"},
    {"name": "amount",      "type": "int"},
    {"name": "status",      "type": "string"},
    {"name": "coupon_id",   "type": "string"},
    {"name": "channel",     "type": "string"}
  ]
}
AVRO

echo "### [10-8] (3') 반대로 FORWARD 에서 기본값 없는 필드 '삭제'는 거부됩니다"
check_compat orders-value <<'AVRO'
{
  "type": "record", "name": "Order", "namespace": "shop.order",
  "fields": [
    {"name": "order_id",    "type": "string"},
    {"name": "customer_id", "type": "string"},
    {"name": "amount",      "type": "int"}
  ]
}
AVRO

# ===========================================================================
# [10-9] 스키마 진화 규칙 — 타입 승격
# ===========================================================================
echo "### [10-9] BACKWARD 로 되돌리고 int → long 승격 검사 → true"
set_compat orders-value BACKWARD
check_compat orders-value <<'AVRO'
{
  "type": "record", "name": "Order", "namespace": "shop.order",
  "fields": [
    {"name": "order_id",    "type": "string"},
    {"name": "customer_id", "type": "string"},
    {"name": "amount",      "type": "long"},
    {"name": "status",      "type": "string"}
  ]
}
AVRO

echo "### [10-9] 이름 변경은 aliases 로 — 그냥 바꾸면 삭제+추가로 해석됩니다"
check_compat orders-value <<'AVRO'
{
  "type": "record", "name": "Order", "namespace": "shop.order",
  "fields": [
    {"name": "order_id",     "type": "string"},
    {"name": "customer_ref", "type": "string", "aliases": ["customer_id"], "default": "UNKNOWN"},
    {"name": "amount",       "type": "int"},
    {"name": "status",       "type": "string"}
  ]
}
AVRO

# ===========================================================================
# [10-10] 정리 (실습 마무리)
# ===========================================================================
echo "### [10-10] subject 소프트 삭제 → 완전 삭제"
curl -s -X DELETE "$SR_URL/subjects/orders-value" || true; echo
curl -s -X DELETE "$SR_URL/subjects/orders-value?permanent=true" || true; echo
curl -s -X DELETE "$SR_URL/subjects/s10_avro-value" || true; echo
curl -s -X DELETE "$SR_URL/subjects/s10_avro-value?permanent=true" || true; echo
curl -s "$SR_URL/subjects"; echo    # [] 이어야 정상

echo "### [10-10] 실습 토픽 / 컨슈머 그룹 삭제"
K kafka-topics.sh --bootstrap-server $BS --delete --topic s10_types || true
K kafka-topics.sh --bootstrap-server $BS --delete --topic s10_avro  || true
K kafka-consumer-groups.sh --bootstrap-server $BS --delete --group s10-poison || true

echo "### [10-10] Schema Registry 는 Step 12·13 에서 쓰지 않으므로 내려도 됩니다"
echo "  ( cd kafka/docker && docker compose stop schema-registry )"

echo "=== Step 10 practice 완료 ==="

#!/usr/bin/env bash
# ---------------------------------------------------------------------------
# Step 10 — 직렬화와 스키마 : 연습문제 (6문제)
#
# 실행법:
#   bash step-10-serialization/exercise.sh
#   (권장) 편집기로 열어 `# 여기에 작성:` 자리를 채운 뒤 한 블록씩 실행
#
# 전제: docker compose --profile registry up -d 로 Schema Registry 가 떠 있을 것
# 정답: solution.sh
# ---------------------------------------------------------------------------
set -uo pipefail   # 문제 풀이 중 실패해도 계속 진행하도록 -e 는 뺐습니다

BS=kafka-1:9092
SR_URL=http://localhost:8081

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

json_escape() {
  if command -v jq >/dev/null 2>&1; then jq -Rs .
  else python3 -c 'import json,sys; print(json.dumps(sys.stdin.read()))'; fi
}

# ===========================================================================
# 준비 — 문제지가 미리 만들어 두는 것들
# ===========================================================================
echo "### 준비: s10_ex 토픽 생성 + 문자열 3건 투입"
K kafka-topics.sh --bootstrap-server $BS \
  --create --topic s10_ex --partitions 1 --replication-factor 3 2>/dev/null || true
printf 'O-2001\nO-2002\nO-2003\n' | docker exec -i kafka-1 \
  /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server $BS --topic s10_ex


# ===========================================================================
# 문제 1.
#   s10_ex 에는 String 으로 쓴 'O-2001' 같은 값이 들어 있습니다.
#   이것을 IntegerDeserializer 로 읽는 콘솔 컨슈머 명령을 작성하고,
#   출력되는 예외 메시지를 정확히 옮겨 적으십시오.
#   그룹 이름은 s10-ex 를 쓰십시오.
#
#   힌트: --value-deserializer 옵션. 클래스 풀네임은
#         org.apache.kafka.common.serialization.IntegerDeserializer
# ===========================================================================
echo "### 문제 1"
# 여기에 작성:


# 관찰한 예외 메시지 (주석으로 적으십시오):
#


# ===========================================================================
# 문제 2.
#   문제 1 의 컨슈머 그룹 s10-ex 의 랙을 확인하십시오.
#   CURRENT-OFFSET 이 진전하지 않는 것을 확인한 뒤,
#   오프셋을 1 칸 건너뛰어 컨슈머가 다시 돌 수 있게 복구하십시오.
#
#   ⚠️ --reset-offsets 는 그룹에 **활성 멤버가 있으면 거부**됩니다.
#      Error: Assignments can only be reset if the group 's10-ex' is inactive
#      문제 1 의 컨슈머를 반드시 먼저 종료하십시오.
#   힌트: --reset-offsets --topic s10_ex:0 --shift-by N --dry-run / --execute
# ===========================================================================
echo "### 문제 2"
# (a) 랙 확인
# 여기에 작성:


# (b) dry-run 으로 어떻게 바뀔지 먼저 확인
# 여기에 작성:


# (c) 실제 적용
# 여기에 작성:


# ===========================================================================
# 문제 3.
#   아래 heredoc 의 Payment 스키마를 payments-value subject 에 등록하고,
#   응답으로 받은 id 를 변수에 담아 /schemas/ids/{id} 로 다시 조회하십시오.
#
#   힌트: json_escape 로 스키마 본문을 문자열화한 뒤
#         {"schema": <escaped>} 로 감싸 POST 합니다.
#         id 추출은 jq -r .id 또는 grep -o '[0-9]*'
# ===========================================================================
echo "### 문제 3"
PAYMENT_SCHEMA=$(cat <<'AVRO' | tr -d '\n'
{
  "type": "record", "name": "Payment", "namespace": "shop.payment",
  "fields": [
    {"name": "order_id", "type": "string"},
    {"name": "method",   "type": "string"},
    {"name": "amount",   "type": "int"},
    {"name": "result",   "type": "string"}
  ]
}
AVRO
)
ESCAPED=$(printf '%s' "$PAYMENT_SCHEMA" | json_escape)

# (a) 등록 — {"id":N} 이 나와야 합니다
# 여기에 작성:


# (b) 발급된 id 를 변수 ID 에 담기
# 여기에 작성:
# ID=

# (c) /schemas/ids/$ID 로 재조회
# 여기에 작성:


# ===========================================================================
# 문제 4.
#   payments-value 의 호환성을 BACKWARD 로 설정한 뒤,
#   "기본값 없는 refund_id 필드 추가" 를 호환성 검사(등록 아님)에 넣으십시오.
#   is_compatible 이 어떻게 나옵니까?
#
#   힌트: PUT /config/payments-value, POST /compatibility/subjects/.../versions/latest
# ===========================================================================
echo "### 문제 4"
V2_SCHEMA=$(cat <<'AVRO' | tr -d '\n'
{
  "type": "record", "name": "Payment", "namespace": "shop.payment",
  "fields": [
    {"name": "order_id",  "type": "string"},
    {"name": "method",    "type": "string"},
    {"name": "amount",    "type": "int"},
    {"name": "result",    "type": "string"},
    {"name": "refund_id", "type": "string"}
  ]
}
AVRO
)
V2_ESCAPED=$(printf '%s' "$V2_SCHEMA" | json_escape)

# (a) 호환성을 BACKWARD 로
# 여기에 작성:


# (b) 호환성 검사
# 여기에 작성:


# 관찰한 결과 (주석):
#   is_compatible =
#   errorType     =


# ===========================================================================
# 문제 5.
#   호환성을 FORWARD 로 바꾸고 문제 4 와 **완전히 같은** 검사를 다시 하십시오.
#   판정이 뒤집힙니까? 왜 그렇습니까?
# ===========================================================================
echo "### 문제 5"
# (a) 호환성을 FORWARD 로
# 여기에 작성:


# (b) 문제 4 와 같은 검사
# 여기에 작성:


# 왜 뒤집혔는지 한 줄로 (주석):
#


# ===========================================================================
# 문제 6.
#   아래에서 문제지가 Avro 로 1건을 s10_ex_avro 토픽에 씁니다.
#   그 로그 세그먼트를 xxd 로 열어 앞 5바이트(매직 + 스키마 ID)를 읽어내고,
#   읽어낸 ID 로 Registry 를 조회하십시오.
#
#   힌트: 앞 1바이트가 0x00(매직), 다음 4바이트가 빅엔디언 스키마 ID 입니다.
#         세그먼트 경로는 /var/lib/kafka/data/s10_ex_avro-0/00000000000000000000.log
#         tail -c 60 으로 끝부분만 잘라 보는 편이 찾기 쉽습니다.
# ===========================================================================
echo "### 문제 6"
K kafka-topics.sh --bootstrap-server $BS \
  --create --topic s10_ex_avro --partitions 1 --replication-factor 3 2>/dev/null || true
SRX kafka-avro-console-producer \
  --bootstrap-server $BS --topic s10_ex_avro \
  --property schema.registry.url=http://schema-registry:8081 \
  --property value.schema='{"type":"record","name":"Payment","namespace":"shop.payment","fields":[{"name":"order_id","type":"string"},{"name":"method","type":"string"},{"name":"amount","type":"int"},{"name":"result","type":"string"}]}' <<'EOF'
{"order_id":"O-2001","method":"CARD","amount":39000,"result":"APPROVED"}
EOF

# (a) xxd 로 바이트 보기
# 여기에 작성:


# (b) 읽어낸 스키마 ID 로 Registry 조회
# 여기에 작성:


# 읽어낸 ID (주석):
#


# ===========================================================================
# 정리 — 문제를 다 못 풀었어도 이 블록만 실행하면 환경이 깨끗해집니다
# ===========================================================================
echo "### 정리"
curl -s -X DELETE "$SR_URL/subjects/payments-value"                 >/dev/null 2>&1
curl -s -X DELETE "$SR_URL/subjects/payments-value?permanent=true"  >/dev/null 2>&1
curl -s -X DELETE "$SR_URL/subjects/s10_ex_avro-value"                >/dev/null 2>&1
curl -s -X DELETE "$SR_URL/subjects/s10_ex_avro-value?permanent=true" >/dev/null 2>&1
K kafka-topics.sh --bootstrap-server $BS --delete --topic s10_ex       2>/dev/null || true
K kafka-topics.sh --bootstrap-server $BS --delete --topic s10_ex_avro  2>/dev/null || true
K kafka-consumer-groups.sh --bootstrap-server $BS --delete --group s10-ex 2>/dev/null || true
echo "남은 subject:"; curl -s "$SR_URL/subjects"; echo

#!/usr/bin/env bash
# ---------------------------------------------------------------------------
# Step 10 — 직렬화와 스키마 : 연습문제 정답 + 해설
#
# 실행법:
#   bash step-10-serialization/solution.sh
#
# exercise.sh 를 직접 풀어 본 뒤에 여십시오.
# ---------------------------------------------------------------------------
set -uo pipefail

BS=kafka-1:9092
SR_URL=http://localhost:8081

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

json_escape() {
  if command -v jq >/dev/null 2>&1; then jq -Rs .
  else python3 -c 'import json,sys; print(json.dumps(sys.stdin.read()))'; fi
}

echo "### 준비"
K kafka-topics.sh --bootstrap-server $BS \
  --create --topic s10_ex --partitions 1 --replication-factor 3 2>/dev/null || true
printf 'O-2001\nO-2002\nO-2003\n' | docker exec -i kafka-1 \
  /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server $BS --topic s10_ex


# ===========================================================================
# 정답 1 — IntegerDeserializer 로 읽기
# ===========================================================================
echo "### 정답 1"
timeout 20 docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server $BS --topic s10_ex --from-beginning \
  --group s10-ex \
  --value-deserializer org.apache.kafka.common.serialization.IntegerDeserializer || true

# 해설:
#   출력:
#     org.apache.kafka.common.errors.SerializationException:
#       Size of data received by IntegerDeserializer is not 4
#     Processed a total of 0 messages
#
#   'O-2001' 은 UTF-8 로 6바이트입니다. IntegerDeserializer 는 **정확히 4바이트**를
#   요구하는 고정 길이 디코더이므로, 길이가 다르면 즉시 예외를 던집니다.
#   Long 이면 8, Integer 면 4, UUID 면 36(문자열 표현) 입니다.
#
#   ⚠️ 여기서 진짜 무서운 것은 "길이가 우연히 맞는 경우" 입니다.
#      값이 'ABCD'(4바이트) 였다면 IntegerDeserializer 는 **아무 에러도 내지 않고**
#      0x41424344 = 1094861636 이라는 정수를 돌려줍니다.
#      길이가 안 맞으면 시끄럽게 실패하지만, 맞으면 조용히 쓰레기를 돌려줍니다.
#      이 코스의 정체성이 바로 이 두 번째 경우입니다. 타입은 사람이 맞춰야 합니다.
#
#   운영에서라면: 프로듀서/컨슈머의 (de)serializer 설정을 같은 저장소의
#   같은 상수로 관리하거나, Schema Registry 를 써서 타입을 데이터에 붙이십시오.


# ===========================================================================
# 정답 2 — 랙 확인과 오프셋 건너뛰기
# ===========================================================================
echo "### 정답 2 (a) 랙 확인"
K kafka-consumer-groups.sh --bootstrap-server $BS --describe --group s10-ex || true

# 해설 (a):
#   GROUP    TOPIC    PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  ...
#   s10-ex   s10_ex   0          0               3               3    -   -   -
#
#   CURRENT-OFFSET 이 0 입니다. 컨슈머가 붙었다 죽기를 반복해도 이 값은 절대
#   올라가지 않습니다. 역직렬화는 **오프셋을 커밋하기 전에** 실패하기 때문입니다.
#   컨슈머가 이미 종료됐으므로 CONSUMER-ID / HOST / CLIENT-ID 는 '-' 로 나옵니다.
#   살아 있는 채로 무한 재시도 중이라면 여기에 consumer-s10-ex-1-... 이 찍힙니다.

echo "### 정답 2 (b) dry-run"
K kafka-consumer-groups.sh --bootstrap-server $BS \
  --group s10-ex --topic s10_ex:0 --reset-offsets --shift-by 1 --dry-run || true

# 해설 (b):
#   GROUP                          TOPIC                          PARTITION  NEW-OFFSET
#   s10-ex                         s10_ex                         0          1
#
#   --dry-run 은 **아무것도 바꾸지 않고** 결과만 보여줍니다.
#   --execute 는 되돌릴 수 없으므로, 운영에서는 반드시 dry-run 을 먼저 돌리십시오.
#   특히 --to-earliest 를 실수로 쓰면 그룹 전체가 처음부터 재처리를 시작합니다.
#   결제 컨슈머라면 그것만으로 사고입니다.

echo "### 정답 2 (c) 실제 적용"
K kafka-consumer-groups.sh --bootstrap-server $BS \
  --group s10-ex --topic s10_ex:0 --reset-offsets --shift-by 1 --execute || true

# 해설 (c):
#   ⚠️ 활성 멤버가 있으면 이렇게 거부됩니다:
#     Error: Assignments can only be reset if the group 's10-ex' is inactive,
#     but the current state is Stable.
#   컨슈머를 전부 내리고 다시 실행해야 합니다. 이 제약은 "두 주체가 동시에
#   오프셋을 쓰는 것"을 막기 위한 것이며, 우회 방법은 없습니다.
#
#   운영에서라면: 오프셋을 건너뛰기 전에 **버려질 메시지를 반드시 떠서 보관**하십시오.
#     K kafka-dump-log.sh --files /var/lib/kafka/data/s10_ex-0/00000000000000000000.log \
#       --print-data-log | grep 'offset: 0 '
#   건너뛴 메시지는 다시 찾을 수 없고, "무엇을 버렸는지" 를 나중에 반드시 묻습니다.


# ===========================================================================
# 정답 3 — 스키마 등록과 ID 재조회
# ===========================================================================
echo "### 정답 3"
PAYMENT_SCHEMA=$(cat <<'AVRO' | tr -d '\n'
{
  "type": "record", "name": "Payment", "namespace": "shop.payment",
  "fields": [
    {"name": "order_id", "type": "string"},
    {"name": "method",   "type": "string"},
    {"name": "amount",   "type": "int"},
    {"name": "result",   "type": "string"}
  ]
}
AVRO
)
ESCAPED=$(printf '%s' "$PAYMENT_SCHEMA" | json_escape)

# (a) 등록
RESP=$(curl -s -X POST "$SR_URL/subjects/payments-value/versions" \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d "{\"schema\": $ESCAPED}")
echo "$RESP"        # {"id":6}  (앞 실습에서 쓴 ID 다음 번호가 붙습니다)

# (b) id 추출
if command -v jq >/dev/null 2>&1; then
  ID=$(printf '%s' "$RESP" | jq -r .id)
else
  ID=$(printf '%s' "$RESP" | sed -n 's/.*"id":\([0-9]*\).*/\1/p')
fi
echo "ID=$ID"

# (c) ID 로 재조회
curl -s "$SR_URL/schemas/ids/$ID"; echo

# 해설:
#   이 두 단계가 **컨슈머가 매 메시지마다 하는 일** 그 자체입니다.
#     ① 메시지 앞 5바이트에서 스키마 ID 를 읽는다
#     ② GET /schemas/ids/{id} 로 스키마를 가져온다
#     ③ 그 스키마로 나머지 바이트를 역직렬화한다
#   ②는 클라이언트가 캐시하므로 ID 당 **딱 한 번**만 일어납니다. 그래서
#   Schema Registry 가 잠깐 죽어도 이미 본 ID 의 메시지는 계속 처리됩니다.
#   반대로 **새 스키마 ID 가 처음 등장하는 순간** Registry 가 죽어 있으면
#   그 컨슈머는 그 자리에서 멈춥니다. Registry 는 조용한 단일 장애점입니다.
#
#   운영에서라면: Registry 를 최소 2대 이상 띄우고(리더 선출은 자동),
#   클라이언트에 여러 URL 을 콤마로 주십시오.
#     schema.registry.url=http://sr-1:8081,http://sr-2:8081


# ===========================================================================
# 정답 4 — BACKWARD 에서 기본값 없는 필드 추가
# ===========================================================================
echo "### 정답 4"
V2_SCHEMA=$(cat <<'AVRO' | tr -d '\n'
{
  "type": "record", "name": "Payment", "namespace": "shop.payment",
  "fields": [
    {"name": "order_id",  "type": "string"},
    {"name": "method",    "type": "string"},
    {"name": "amount",    "type": "int"},
    {"name": "result",    "type": "string"},
    {"name": "refund_id", "type": "string"}
  ]
}
AVRO
)
V2_ESCAPED=$(printf '%s' "$V2_SCHEMA" | json_escape)

# (a) 호환성을 BACKWARD 로
curl -s -X PUT "$SR_URL/config/payments-value" \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d '{"compatibility":"BACKWARD"}'; echo

# (b) 검사
curl -s -X POST "$SR_URL/compatibility/subjects/payments-value/versions/latest" \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d "{\"schema\": $V2_ESCAPED}"; echo

# 해설:
#   결과:
#   {"is_compatible":false,"messages":["{errorType:'READER_FIELD_MISSING_DEFAULT_VALUE',
#     description:'The field 'refund_id' at path '/fields/4' in the new schema
#     has no default value.', additionalInfo:'refund_id'}"]}
#
#   BACKWARD = "새 스키마(v2)로 만든 **리더**가 옛 데이터(v1)를 읽을 수 있는가".
#   v1 로 쓴 메시지에는 refund_id 바이트가 아예 없습니다. v2 리더는 그 자리를
#   무엇으로 채울지 알 수 없고, 채울 값(default)이 지정되어 있지 않으므로 실패합니다.
#   → 그래서 errorType 이 READER_FIELD_MISSING_DEFAULT_VALUE 입니다.
#
#   고치는 법은 하나뿐입니다. 기본값을 주십시오.
#     {"name":"refund_id","type":["null","string"],"default":null}
#   union 을 쓸 때 **default 의 타입은 union 의 첫 번째 타입과 일치**해야 합니다.
#   ["string","null"] 에 default null 을 주면 스키마 파싱 단계에서 죽습니다.


# ===========================================================================
# 정답 5 — FORWARD 로 바꾸면 판정이 뒤집힌다
# ===========================================================================
echo "### 정답 5"
curl -s -X PUT "$SR_URL/config/payments-value" \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d '{"compatibility":"FORWARD"}'; echo

curl -s -X POST "$SR_URL/compatibility/subjects/payments-value/versions/latest" \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d "{\"schema\": $V2_ESCAPED}"; echo

# 해설:
#   결과: {"is_compatible":true}
#
#   FORWARD = "옛 스키마(v1)로 만든 **리더**가 새 데이터(v2)를 읽을 수 있는가".
#   v1 리더는 자기가 모르는 refund_id 필드를 만나면 그냥 **건너뜁니다**.
#   채울 값이 필요 없으므로 default 도 필요 없습니다. 그래서 통과합니다.
#
#   한 줄 요약:
#     BACKWARD → 새 리더가 옛 데이터를 읽는다 → **컨슈머를 먼저 배포**
#     FORWARD  → 옛 리더가 새 데이터를 읽는다 → **프로듀서를 먼저 배포**
#
#   같은 변경이 어떤 수준에서는 막히고 어떤 수준에서는 통과합니다.
#   즉 호환성 수준은 "안전/불안전" 의 등급이 아니라 **배포 순서의 선언**입니다.
#   조직이 배포 순서를 통제할 수 없다면 FULL_TRANSITIVE 를 쓰는 것이
#   결국 가장 싸게 먹힙니다(기본값 있는 추가·삭제만 허용).
#
#   운영에서라면: 이 compatibility 검사를 **CI 파이프라인에 넣으십시오.**
#   빌드 단계에서 curl 한 번으로 is_compatible 을 확인하고 false 면 빌드 실패.
#   운영 배포 후에 알게 되는 것과 비용이 100배 차이납니다.

# 뒷정리를 위해 BACKWARD 로 복구
curl -s -X PUT "$SR_URL/config/payments-value" \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d '{"compatibility":"BACKWARD"}' >/dev/null


# ===========================================================================
# 정답 6 — 와이어 포맷 바이트 읽기
# ===========================================================================
echo "### 정답 6"
K kafka-topics.sh --bootstrap-server $BS \
  --create --topic s10_ex_avro --partitions 1 --replication-factor 3 2>/dev/null || true
SRX kafka-avro-console-producer \
  --bootstrap-server $BS --topic s10_ex_avro \
  --property schema.registry.url=http://schema-registry:8081 \
  --property value.schema='{"type":"record","name":"Payment","namespace":"shop.payment","fields":[{"name":"order_id","type":"string"},{"name":"method","type":"string"},{"name":"amount","type":"int"},{"name":"result","type":"string"}]}' <<'EOF'
{"order_id":"O-2001","method":"CARD","amount":39000,"result":"APPROVED"}
EOF

# (a) xxd
docker exec kafka-1 sh -c \
  'tail -c 60 /var/lib/kafka/data/s10_ex_avro-0/00000000000000000000.log | xxd | tail -4'

# 해설 (a):
#   00000000: 0000 0000 0700 0000 0000 0000 0000 0000  ................
#   00000010: 4f2d 3230 3031 0843 4152 4480 e004 1041  O-2001.CARD....A
#   00000020: 5050 524f 5645 44                        PPROVED
#
#   앞에서부터 한 바이트씩 읽습니다.
#     00              → 매직바이트. Confluent 와이어 포맷 버전. 현재 0x00 하나뿐.
#     00 00 00 07     → 스키마 ID = 7 (빅엔디언 int32)
#     0c              → Avro 문자열 길이 6 의 zigzag varint  ("O-2001")
#     08              → 길이 4                              ("CARD")
#     80 e0 04        → 39000 의 zigzag varint (3바이트)
#     10 + 8바이트     → 길이 8 + "APPROVED"
#
#   **필드 이름이 어디에도 없습니다.** 이것이 Avro 가 JSON 의 약 1/3.5 크기인
#   이유이자, 스키마 없이는 절대 못 읽는 이유입니다.
#   빅엔디언 4바이트를 셸에서 확인하려면:  printf '%d\n' 0x00000007  → 7

# (b) 읽어낸 ID 로 조회
curl -s "$SR_URL/schemas/ids/7"; echo

# 해설 (b):
#   {"schema":"{\"type\":\"record\",\"name\":\"Payment\",...}"}
#
#   ID 가 예제와 다르게 나오는 것이 정상입니다. 스키마 ID 는 **클러스터 전역 시퀀스**라
#   앞 실습에서 몇 개를 등록했느냐에 따라 달라집니다. 중요한 것은 숫자 자체가 아니라
#   "메시지 앞 5바이트만 보면 어떤 스키마인지 알 수 있다" 는 구조입니다.
#
#   운영에서라면: 이 5바이트 덕분에 **한 토픽에 여러 버전이 섞여 있어도** 각각
#   올바르게 읽힙니다. v1 로 쓴 옛 메시지와 v3 로 쓴 새 메시지가 나란히 있어도
#   컨슈머는 메시지마다 ID 를 보고 알맞은 writer 스키마를 골라 씁니다.
#   그래서 스키마 진화가 "토픽을 비우고 다시 시작" 없이 가능한 것입니다.


# ===========================================================================
# 정리
# ===========================================================================
echo "### 정리"
curl -s -X DELETE "$SR_URL/subjects/payments-value"                   >/dev/null 2>&1
curl -s -X DELETE "$SR_URL/subjects/payments-value?permanent=true"    >/dev/null 2>&1
curl -s -X DELETE "$SR_URL/subjects/s10_ex_avro-value"                >/dev/null 2>&1
curl -s -X DELETE "$SR_URL/subjects/s10_ex_avro-value?permanent=true" >/dev/null 2>&1
K kafka-topics.sh --bootstrap-server $BS --delete --topic s10_ex      2>/dev/null || true
K kafka-topics.sh --bootstrap-server $BS --delete --topic s10_ex_avro 2>/dev/null || true
K kafka-consumer-groups.sh --bootstrap-server $BS --delete --group s10-ex 2>/dev/null || true
echo "남은 subject:"; curl -s "$SR_URL/subjects"; echo
echo "=== Step 10 solution 완료 ==="