스키마 배포 후 전체 컨슈머가 죽었다

// v1 (기존, 등록됨)
{
  "type": "record", "name": "Order", "namespace": "com.example.orders",
  "fields": [
    {"name": "orderId",       "type": "string"},
    {"name": "memberId",      "type": "string"},
    {"name": "amount",        "type": "long"},
    {"name": "legacyCouponId","type": "string"}
  ]
}

// v2 (배포하려던 것)
{
  "type": "record", "name": "Order", "namespace": "com.example.orders",
  "fields": [
    {"name": "orderId",  "type": "string"},
    {"name": "memberId", "type": "string"},
    {"name": "amount",   "type": "string"},   // ← long 에서 string 으로 타입 변경
    {"name": "currency", "type": "string"}    // ← default 없는 신규 필드
  ]
}
// legacyCouponId 제거 (default 없는 필드였음)

$ curl -sX PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \
    --data '{"compatibility": "NONE"}' \
    http://schema-registry:8081/config/orders-value
{"compatibility":"NONE"}

$ # 등록이 통과됨
$ curl -sX POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
    --data @order-v2.json \
    http://schema-registry:8081/subjects/orders-value/versions
{"id":1418}

ERROR Application shutting down due to unhandled exception
org.apache.kafka.common.errors.RecordDeserializationException: Error deserializing VALUE for partition orders-11 at offset 88413902. If needed, please seek past the record to continue consumption.
	at org.apache.kafka.clients.consumer.internals.CompletedFetch.newRecordDeserializationException(CompletedFetch.java:...)
Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id 1418
Caused by: org.apache.avro.AvroTypeException: Found string, expecting long
	at org.apache.avro.io.ResolvingDecoder.doAction(ResolvingDecoder.java:...)

org.apache.kafka.common.KafkaException: Received exception when fetching the next record from orders-11. If needed, please seek past the record to continue consumption.

$ curl -sX POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
    --data @order-v2.json \
    http://schema-registry:8081/subjects/orders-value/versions -w '\nHTTP %{http_code}\n'

{"error_code":40901,"message":"Schema being registered is incompatible with an earlier schema for subject \"orders-value\", details: [{errorType:'TYPE_MISMATCH', description:'The type (path \"/fields/2/type\") is not compatible ...'}]"}
HTTP 409

org.apache.kafka.common.errors.SerializationException: Error registering Avro schema
Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Schema being registered is incompatible with an earlier schema for subject "orders-value", details: [...]; error code: 40901

# 전역 설정
$ curl -s http://schema-registry:8081/config
{"compatibilityLevel":"BACKWARD"}

# subject 설정 — 여기가 오버라이드되어 있었습니다
$ curl -s http://schema-registry:8081/config/orders-value
{"compatibilityLevel":"NONE"}

# subject 의 버전 목록과 최신 스키마
$ curl -s http://schema-registry:8081/subjects/orders-value/versions
[1,2,3,4,5]
$ curl -s http://schema-registry:8081/subjects/orders-value/versions/latest | jq -r '.id, .version'
1418
5

# 문제의 스키마 ID 로 직접 조회 (로그에 찍힌 id)
$ curl -s http://schema-registry:8081/schemas/ids/1418 | jq -r '.schema' | jq .

services:
  broker:
    image: apache/kafka:4.3.1
    hostname: broker
    container_name: broker
    ports:
      - '9092:9092'
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: 'broker,controller'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
      KAFKA_LISTENERS: 'CONTROLLER://:29093,PLAINTEXT://:19092,PLAINTEXT_HOST://:9092'
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://broker:19092,PLAINTEXT_HOST://localhost:9092'
      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@broker:29093'
      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
      KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
      CLUSTER_ID: '4L6g3nShT-eMCtK--X86sw'
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'

  schema-registry:
    image: confluentinc/cp-schema-registry:latest   # 사용 중인 CP 버전으로 고정하세요
    hostname: schema-registry
    container_name: schema-registry
    depends_on: [broker]
    ports:
      - '8081:8081'
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'PLAINTEXT://broker:19092'
      SCHEMA_REGISTRY_LISTENERS: 'http://0.0.0.0:8081'
      SCHEMA_REGISTRY_KAFKASTORE_TOPIC_REPLICATION_FACTOR: 1

docker compose up -d
SR=http://localhost:8081
CT='Content-Type: application/vnd.schemaregistry.v1+json'

# 0. 기본 호환성 모드 확인 → BACKWARD
curl -s $SR/config

# 1. v1 스키마 등록
cat > v1.json <<'EOF'
{"schema":"{\"type\":\"record\",\"name\":\"Order\",\"namespace\":\"com.example.orders\",\"fields\":[{\"name\":\"orderId\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"long\"},{\"name\":\"legacyCouponId\",\"type\":\"string\"}]}"}
EOF
curl -sX POST -H "$CT" --data @v1.json $SR/subjects/orders-value/versions

# 2. v2(타입 변경 + default 없는 필드 추가 + 필드 제거) 등록 시도 → 409
cat > v2.json <<'EOF'
{"schema":"{\"type\":\"record\",\"name\":\"Order\",\"namespace\":\"com.example.orders\",\"fields\":[{\"name\":\"orderId\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"string\"},{\"name\":\"currency\",\"type\":\"string\"}]}"}
EOF
curl -sX POST -H "$CT" --data @v2.json $SR/subjects/orders-value/versions -w '\nHTTP %{http_code}\n'
#   → {"error_code":40901,"message":"Schema being registered is incompatible with an earlier
#      schema for subject \"orders-value\", details: [...]"}
#     HTTP 409     ★ 이것이 안전장치입니다

# 3. 등록 전에 호환성만 검사하는 엔드포인트 (CI 에서 이걸 써야 합니다)
curl -sX POST -H "$CT" --data @v2.json \
  "$SR/compatibility/subjects/orders-value/versions/latest?verbose=true"
#   → {"is_compatible":false,"messages":[...]}

# 4. 사고 재현 — 모드를 NONE 으로 바꿔 우회
curl -sX PUT -H "$CT" --data '{"compatibility":"NONE"}' $SR/config/orders-value
curl -sX POST -H "$CT" --data @v2.json $SR/subjects/orders-value/versions
#   → {"id":N}  등록 성공. 이 순간 사고가 확정됩니다.

# 5. 컨슈머(v1 리더)가 v2 데이터를 읽으면 실패합니다.
#    kafka-avro-console-producer / consumer 로 확인할 수 있습니다.
docker exec -it schema-registry kafka-avro-console-producer \
  --bootstrap-server broker:19092 --topic orders \
  --property schema.registry.url=http://schema-registry:8081 \
  --property value.schema.id=N          # 4단계에서 받은 id
# {"orderId":"ORD-1","amount":"34900","currency":"KRW"}

# v1 스키마를 리더로 지정해 읽으면 역직렬화가 실패합니다
docker exec -it schema-registry kafka-avro-console-consumer \
  --bootstrap-server broker:19092 --topic orders --from-beginning \
  --property schema.registry.url=http://schema-registry:8081

# BACKWARD 에서는 "default 있는 필드 추가"가 통과합니다
curl -sX PUT -H "$CT" --data '{"compatibility":"BACKWARD"}' $SR/config/orders-value

cat > v1b.json <<'EOF'
{"schema":"{\"type\":\"record\",\"name\":\"Order\",\"namespace\":\"com.example.orders\",\"fields\":[{\"name\":\"orderId\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"long\"},{\"name\":\"legacyCouponId\",\"type\":\"string\"},{\"name\":\"currency\",\"type\":\"string\",\"default\":\"KRW\"}]}"}
EOF
curl -sX POST -H "$CT" --data @v1b.json \
  "$SR/compatibility/subjects/orders-value/versions/latest?verbose=true"
#   → {"is_compatible":true}

# FORWARD 로 바꾸면 "default 없는 필드 추가"도 통과합니다
curl -sX PUT -H "$CT" --data '{"compatibility":"FORWARD"}' $SR/config/orders-value
cat > v1c.json <<'EOF'
{"schema":"{\"type\":\"record\",\"name\":\"Order\",\"namespace\":\"com.example.orders\",\"fields\":[{\"name\":\"orderId\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"long\"},{\"name\":\"legacyCouponId\",\"type\":\"string\"},{\"name\":\"channel\",\"type\":\"string\"}]}"}
EOF
curl -sX POST -H "$CT" --data @v1c.json \
  "$SR/compatibility/subjects/orders-value/versions/latest?verbose=true"
#   → {"is_compatible":true}   ★ 같은 변경이 모드에 따라 결과가 다릅니다

# 원래대로 되돌리기 — subject 오버라이드 삭제
curl -sX DELETE $SR/config/orders-value
curl -s $SR/config/orders-value      # → 404 (전역 설정을 따름)

# 1. 프로듀서를 v1 으로 롤백한다 (가장 먼저. 나쁜 레코드 생성을 멈춥니다)
kubectl rollout undo deploy/order-producer

# 2. 호환성 모드를 되돌린다 (subject 오버라이드 제거 → 전역 BACKWARD 를 따름)
curl -sX DELETE http://schema-registry:8081/config/orders-value

# 3. 잘못 등록된 스키마 버전을 soft delete 한다
#    (hard delete 는 permanent=true. 이미 그 id 로 쓰인 레코드가 있으면 읽을 수 없게 되므로 주의)
curl -sX DELETE http://schema-registry:8081/subjects/orders-value/versions/5

# 4. 컨슈머가 나쁜 레코드를 건너뛰게 한다.
#    예외 메시지가 파티션과 오프셋을 알려 주므로 그 지점 다음으로 seek 합니다.
#    컨슈머를 내린 뒤 오프셋을 앞으로 밀어 줍니다.
kubectl scale deploy/settlement-service --replicas=0

kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 \
  --group order-settlement --topic orders:11 \
  --reset-offsets --shift-by 1 --dry-run     # 확인 후 --execute

kubectl scale deploy/settlement-service --replicas=8

#!/usr/bin/env bash
set -euo pipefail
SR="${SCHEMA_REGISTRY_URL}"
CT='Content-Type: application/vnd.schemaregistry.v1+json'

for f in schemas/*.avsc; do
  subject="$(basename "$f" .avsc)-value"
  payload=$(jq -Rs '{schema: .}' < "$f")

  # 1) 호환성 검사 — 등록하지 않고 결과만 받습니다
  res=$(curl -sf -X POST -H "$CT" --data "$payload" \
        "$SR/compatibility/subjects/$subject/versions/latest?verbose=true")
  ok=$(echo "$res" | jq -r '.is_compatible')

  if [ "$ok" != "true" ]; then
    echo "INCOMPATIBLE: $subject"
    echo "$res" | jq -r '.messages[]?'
    echo "→ 호환성 모드를 바꿔 우회하지 마세요. 스키마 변경을 쪼개세요."
    exit 1
  fi

  # 2) 통과했으면 등록
  curl -sf -X POST -H "$CT" --data "$payload" "$SR/subjects/$subject/versions" \
    | jq -r '"registered \($subject) id=\(.id)"'
done