#Step 12 — Kafka Connect

cd kafka/docker
mkdir -p connect-data
docker compose --profile connect up -d

docker logs -f kafka-connect 2>&1 | grep -E 'Kafka Connect started|Finished starting'

curl -s localhost:8083/ | jq

kt --list | grep '^connect-'

kt --describe --topic connect-configs

kt --describe --topic connect-offsets

kt --describe --topic connect-status

curl -s localhost:8083/connector-plugins | jq

curl -s localhost:8083/connectors | jq

curl -s -X POST localhost:8083/connectors \
  -H 'Content-Type: application/json' \
  -d '{ "name": "커넥터이름", "config": { ... } }' | jq

kt --create --topic s12_file --partitions 3 --replication-factor 3

cd kafka/docker
cat > connect-data/orders.txt <<'EOF'
{"order_id":"O-1001","customer_id":"C001","amount":39000,"status":"CREATED"}
{"order_id":"O-1002","customer_id":"C002","amount":12500,"status":"CREATED"}
{"order_id":"O-1003","customer_id":"C001","amount":88000,"status":"CREATED"}
EOF
docker exec kafka-connect cat /data/orders.txt

curl -s -X POST localhost:8083/connectors \
  -H 'Content-Type: application/json' -d '{
  "name": "file-source",
  "config": {
    "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
    "tasks.max": "1",
    "file": "/data/orders.txt",
    "topic": "s12_file",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.storage.StringConverter"
  }
}' | jq

curl -s localhost:8083/connectors/file-source/status | jq

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

echo '{"order_id":"O-1004","customer_id":"C003","amount":5000,"status":"CREATED"}' \
  >> connect-data/orders.txt
sleep 3
docker exec kafka-1 /opt/kafka/bin/kafka-get-offsets.sh \
  --bootstrap-server kafka-1:9092 --topic s12_file

curl -s localhost:8083/connectors/file-source/offsets | jq

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

curl -s -X POST localhost:8083/connectors \
  -H 'Content-Type: application/json' -d '{
  "name": "file-sink",
  "config": {
    "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
    "tasks.max": "1",
    "topics": "s12_file",
    "file": "/data/orders-out.txt",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.storage.StringConverter"
  }
}' | jq -r '.name + " → " + .type'

sleep 5
docker exec kafka-connect cat /data/orders-out.txt

kcg --list

kcg --describe --group connect-file-sink

curl -s -X DELETE localhost:8083/connectors/file-source -w '%{http_code}\n'
curl -s localhost:8083/connectors | jq

cat > connect-data/orders.txt <<'EOF'
{"order_id":"O-2001","customer_id":"C005","amount":77000,"status":"CREATED"}
{"order_id":"O-2002","customer_id":"C006","amount":31000,"status":"CREATED"}
EOF
wc -c connect-data/orders.txt

curl -s -X PUT localhost:8083/connectors/file-source/config \
  -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
  "tasks.max": "1",
  "file": "/data/orders.txt",
  "topic": "s12_file",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.storage.StringConverter"
}' | jq -r '.name'

sleep 5
docker exec kafka-1 /opt/kafka/bin/kafka-get-offsets.sh \
  --bootstrap-server kafka-1:9092 --topic s12_file

curl -s localhost:8083/connectors/file-source/offsets | jq

curl -s -X PUT localhost:8083/connectors/file-source/stop -w '%{http_code}\n'
sleep 2
curl -s -X DELETE localhost:8083/connectors/file-source/offsets | jq

curl -s -X PUT localhost:8083/connectors/file-source/resume -w '%{http_code}\n'
sleep 5
docker exec kafka-1 /opt/kafka/bin/kafka-get-offsets.sh \
  --bootstrap-server kafka-1:9092 --topic s12_file

curl -s -X PUT localhost:8083/connectors/file-sink/config \
  -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
  "tasks.max": "1",
  "topics": "s12_file",
  "file": "/proc/nope/orders-out.txt",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.storage.StringConverter"
}' >/dev/null

sleep 5
curl -s localhost:8083/connectors/file-sink/status | jq

curl -s localhost:8083/connectors/file-sink/status | jq -r '.tasks[0].trace' | head -3

curl -s -X PUT localhost:8083/connectors/file-sink/config \
  -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
  "tasks.max": "1",
  "topics": "s12_file",
  "file": "/data/orders-out.txt",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.storage.StringConverter"
}' >/dev/null

curl -s -X POST 'localhost:8083/connectors/file-sink/restart?includeTasks=true&onlyFailed=true' -w '%{http_code}\n'
sleep 5
curl -s localhost:8083/connectors/file-sink/status | jq -r '[.connector.state] + [.tasks[].state] | join(",")'

kt --create --topic s12_smt.s12_file --partitions 3 --replication-factor 3

curl -s -X PUT localhost:8083/connectors/file-source-smt/config \
  -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
  "tasks.max": "1",
  "file": "/data/orders-smt.txt",
  "topic": "s12_file",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter.schemas.enable": "false",

  "transforms": "parse,mkKey,extractKey,addTs,route",

  "transforms.parse.type": "org.apache.kafka.connect.transforms.HoistField$Value",
  "transforms.parse.field": "line",

  "transforms.mkKey.type": "org.apache.kafka.connect.transforms.ValueToKey",
  "transforms.mkKey.fields": "line",

  "transforms.extractKey.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
  "transforms.extractKey.field": "line",

  "transforms.addTs.type": "org.apache.kafka.connect.transforms.InsertField$Value",
  "transforms.addTs.timestamp.field": "ingested_at",

  "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.route.regex": "(.*)",
  "transforms.route.replacement": "s12_smt.$1"
}' | jq -r '.name'

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

kt --create --topic s12_dlq_src --partitions 1 --replication-factor 3
kt --create --topic s12_dlq --partitions 1 --replication-factor 3

docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka-1:9092 --topic s12_dlq_src <<'EOF'
{"order_id":"O-4001","amount":10000}
NOT-A-JSON
{"order_id":"O-4002","amount":20000}
EOF

curl -s -X PUT localhost:8083/connectors/dlq-sink/config \
  -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
  "tasks.max": "1",
  "topics": "s12_dlq_src",
  "file": "/data/dlq-out.txt",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter.schemas.enable": "false"
}' >/dev/null
sleep 6
curl -s localhost:8083/connectors/dlq-sink/status | jq -r '[.connector.state] + [.tasks[].state] | join(",")'

curl -s localhost:8083/connectors/dlq-sink/status | jq -r '.tasks[0].trace' | head -4

curl -s -X PUT localhost:8083/connectors/dlq-sink/config \
  -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
  "tasks.max": "1",
  "topics": "s12_dlq_src",
  "file": "/data/dlq-out.txt",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter.schemas.enable": "false",

  "errors.tolerance": "all",
  "errors.log.enable": "true",
  "errors.log.include.messages": "true",
  "errors.deadletterqueue.topic.name": "s12_dlq",
  "errors.deadletterqueue.topic.replication.factor": "3",
  "errors.deadletterqueue.context.headers.enable": "true"
}' >/dev/null

curl -s -X POST 'localhost:8083/connectors/dlq-sink/restart?includeTasks=true' -w '%{http_code}\n'
sleep 6
curl -s localhost:8083/connectors/dlq-sink/status | jq -r '[.connector.state] + [.tasks[].state] | join(",")'

docker exec kafka-connect cat /data/dlq-out.txt

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

docker exec kafka-connect confluent-hub install --no-prompt \
  confluentinc/kafka-connect-jdbc:10.7.6
docker exec kafka-connect bash -c 'curl -sL -o /usr/share/confluent-hub-components/confluentinc-kafka-connect-jdbc/lib/mysql-connector-j-8.3.0.jar \
  https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.3.0/mysql-connector-j-8.3.0.jar'
docker restart kafka-connect

for c in $(curl -s localhost:8083/connectors | jq -r '.[]'); do
  curl -s -X DELETE localhost:8083/connectors/$c -w "$c deleted %{http_code}\n"
done

kt --list | grep -E '^s12_' | xargs -I{} docker exec kafka-1 \
  /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka-1:9092 --delete --topic {}
kt --list | grep -E '^s12_' || echo "s12_ 토픽 없음"

kcg --list

kcg --delete --group connect-file-sink --group connect-dlq-sink
kcg --list || echo "그룹 없음"

# Connect 를 완전히 초기화하고 싶을 때만
docker compose stop kafka-connect
kt --delete --topic connect-configs
kt --delete --topic connect-offsets
kt --delete --topic connect-status
docker compose --profile connect up -d

docker compose stop kafka-connect

#!/usr/bin/env bash
set -euo pipefail
# ============================================================================
# Step 12 — Kafka Connect / practice.sh
#
# 실행법:
#   cd kafka/docker && mkdir -p connect-data && docker compose --profile connect up -d
#   bash ../step-12-connect/practice.sh
#
#   한 단계씩 보려면:  bash -x practice.sh
#
# 요구사항: jq, curl
#   macOS:  brew install jq
#   Debian: apt-get install -y jq
#
# 이 스크립트는 kafka/docker 디렉터리를 기준으로 connect-data/ 에 파일을 씁니다.
# 스크립트 어디서 실행해도 되도록 DOCKER_DIR 을 자동으로 찾습니다.
# ============================================================================

BS=kafka-1:9092
CONNECT=localhost:8083
DOCKER_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../docker" && pwd)"
DATA_DIR="$DOCKER_DIR/connect-data"
mkdir -p "$DATA_DIR"

# 컨테이너 안 CLI 헬퍼
K()  { docker exec kafka-1 /opt/kafka/bin/"$@"; }
KT() { K kafka-topics.sh --bootstrap-server "$BS" "$@"; }
KCG(){ K kafka-consumer-groups.sh --bootstrap-server "$BS" "$@"; }
KGO(){ K kafka-get-offsets.sh --bootstrap-server "$BS" "$@"; }

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

# 커넥터 + 모든 태스크 상태를 한 줄로. 함정 C 대응 헬퍼.
all_states() {
  curl -s "$CONNECT/connectors/$1/status" \
    | jq -r '[.connector.state] + [.tasks[].state] | join(",")'
}

# ----------------------------------------------------------------------------
# [12-0] 실습 준비 — Connect 워커가 뜰 때까지 대기
#   워커 기동에 40~70초가 걸립니다. 이 대기 없이 바로 등록하면 Connection refused.
# ----------------------------------------------------------------------------
wait_connect() {
  echo "Connect 워커 대기 중..."
  for i in $(seq 1 60); do
    if curl -sf "$CONNECT/" >/dev/null 2>&1; then
      echo "  → 준비 완료 (${i}회 시도)"
      return 0
    fi
    sleep 2
  done
  echo "Connect 워커가 뜨지 않았습니다. docker logs kafka-connect 를 확인하세요." >&2
  exit 1
}

hr "[12-0] 실습 준비"
wait_connect
curl -s "$CONNECT/" | jq
# kafka_cluster_id 가 MkU3OEVBNTcwNTJENDM2Qk 여야 우리 클러스터에 붙은 것입니다.

# ----------------------------------------------------------------------------
# [12-3] 내부 토픽 3종
# ----------------------------------------------------------------------------
hr "[12-3] 내부 토픽 3종"
KT --list | grep '^connect-' || true
echo "--- connect-configs (파티션 1 강제, compact) ---"
KT --describe --topic connect-configs
echo "--- connect-offsets (25 파티션, compact) ---"
KT --describe --topic connect-offsets | head -4
echo "--- connect-status (5 파티션, compact) ---"
KT --describe --topic connect-status | head -3

# ----------------------------------------------------------------------------
# [12-4] REST API — 플러그인 목록
#   JDBC 가 없다는 것을 확인합니다. 기본 이미지에는 FileStream + MirrorMaker2 뿐.
# ----------------------------------------------------------------------------
hr "[12-4] 설치된 커넥터 플러그인"
curl -s "$CONNECT/connector-plugins" | jq -r '.[] | "\(.type)\t\(.class)"'
echo "--- 현재 커넥터 목록 ---"
curl -s "$CONNECT/connectors" | jq

# ----------------------------------------------------------------------------
# [12-5] FileStreamSource — 파일을 토픽으로
# ----------------------------------------------------------------------------
hr "[12-5] FileStreamSource"
KT --create --topic s12_file --partitions 3 --replication-factor 3 --if-not-exists

cat > "$DATA_DIR/orders.txt" <<'EOF'
{"order_id":"O-1001","customer_id":"C001","amount":39000,"status":"CREATED"}
{"order_id":"O-1002","customer_id":"C002","amount":12500,"status":"CREATED"}
{"order_id":"O-1003","customer_id":"C001","amount":88000,"status":"CREATED"}
EOF
docker exec kafka-connect cat /data/orders.txt

curl -s -X POST "$CONNECT/connectors" -H 'Content-Type: application/json' -d '{
  "name": "file-source",
  "config": {
    "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
    "tasks.max": "1",
    "file": "/data/orders.txt",
    "topic": "s12_file",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.storage.StringConverter"
  }
}' | jq

# 등록 응답의 "tasks": [] 는 정상입니다. 태스크 배정 전에 응답이 돌아옵니다.
sleep 4
echo "--- status ---"
curl -s "$CONNECT/connectors/file-source/status" | jq

echo "--- s12_file 내용 ---"
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic s12_file \
  --from-beginning --property print.offset=true --timeout-ms 5000 || true

# 파일에 한 줄 추가 → tail -f 처럼 따라옵니다
echo '{"order_id":"O-1004","customer_id":"C003","amount":5000,"status":"CREATED"}' >> "$DATA_DIR/orders.txt"
sleep 4
echo "--- 오프셋 (키가 없으므로 전부 파티션 0 에 몰립니다) ---"
KGO --topic s12_file

echo "--- 소스 오프셋 (position 은 '바이트' 위치입니다) ---"
curl -s "$CONNECT/connectors/file-source/offsets" | jq

echo "--- connect-offsets 토픽 원본 ---"
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic connect-offsets \
  --from-beginning --property print.key=true --timeout-ms 5000 || true

# ----------------------------------------------------------------------------
# [12-6] FileStreamSink — 토픽을 다시 파일로
#   소스는 "topic"(단수), 싱크는 "topics"(복수). 한 글자 차이입니다.
# ----------------------------------------------------------------------------
hr "[12-6] FileStreamSink"
curl -s -X POST "$CONNECT/connectors" -H 'Content-Type: application/json' -d '{
  "name": "file-sink",
  "config": {
    "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
    "tasks.max": "1",
    "topics": "s12_file",
    "file": "/data/orders-out.txt",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.storage.StringConverter"
  }
}' | jq -r '.name + " → " + .type'
sleep 6
docker exec kafka-connect cat /data/orders-out.txt

# ----------------------------------------------------------------------------
# [12-7-A] 함정 A — 소스와 싱크의 오프셋 저장 위치가 다릅니다
#   싱크만 컨슈머 그룹(connect-<이름>)을 만듭니다. 소스는 안 만듭니다.
# ----------------------------------------------------------------------------
hr "[12-7-A] 함정 A — 오프셋 저장 위치"
echo "--- 컨슈머 그룹 목록 (connect-file-sink 만 있고 connect-file-source 는 없음) ---"
KCG --list
echo "--- 싱크는 완전히 평범한 컨슈머 그룹입니다 ---"
KCG --describe --group connect-file-sink

# ----------------------------------------------------------------------------
# [12-7-B] 함정 B — 커넥터를 삭제해도 오프셋이 남습니다
#   ★ 이 구간은 순서를 건너뛰지 마세요. 중간을 생략하면 함정이 재현되지 않습니다.
# ----------------------------------------------------------------------------
hr "[12-7-B] 함정 B — 삭제해도 남는 오프셋"

echo "--- ① 커넥터 삭제 ---"
curl -s -X DELETE "$CONNECT/connectors/file-source" -w 'HTTP %{http_code}\n'
sleep 2
curl -s "$CONNECT/connectors" | jq

echo "--- ② 파일을 완전히 새로 씁니다 (152바이트, 2줄) ---"
cat > "$DATA_DIR/orders.txt" <<'EOF'
{"order_id":"O-2001","customer_id":"C005","amount":77000,"status":"CREATED"}
{"order_id":"O-2002","customer_id":"C006","amount":31000,"status":"CREATED"}
EOF
wc -c "$DATA_DIR/orders.txt"

echo "--- ③ 같은 이름으로 재생성 (PUT /config 는 멱등) ---"
curl -s -X PUT "$CONNECT/connectors/file-source/config" -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
  "tasks.max": "1",
  "file": "/data/orders.txt",
  "topic": "s12_file",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.storage.StringConverter"
}' | jq -r '.name'
sleep 6

echo "--- ④ 오프셋이 4 에서 안 늘어납니다. 에러도 안 납니다. ---"
KGO --topic s12_file
all_states file-source     # RUNNING,RUNNING — 멀쩡합니다

echo "--- ⑤ 범인: position 304 가 남아 있습니다 (새 파일은 152바이트뿐) ---"
curl -s "$CONNECT/connectors/file-source/offsets" | jq

echo "--- ⑥ 해결: stop → DELETE /offsets → resume ---"
curl -s -X PUT "$CONNECT/connectors/file-source/stop" -w 'HTTP %{http_code}\n'
sleep 3
# DELETE /offsets 는 커넥터가 STOPPED 여야 합니다. RUNNING 이면 409 Conflict.
curl -s -X DELETE "$CONNECT/connectors/file-source/offsets" | jq
curl -s -X PUT "$CONNECT/connectors/file-source/resume" -w 'HTTP %{http_code}\n'
sleep 6

echo "--- ⑦ 4 → 6. 새 파일의 2줄이 들어왔습니다. ---"
KGO --topic s12_file

# 대안: 커넥터 이름을 file-source-v2 로 바꾸면 오프셋 키가 달라져 처음부터 읽습니다.
#       운영에서는 이 방법(이름 버저닝)이 더 안전합니다. 되돌리기가 쉽습니다.

# ----------------------------------------------------------------------------
# [12-7-C] 함정 C — 커넥터는 RUNNING, 태스크는 FAILED
#   /proc 아래는 컨테이너 안에서도 새 파일을 못 만듭니다. 확실히 실패합니다.
# ----------------------------------------------------------------------------
hr "[12-7-C] 함정 C — 태스크만 죽는다"

curl -s -X PUT "$CONNECT/connectors/file-sink/config" -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
  "tasks.max": "1",
  "topics": "s12_file",
  "file": "/proc/nope/orders-out.txt",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.storage.StringConverter"
}' >/dev/null
sleep 6

echo "--- .connector.state 만 보면 (이것만 보는 모니터링은 장애를 놓칩니다) ---"
curl -s "$CONNECT/connectors/file-sink/status" | jq -r '.connector.state'
echo "--- 태스크까지 보면 ---"
all_states file-sink       # RUNNING,FAILED

echo "--- trace 필드 (스택트레이스 전문이 들어옵니다) ---"
curl -s "$CONNECT/connectors/file-sink/status" | jq -r '.tasks[0].trace' | head -4

echo "--- 운영용: 전 커넥터 일괄 점검 ---"
for c in $(curl -s "$CONNECT/connectors" | jq -r '.[]'); do
  s=$(curl -s "$CONNECT/connectors/$c/status")
  echo "$c $(echo "$s" | jq -r '.connector.state') tasks=$(echo "$s" | jq -r '[.tasks[].state] | join(",")')"
done

echo "--- 고치고 재시작. 설정만 고쳐서는 죽은 태스크가 안 살아납니다. ---"
curl -s -X PUT "$CONNECT/connectors/file-sink/config" -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
  "tasks.max": "1",
  "topics": "s12_file",
  "file": "/data/orders-out.txt",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.storage.StringConverter"
}' >/dev/null
# includeTasks=true 가 없으면 커넥터만 재시작하고 태스크는 FAILED 로 남습니다.
curl -s -X POST "$CONNECT/connectors/file-sink/restart?includeTasks=true&onlyFailed=true" -w 'HTTP %{http_code}\n'
sleep 6
all_states file-sink       # RUNNING,RUNNING

# ----------------------------------------------------------------------------
# [12-4b] pause / resume / config 조회
# ----------------------------------------------------------------------------
hr "[12-4b] pause / resume"
curl -s -X PUT "$CONNECT/connectors/file-sink/pause" -w 'pause HTTP %{http_code}\n'
sleep 3
curl -s "$CONNECT/connectors/file-sink/status" | jq -r '.connector.state'   # PAUSED
curl -s -X PUT "$CONNECT/connectors/file-sink/resume" -w 'resume HTTP %{http_code}\n'
sleep 3
all_states file-sink

echo "--- 설정만 조회 ---"
curl -s "$CONNECT/connectors/file-sink/config" | jq

# ----------------------------------------------------------------------------
# [12-8] SMT — 키 승격 + 타임스탬프 + 라우팅
#   transforms 의 쉼표 순서가 곧 적용 순서입니다.
#   맨 앞 HoistField$Value 를 빼면 "Only Struct objects supported" 로 죽습니다.
# ----------------------------------------------------------------------------
hr "[12-8] SMT 체인"
KT --create --topic "s12_smt.s12_file" --partitions 3 --replication-factor 3 --if-not-exists

cat > "$DATA_DIR/orders-smt.txt" <<'EOF'
{"order_id":"O-3001","customer_id":"C007","amount":21000,"status":"CREATED"}
{"order_id":"O-3002","customer_id":"C008","amount":45000,"status":"CREATED"}
EOF

curl -s -X PUT "$CONNECT/connectors/file-source-smt/config" -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
  "tasks.max": "1",
  "file": "/data/orders-smt.txt",
  "topic": "s12_file",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter.schemas.enable": "false",

  "transforms": "parse,mkKey,extractKey,addTs,route",

  "transforms.parse.type": "org.apache.kafka.connect.transforms.HoistField$Value",
  "transforms.parse.field": "line",

  "transforms.mkKey.type": "org.apache.kafka.connect.transforms.ValueToKey",
  "transforms.mkKey.fields": "line",

  "transforms.extractKey.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
  "transforms.extractKey.field": "line",

  "transforms.addTs.type": "org.apache.kafka.connect.transforms.InsertField$Value",
  "transforms.addTs.timestamp.field": "ingested_at",

  "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.route.regex": "(.*)",
  "transforms.route.replacement": "s12_smt.$1"
}' | jq -r '.name'
sleep 7

echo "--- 변환 후: 키가 생겼고, 파티션이 분산되고, ingested_at 이 붙었습니다 ---"
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic "s12_smt.s12_file" \
  --from-beginning --property print.key=true --property print.partition=true \
  --timeout-ms 5000 || true

# ----------------------------------------------------------------------------
# [12-9] 에러 처리와 DLQ
#   일부러 두 번 설정합니다. 먼저 DLQ 없이(태스크 죽음), 그다음 DLQ 붙여서(통과).
# ----------------------------------------------------------------------------
hr "[12-9] DLQ"
KT --create --topic s12_dlq_src --partitions 1 --replication-factor 3 --if-not-exists
KT --create --topic s12_dlq     --partitions 1 --replication-factor 3 --if-not-exists

docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server "$BS" --topic s12_dlq_src <<'EOF'
{"order_id":"O-4001","amount":10000}
NOT-A-JSON
{"order_id":"O-4002","amount":20000}
EOF

echo "--- ① DLQ 없이: 한 건 때문에 전체가 멈춥니다 ---"
curl -s -X PUT "$CONNECT/connectors/dlq-sink/config" -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
  "tasks.max": "1",
  "topics": "s12_dlq_src",
  "file": "/data/dlq-out.txt",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter.schemas.enable": "false"
}' >/dev/null
sleep 8
all_states dlq-sink        # RUNNING,FAILED
curl -s "$CONNECT/connectors/dlq-sink/status" | jq -r '.tasks[0].trace' | head -4

echo "--- ② DLQ 를 붙입니다 ---"
curl -s -X PUT "$CONNECT/connectors/dlq-sink/config" -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
  "tasks.max": "1",
  "topics": "s12_dlq_src",
  "file": "/data/dlq-out.txt",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter.schemas.enable": "false",

  "errors.tolerance": "all",
  "errors.log.enable": "true",
  "errors.log.include.messages": "true",
  "errors.deadletterqueue.topic.name": "s12_dlq",
  "errors.deadletterqueue.topic.replication.factor": "3",
  "errors.deadletterqueue.context.headers.enable": "true"
}' >/dev/null
curl -s -X POST "$CONNECT/connectors/dlq-sink/restart?includeTasks=true" -w 'HTTP %{http_code}\n'
sleep 8
all_states dlq-sink        # RUNNING,RUNNING

echo "--- 정상 2건은 통과했습니다 ---"
docker exec kafka-connect cat /data/dlq-out.txt

echo "--- 깨진 1건은 DLQ 로. 헤더에 원인이 붙습니다. ---"
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic s12_dlq \
  --from-beginning --property print.headers=true --timeout-ms 5000 || true

# ⚠️ errors.tolerance=all 을 DLQ 없이 쓰면 실패 레코드가 어디에도 안 남고 사라집니다.
#    커넥터는 RUNNING, 랙은 0, 알람은 안 울립니다. 반드시 DLQ 와 함께 쓰세요.

# ----------------------------------------------------------------------------
# [12-10] JDBC — 플러그인 설치가 필요하므로 기본 실행에서 제외했습니다.
#         해 보고 싶으면 아래 주석을 푸세요. 수백 MB 를 내려받습니다.
# ----------------------------------------------------------------------------
# docker exec kafka-connect confluent-hub install --no-prompt \
#   confluentinc/kafka-connect-jdbc:10.7.6
# docker restart kafka-connect && wait_connect
# curl -s $CONNECT/connector-plugins | jq -r '.[].class' | grep -i jdbc
#
# JDBC 소스 mode 4종:
#   bulk                    매번 전체 재전송
#   incrementing            INSERT 만. UPDATE/DELETE 못 잡음
#   timestamp               WHERE ts > ?  ← 동일 타임스탬프 행을 건너뜁니다 (함정)
#   timestamp+incrementing  WHERE ts > ? OR (ts = ? AND id > ?)  ← 정답
# 어느 모드든 DELETE 는 못 잡습니다. 필요하면 소프트 삭제 또는 CDC(Debezium).

# ----------------------------------------------------------------------------
# [12-11] 정리
# ----------------------------------------------------------------------------
hr "[12-11] 정리"

echo "--- 커넥터 전부 삭제 ---"
for c in $(curl -s "$CONNECT/connectors" | jq -r '.[]'); do
  curl -s -X DELETE "$CONNECT/connectors/$c" -w "$c deleted HTTP %{http_code}\n"
done
sleep 3

echo "--- s12_ 토픽 삭제 ---"
for t in $(KT --list | grep -E '^s12_' || true); do
  KT --delete --topic "$t" && echo "deleted $t"
done
KT --list | grep -E '^s12_' || echo "s12_ 토픽 없음"

echo "--- 컨슈머 그룹 삭제 (커넥터를 지워도 그룹은 남습니다 — 함정 A 의 연장) ---"
KCG --list
for g in $(KCG --list | grep -E '^connect-' || true); do
  KCG --delete --group "$g" || true
done
KCG --list

echo "--- 실습 파일 정리 ---"
rm -f "$DATA_DIR/orders.txt" "$DATA_DIR/orders-out.txt" \
      "$DATA_DIR/orders-smt.txt" "$DATA_DIR/dlq-out.txt"

# 내부 토픽 3종(connect-configs/offsets/status)은 지우지 않습니다.
# 완전 초기화가 필요할 때만:
#   docker compose stop kafka-connect
#   kt --delete --topic connect-configs
#   kt --delete --topic connect-offsets
#   kt --delete --topic connect-status
#   docker compose --profile connect up -d

hr "practice.sh 완료"
echo "Step 13 은 Connect 가 필요 없습니다. 메모리가 부족하면:"
echo "  docker compose stop kafka-connect"

#!/usr/bin/env bash
set -euo pipefail
# ============================================================================
# Step 12 — Kafka Connect / exercise.sh   (문제 6개)
#
# 실행법:
#   cd kafka/docker && docker compose --profile connect up -d
#   bash ../step-12-connect/exercise.sh
#
# 각 문제의 "# 여기에 작성:" 아래를 직접 채우세요.
# 정답은 solution.sh 에 있습니다. 먼저 풀어 본 뒤에 여세요.
#
# 문제 1·5·6 = 커넥터를 직접 설계하는 문제
# 문제 2·3·4 = 이미 만들어진 상태를 관찰·진단하는 문제
# ============================================================================

BS=kafka-1:9092
CONNECT=localhost:8083
DOCKER_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../docker" && pwd)"
DATA_DIR="$DOCKER_DIR/connect-data"
mkdir -p "$DATA_DIR"

K()  { docker exec kafka-1 /opt/kafka/bin/"$@"; }
KT() { K kafka-topics.sh --bootstrap-server "$BS" "$@"; }
KCG(){ K kafka-consumer-groups.sh --bootstrap-server "$BS" "$@"; }
KGO(){ K kafka-get-offsets.sh --bootstrap-server "$BS" "$@"; }
hr() { echo; echo "------------------------------------------------------------"; echo "  $*"; echo "------------------------------------------------------------"; }

for i in $(seq 1 60); do curl -sf "$CONNECT/" >/dev/null 2>&1 && break; sleep 2; done


# ============================================================================
# 문제 1. FileStreamSource 커넥터 등록 + 상태 한 줄 확인
#
#   /data/q1.txt 를 읽어 s12ex_q1 토픽으로 보내는 커넥터 "q1-source" 를 만드세요.
#   그다음, 커넥터 상태와 "모든 태스크의 상태"를 쉼표로 이어 한 줄로 출력하세요.
#   기대 출력: RUNNING,RUNNING
#
#   힌트: jq 로 두 배열을 이어 붙입니다.
# ============================================================================
hr "문제 1"
KT --create --topic s12ex_q1 --partitions 3 --replication-factor 3 --if-not-exists
cat > "$DATA_DIR/q1.txt" <<'EOF'
{"order_id":"O-9001","customer_id":"C001","amount":11000}
{"order_id":"O-9002","customer_id":"C002","amount":22000}
EOF

# 여기에 작성: q1-source 커넥터를 등록하는 curl 명령


# 여기에 작성: 커넥터 + 모든 태스크 상태를 "RUNNING,RUNNING" 형태로 출력하는 명령



# ============================================================================
# 문제 2. "커넥터를 다시 만들었는데 데이터가 안 들어옵니다"
#
#   아래 준비 블록이 q2-source 커넥터를 만들고 → 데이터를 흘리고 → 삭제하고
#   → 파일을 완전히 새 내용으로 갈아엎고 → 같은 이름으로 재생성합니다.
#   그런데 새 파일의 3줄이 토픽에 들어오지 않습니다.
#
#   (a) 원인을 확인하는 명령 한 줄
#   (b) 고치는 명령 세 줄
#   (c) 고쳐졌는지 검증하는 명령 한 줄
#   을 쓰세요.
# ============================================================================
hr "문제 2 — 준비 (여기는 손대지 마세요)"
KT --create --topic s12ex_q2 --partitions 1 --replication-factor 3 --if-not-exists
printf '%s\n' 'A-1' 'A-2' 'A-3' 'A-4' 'A-5' > "$DATA_DIR/q2.txt"
curl -s -X PUT "$CONNECT/connectors/q2-source/config" -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
  "tasks.max": "1", "file": "/data/q2.txt", "topic": "s12ex_q2",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.storage.StringConverter"
}' >/dev/null
sleep 6
echo "5건 들어왔는지:"; KGO --topic s12ex_q2
curl -s -X DELETE "$CONNECT/connectors/q2-source" >/dev/null; sleep 3
printf '%s\n' 'B-1' 'B-2' 'B-3' > "$DATA_DIR/q2.txt"      # 파일을 통째로 교체
curl -s -X PUT "$CONNECT/connectors/q2-source/config" -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
  "tasks.max": "1", "file": "/data/q2.txt", "topic": "s12ex_q2",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.storage.StringConverter"
}' >/dev/null
sleep 6
echo "재생성 후 (여전히 5? 8이 되어야 정상 아닙니까?):"; KGO --topic s12ex_q2
echo "커넥터 상태 (멀쩡합니다. 에러가 없습니다.):"
curl -s "$CONNECT/connectors/q2-source/status" | jq -r '[.connector.state] + [.tasks[].state] | join(",")'

hr "문제 2 — 여기부터 여러분 차례"

# 여기에 작성: (a) 원인을 확인하는 명령


# 여기에 작성: (b) 고치는 명령 세 줄


# 여기에 작성: (c) 검증 — s12ex_q2 오프셋이 8이 되어야 합니다



# ============================================================================
# 문제 3. connect-configs 토픽의 cleanup.policy 와 파티션 수
#
#   (a) 두 값을 확인하는 명령을 쓰세요.
#   (b) 그 값이어야 하는 이유를 각각 한 문장으로 주석에 쓰세요.
#       - cleanup.policy 가 delete 라면 무슨 일이 생깁니까?
#       - 파티션이 3개라면 무슨 일이 생깁니까?
# ============================================================================
hr "문제 3"

# 여기에 작성: (a) 확인 명령


# 여기에 작성: (b) 이유 두 줄 (주석으로)
#   cleanup.policy=compact 여야 하는 이유:
#   PartitionCount=1 이어야 하는 이유:



# ============================================================================
# 문제 4. 커넥터가 얼마나 밀렸는지 확인하기
#
#   아래 준비 블록이 q4-sink(싱크)와 q4-source(소스)를 만듭니다.
#   각각 "얼마나 밀렸는지"를 확인하는 명령을 쓰세요.
#   두 명령은 완전히 다릅니다. 왜 다른지도 주석으로 쓰세요.
# ============================================================================
hr "문제 4 — 준비"
KT --create --topic s12ex_q4 --partitions 3 --replication-factor 3 --if-not-exists
printf '%s\n' 'X-1' 'X-2' 'X-3' 'X-4' > "$DATA_DIR/q4.txt"
curl -s -X PUT "$CONNECT/connectors/q4-source/config" -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
  "tasks.max": "1", "file": "/data/q4.txt", "topic": "s12ex_q4",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.storage.StringConverter"
}' >/dev/null
curl -s -X PUT "$CONNECT/connectors/q4-sink/config" -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
  "tasks.max": "1", "topics": "s12ex_q4", "file": "/data/q4-out.txt",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.storage.StringConverter"
}' >/dev/null
sleep 8

hr "문제 4 — 여기부터"

# 여기에 작성: (a) 싱크 커넥터 q4-sink 가 얼마나 밀렸는지


# 여기에 작성: (b) 소스 커넥터 q4-source 는 어디까지 읽었는지


# 여기에 작성: (c) 왜 두 명령이 다릅니까? (주석)



# ============================================================================
# 문제 5. SMT 로 키를 만들어 파티션 분산시키기
#
#   /data/q5.txt 를 s12ex_q5(3파티션)로 보내되, SMT 로 값을 키로 승격시켜
#   메시지가 여러 파티션에 흩어지게 만드세요.
#
#   정답 조건: kafka-get-offsets.sh 출력의 세 줄이 전부 0 이 아니어야 합니다.
#   힌트: FileStreamSource 는 값을 "문자열"로 내보냅니다. ValueToKey 를 바로 걸면
#         DataException 이 납니다. 한 단계가 더 필요합니다.
# ============================================================================
hr "문제 5"
KT --create --topic s12ex_q5 --partitions 3 --replication-factor 3 --if-not-exists
cat > "$DATA_DIR/q5.txt" <<'EOF'
C001
C002
C003
C001
C002
C003
EOF

# 여기에 작성: q5-source 커넥터 (SMT 포함)


# 여기에 작성: 파티션이 분산됐는지 확인



# ============================================================================
# 문제 6. DLQ 로 깨진 레코드를 격리하고 원본 위치를 추적하기
#
#   s12ex_q6_src 에는 정상 JSON 2건과 깨진 값 2건이 들어 있습니다.
#   - 정상 2건은 /data/q6-out.txt 로 나가고
#   - 깨진 2건은 s12ex_q6_dlq 토픽으로 가고
#   - DLQ 메시지의 헤더로 원본 토픽/파티션/오프셋을 알 수 있게
#   커넥터 "q6-sink" 를 만드세요.
# ============================================================================
hr "문제 6"
KT --create --topic s12ex_q6_src --partitions 1 --replication-factor 3 --if-not-exists
KT --create --topic s12ex_q6_dlq --partitions 1 --replication-factor 3 --if-not-exists
docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server "$BS" --topic s12ex_q6_src <<'EOF'
{"id":"A","v":1}
OOPS
{"id":"B","v":2}
{"unclosed":
EOF

# 여기에 작성: q6-sink 커넥터


# 여기에 작성: 정상 2건이 파일로 나갔는지 확인


# 여기에 작성: DLQ 2건과 헤더 확인



# ============================================================================
# 정리 — 문제를 안 풀고 정리만 돌려도 에러 없이 끝납니다.
# ============================================================================
hr "정리"
for c in q1-source q2-source q4-source q4-sink q5-source q6-sink; do
  curl -s -X DELETE "$CONNECT/connectors/$c" >/dev/null 2>&1 || true
done
sleep 3
for t in $(KT --list | grep -E '^s12ex_' || true); do KT --delete --topic "$t" || true; done
for g in $(KCG --list | grep -E '^connect-q' || true); do KCG --delete --group "$g" || true; done
rm -f "$DATA_DIR"/q[0-9]*.txt "$DATA_DIR"/q[0-9]*-out.txt
echo "정리 완료"

#!/usr/bin/env bash
set -euo pipefail
# ============================================================================
# Step 12 — Kafka Connect / solution.sh   (정답 + 해설)
#
# 실행법:
#   cd kafka/docker && docker compose --profile connect up -d
#   bash ../step-12-connect/solution.sh
#
# exercise.sh 를 먼저 풀어 본 뒤에 여세요.
# ============================================================================

BS=kafka-1:9092
CONNECT=localhost:8083
DOCKER_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../docker" && pwd)"
DATA_DIR="$DOCKER_DIR/connect-data"
mkdir -p "$DATA_DIR"

K()  { docker exec kafka-1 /opt/kafka/bin/"$@"; }
KT() { K kafka-topics.sh --bootstrap-server "$BS" "$@"; }
KCG(){ K kafka-consumer-groups.sh --bootstrap-server "$BS" "$@"; }
KGO(){ K kafka-get-offsets.sh --bootstrap-server "$BS" "$@"; }
hr() { echo; echo "============================================================"; echo "  $*"; echo "============================================================"; }

for i in $(seq 1 60); do curl -sf "$CONNECT/" >/dev/null 2>&1 && break; sleep 2; done


# ============================================================================
# 정답 1 — 커넥터 등록 + "모든 태스크" 상태 확인
# ============================================================================
hr "정답 1"
KT --create --topic s12ex_q1 --partitions 3 --replication-factor 3 --if-not-exists
cat > "$DATA_DIR/q1.txt" <<'EOF'
{"order_id":"O-9001","customer_id":"C001","amount":11000}
{"order_id":"O-9002","customer_id":"C002","amount":22000}
EOF

curl -s -X PUT "$CONNECT/connectors/q1-source/config" -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
  "tasks.max": "1",
  "file": "/data/q1.txt",
  "topic": "s12ex_q1",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.storage.StringConverter"
}' | jq -r '.name'
sleep 6

curl -s "$CONNECT/connectors/q1-source/status" \
  | jq -r '[.connector.state] + [.tasks[].state] | join(",")'

# 해설:
#   POST /connectors 대신 PUT /connectors/{name}/config 를 썼습니다. PUT 은 멱등이라
#   이미 있으면 갱신하고 없으면 생성합니다. POST 는 중복이면 409 Conflict 를 내므로
#   CI/CD 나 반복 실행되는 스크립트에서는 항상 PUT /config 를 씁니다.
#
#   ★ 이 문제의 핵심은 jq 표현식입니다.
#       jq -r '.connector.state'                     ← 이것만 보면 안 됩니다
#       jq -r '[.connector.state] + [.tasks[].state] | join(",")'   ← 이게 맞습니다
#
#   커넥터는 "일을 몇 조각으로 나눌지" 계획만 세우고, 실제 데이터는 태스크가 옮깁니다.
#   그래서 태스크가 전부 죽어도 커넥터는 RUNNING 일 수 있습니다(본문 함정 C).
#   .connector.state 만 보는 모니터링은 이런 장애에서 절대 알람을 울리지 않습니다.
#
#   운영에서는 전 커넥터를 순회하며 FAILED 를 찾는 형태로 씁니다:
for c in $(curl -s "$CONNECT/connectors" | jq -r '.[]'); do
  s=$(curl -s "$CONNECT/connectors/$c/status")
  line="$c $(echo "$s" | jq -r '.connector.state') tasks=$(echo "$s" | jq -r '[.tasks[].state] | join(",")')"
  echo "$line"
  case "$line" in *FAILED*) echo "  ↑ 알람 대상" ;; esac
done


# ============================================================================
# 정답 2 — 커넥터를 지웠는데 오프셋이 남아 이어서 읽는 문제
# ============================================================================
hr "정답 2 — 준비 (문제 상황 재현)"
KT --create --topic s12ex_q2 --partitions 1 --replication-factor 3 --if-not-exists
printf '%s\n' 'A-1' 'A-2' 'A-3' 'A-4' 'A-5' > "$DATA_DIR/q2.txt"
curl -s -X PUT "$CONNECT/connectors/q2-source/config" -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
  "tasks.max": "1", "file": "/data/q2.txt", "topic": "s12ex_q2",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.storage.StringConverter"
}' >/dev/null
sleep 6
curl -s -X DELETE "$CONNECT/connectors/q2-source" >/dev/null; sleep 3
printf '%s\n' 'B-1' 'B-2' 'B-3' > "$DATA_DIR/q2.txt"
curl -s -X PUT "$CONNECT/connectors/q2-source/config" -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
  "tasks.max": "1", "file": "/data/q2.txt", "topic": "s12ex_q2",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.storage.StringConverter"
}' >/dev/null
sleep 6
echo "재생성 후 오프셋 (5 그대로):"; KGO --topic s12ex_q2

hr "정답 2 — 진단과 해결"

# (a) 원인 확인 — connect-offsets 에 이전 position 이 남아 있습니다.
curl -s "$CONNECT/connectors/q2-source/offsets" | jq

# (b) 해결 — stop → DELETE /offsets → resume
curl -s -X PUT    "$CONNECT/connectors/q2-source/stop"    -w 'stop HTTP %{http_code}\n'
sleep 3
curl -s -X DELETE "$CONNECT/connectors/q2-source/offsets" | jq -r '.message'
curl -s -X PUT    "$CONNECT/connectors/q2-source/resume"  -w 'resume HTTP %{http_code}\n'
sleep 7

# (c) 검증 — 5 + 3 = 8
KGO --topic s12ex_q2

# 해설:
#   DELETE /connectors/{name} 은 connect-configs 에서 "설정"만 지웁니다.
#   connect-offsets 토픽의 레코드는 그대로 남습니다. 키가
#       ["q2-source",{"filename":"/data/q2.txt"}]
#   인데, 같은 이름 + 같은 파일 경로로 재생성하면 키가 정확히 일치하므로
#   Connect 는 "아, 이 커넥터는 20바이트째까지 읽었었지" 하고 그 지점부터 재개합니다.
#   새 파일이 그보다 짧으면 읽을 게 없고, 커넥터는 RUNNING 인 채 아무것도 안 합니다.
#   ★ 에러도 경고도 안 납니다. 이것이 함정 B 입니다.
#
#   왜 stop 이 먼저여야 합니까:
#     RUNNING 상태에서 DELETE /offsets 를 호출하면 이렇게 거부됩니다.
#       HTTP 409 Conflict
#       {"error_code":409,"message":"Connectors must be in the STOPPED state before
#        their offsets can be modified. This can be done for the specified connector
#        by issuing a 'PUT' request to the '/connectors/q2-source/stop' endpoint"}
#     오프셋을 바꾸는 도중에 태스크가 계속 오프셋을 쓰면 경쟁 상태가 되기 때문입니다.
#
#   PUT /pause 가 아니라 PUT /stop 인 것에 주의하세요 (Kafka 3.5+).
#     pause : 태스크를 살려 둔 채 데이터만 안 옮김. 리소스는 계속 씀. → 오프셋 수정 불가
#     stop  : 태스크를 전부 내림. 설정만 남음.                        → 오프셋 수정 가능
#
#   ★ 운영에서 더 권하는 방법은 오프셋 삭제가 아니라 "커넥터 이름 버저닝"입니다.
#       q2-source → q2-source-v2
#     오프셋 키가 달라지므로 자동으로 처음부터 읽고, 문제가 생기면 옛 커넥터를
#     그대로 되살릴 수 있습니다. 오프셋 삭제는 되돌릴 수 없는 파괴적 작업입니다.
#     3.6 미만 버전에는 DELETE /offsets 자체가 없어서 이 방법밖에 없기도 합니다.


# ============================================================================
# 정답 3 — connect-configs 의 cleanup.policy 와 파티션 수
# ============================================================================
hr "정답 3"
KT --describe --topic connect-configs

# 기대 출력:
#   Topic: connect-configs  TopicId: ...  PartitionCount: 1  ReplicationFactor: 3
#     Configs: cleanup.policy=compact
#
# 해설:
#   cleanup.policy=compact 여야 하는 이유
#     이 토픽의 키는 커넥터 이름이고, 값은 그 커넥터의 최신 설정 JSON 입니다.
#     "키별 최신 값"만 의미가 있으므로 compact 가 맞습니다.
#     delete 였다면 retention.ms(기본 7일)가 지나는 순간 설정 레코드가 삭제되고,
#     ★ 워커를 재시작하는 순간 커넥터가 전부 사라집니다. 재시작 전까지는 메모리에
#     설정이 남아 있어 멀쩡히 돌아가므로, 사고는 몇 주 뒤 예정된 재배포 때 터집니다.
#     connect-offsets, connect-status 도 같은 이유로 셋 다 compact 입니다.
#
#   PartitionCount=1 이어야 하는 이유
#     Connect 는 설정 변경 이벤트를 "로그 순서대로 재생"해서 현재 상태를 만듭니다.
#     "A 생성 → A 수정 → A 삭제 → A 재생성" 이 순서대로 읽혀야 최종 상태가 맞습니다.
#     파티션이 2개 이상이면 삭제와 재생성이 다른 파티션에 들어갈 수 있고,
#     병렬로 읽으면 순서가 뒤집혀 "방금 만든 커넥터가 사라지는" 일이 생깁니다.
#     그래서 Connect 는 아예 기동을 거부합니다:
#       org.apache.kafka.connect.errors.ConnectException: Topic 'connect-configs'
#         supplied via the 'config.storage.topic' property is required to have a
#         single partition in order to guarantee consistency, but found 3 partitions.
#
#   ⚠️ 실무 주의: 토픽을 사전 승인 절차로 손수 만드는 조직에서 이 두 값을 빼먹는
#      일이 흔합니다. 손으로 만들 거면 반드시:
#        kt --create --topic connect-configs --partitions 1 --replication-factor 3 \
#           --config cleanup.policy=compact


# ============================================================================
# 정답 4 — 싱크와 소스의 "밀림"을 확인하는 법
# ============================================================================
hr "정답 4 — 준비"
KT --create --topic s12ex_q4 --partitions 3 --replication-factor 3 --if-not-exists
printf '%s\n' 'X-1' 'X-2' 'X-3' 'X-4' > "$DATA_DIR/q4.txt"
curl -s -X PUT "$CONNECT/connectors/q4-source/config" -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
  "tasks.max": "1", "file": "/data/q4.txt", "topic": "s12ex_q4",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.storage.StringConverter"
}' >/dev/null
curl -s -X PUT "$CONNECT/connectors/q4-sink/config" -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
  "tasks.max": "1", "topics": "s12ex_q4", "file": "/data/q4-out.txt",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.storage.StringConverter"
}' >/dev/null
sleep 8

hr "정답 4"

# (a) 싱크 — 컨슈머 그룹의 LAG 컬럼
KCG --describe --group connect-q4-sink

# (b) 소스 — REST 로 소스 시스템 위치(여기서는 파일 바이트 오프셋)
curl -s "$CONNECT/connectors/q4-source/offsets" | jq

# (c) 왜 다릅니까 — 해설:
#   ┌──────────┬─────────────────────────────┬──────────────────────────────────┐
#   │          │ 소스 커넥터                  │ 싱크 커넥터                       │
#   ├──────────┼─────────────────────────────┼──────────────────────────────────┤
#   │ 저장 위치 │ connect-offsets 토픽        │ 컨슈머 그룹 connect-<이름>        │
#   │          │                             │ (= __consumer_offsets)           │
#   │ 조회     │ GET /connectors/{n}/offsets │ kcg --describe --group connect-…  │
#   │ 오프셋 의미│ 소스 시스템의 위치           │ Kafka 토픽의 오프셋               │
#   │          │ (파일 바이트, DB 커서, LSN)  │                                  │
#   │ 리셋     │ DELETE /connectors/{n}/offsets │ kcg --reset-offsets 또는 위와 동일│
#   └──────────┴─────────────────────────────┴──────────────────────────────────┘
#
#   싱크 커넥터는 그냥 컨슈머입니다. Kafka 가 이미 오프셋 관리 메커니즘을 갖고 있으니
#   그걸 그대로 씁니다. 그래서 Step 05·06 에서 배운 랙 모니터링이 그대로 적용됩니다.
#
#   ★ 이 문제의 진짜 답: 소스 커넥터에는 "랙"이라는 개념 자체가 없습니다.
#     소스는 Kafka 밖의 무언가를 읽으므로, "얼마나 뒤처졌는가"를 재려면
#     소스 시스템 쪽 지표가 필요합니다. 예를 들어
#       - JDBC 소스: max(updated_at) - 커넥터가 저장한 timestamp
#       - Debezium:  DB 의 현재 LSN/GTID - 커넥터가 저장한 LSN
#     Connect 가 알려 주는 것은 "내가 어디까지 읽었다" 뿐이고,
#     "소스에 얼마나 더 있는지"는 Connect 가 모릅니다.
#
#   그리고 컨슈머 그룹은 커넥터를 지워도 남습니다:
#     curl -X DELETE .../connectors/q4-sink   → 그룹 connect-q4-sink 는 그대로
#     kcg --delete --group connect-q4-sink    → 이걸 따로 해야 합니다


# ============================================================================
# 정답 5 — SMT 로 키를 만들어 파티션 분산시키기
# ============================================================================
hr "정답 5"
KT --create --topic s12ex_q5 --partitions 3 --replication-factor 3 --if-not-exists
printf '%s\n' C001 C002 C003 C001 C002 C003 > "$DATA_DIR/q5.txt"

curl -s -X PUT "$CONNECT/connectors/q5-source/config" -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
  "tasks.max": "1",
  "file": "/data/q5.txt",
  "topic": "s12ex_q5",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.storage.StringConverter",

  "transforms": "hoist,mkKey,extractKey",

  "transforms.hoist.type": "org.apache.kafka.connect.transforms.HoistField$Value",
  "transforms.hoist.field": "cid",

  "transforms.mkKey.type": "org.apache.kafka.connect.transforms.ValueToKey",
  "transforms.mkKey.fields": "cid",

  "transforms.extractKey.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
  "transforms.extractKey.field": "cid"
}' | jq -r '.name'
sleep 8

echo "--- 파티션 분산 확인 (세 줄 전부 0 이 아니어야 정답) ---"
KGO --topic s12ex_q5
echo "--- 키가 붙었는지 ---"
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic s12ex_q5 \
  --from-beginning --property print.key=true --property print.partition=true \
  --timeout-ms 5000 || true

# 해설:
#   3단 체인이 필요합니다. 왜 3단입니까?
#
#     ① HoistField$Value  — 문자열 "C001" 을 {"cid":"C001"} 이라는 Map 으로 감쌉니다
#     ② ValueToKey        — 값의 cid 필드를 키로 복사 → 키가 {"cid":"C001"}
#     ③ ExtractField$Key  — 키에서 cid 를 꺼내 평문 "C001" 로 만듭니다
#
#   ★ ① 을 빼고 ValueToKey 를 바로 걸면 이렇게 죽습니다:
#       org.apache.kafka.connect.errors.DataException: Only Struct objects supported
#         for [copying fields from value to key], found: java.lang.String
#     ValueToKey / InsertField / ReplaceField 같은 "필드를 다루는" SMT 는 값이
#     Struct 또는 Map 이어야 합니다. FileStreamSource 는 값을 평범한 String 으로
#     내보내므로 필드라는 개념이 없습니다.
#
#   ★ ③ 을 빼면 키는 {"cid":"C001"} 이라는 Map 이 됩니다. StringConverter 로는
#     직렬화가 안 되어 또 DataException 이 납니다. 키가 평문이어야 합니다.
#
#   실무에서는 이 3단이 거의 필요 없습니다. JDBC 소스나 Debezium 은 처음부터
#   스키마 있는 Struct 를 내보내므로 ②③ 두 단계면 끝납니다.
#   파일 소스가 유별난 것입니다.
#
#   왜 키가 필요합니까 — 키가 null 이면 프로듀서가 스티키 파티셔닝으로 한 파티션에
#   몰아 넣습니다(Step 04). 키가 생기면 murmur2(key) % partitions 로 흩어지고,
#   동시에 "같은 고객의 메시지는 항상 같은 파티션"이라는 순서 보장도 얻습니다.


# ============================================================================
# 정답 6 — DLQ + 에러 컨텍스트 헤더
# ============================================================================
hr "정답 6"
KT --create --topic s12ex_q6_src --partitions 1 --replication-factor 3 --if-not-exists
KT --create --topic s12ex_q6_dlq --partitions 1 --replication-factor 3 --if-not-exists
docker exec -i kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server "$BS" --topic s12ex_q6_src <<'EOF'
{"id":"A","v":1}
OOPS
{"id":"B","v":2}
{"unclosed":
EOF

curl -s -X PUT "$CONNECT/connectors/q6-sink/config" -H 'Content-Type: application/json' -d '{
  "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
  "tasks.max": "1",
  "topics": "s12ex_q6_src",
  "file": "/data/q6-out.txt",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter.schemas.enable": "false",

  "errors.tolerance": "all",
  "errors.log.enable": "true",
  "errors.log.include.messages": "true",
  "errors.deadletterqueue.topic.name": "s12ex_q6_dlq",
  "errors.deadletterqueue.topic.replication.factor": "3",
  "errors.deadletterqueue.context.headers.enable": "true"
}' | jq -r '.name'
sleep 10

echo "--- 커넥터 상태 (RUNNING,RUNNING 이어야 합니다) ---"
curl -s "$CONNECT/connectors/q6-sink/status" | jq -r '[.connector.state] + [.tasks[].state] | join(",")'

echo "--- 정상 2건 ---"
docker exec kafka-connect cat /data/q6-out.txt

echo "--- DLQ 2건 + 헤더 ---"
K kafka-console-consumer.sh --bootstrap-server "$BS" --topic s12ex_q6_dlq \
  --from-beginning --property print.headers=true --timeout-ms 6000 || true

# 해설:
#   3종 세트가 전부 필요합니다.
#     errors.tolerance=all                            ← 없으면 태스크가 즉시 죽습니다
#     errors.deadletterqueue.topic.name=...           ← 없으면 실패 레코드가 증발합니다
#     errors.deadletterqueue.context.headers.enable=true ← 없으면 원인을 알 수 없습니다
#
#   ★ 세 번째가 이 문제의 핵심입니다. 헤더 없이 DLQ 만 켜면 이렇게 나옵니다:
#       OOPS
#       {"unclosed":
#     원본 바이트만 덜렁 있습니다. 어느 토픽에서 왔는지, 어느 오프셋인지,
#     무엇 때문에 실패했는지 알 방법이 전혀 없습니다. 쓰레기통일 뿐입니다.
#
#     headers.enable=true 면 이렇게 나옵니다:
#       __connect.errors.topic:s12ex_q6_src,
#       __connect.errors.partition:0,
#       __connect.errors.offset:1,
#       __connect.errors.connector.name:q6-sink,
#       __connect.errors.task.id:0,
#       __connect.errors.stage:VALUE_CONVERTER,
#       __connect.errors.class.name:org.apache.kafka.connect.json.JsonConverter,
#       __connect.errors.exception.class.name:org.apache.kafka.connect.errors.DataException,
#       __connect.errors.exception.message:Converting byte[] to Kafka Connect data
#         failed due to serialization error:                    OOPS
#
#     stage 가 특히 유용합니다. KEY_CONVERTER / VALUE_CONVERTER / TRANSFORMATION /
#     TASK_PUT 중 어디서 터졌는지 알려 주므로, 문제가 "데이터"인지 "SMT 설정"인지
#     "대상 시스템"인지 즉시 갈립니다.
#     offset 을 알면 재처리도 가능합니다. 원본 토픽의 그 오프셋을 고쳐서 다시 넣으면 됩니다.
#
#   ⚠️ errors.tolerance=all 을 DLQ 없이 쓰는 것이 가장 위험합니다.
#      실패 레코드가 어디에도 안 남고 사라지는데, 커넥터는 RUNNING 이고 랙은 0 이라
#      모니터링이 정상으로 보입니다. 이 코스가 계속 말하는 "조용한 유실"의 Connect 판입니다.
#      원칙: errors.tolerance=all 을 쓸 거면 반드시 DLQ 를 함께 설정하고,
#            DLQ 토픽에 메시지가 들어오는 순간 알람을 겁니다.
#
#   ⚠️ DLQ 는 싱크 커넥터 전용입니다. 소스 커넥터에는 errors.deadletterqueue.* 가
#      없습니다. 소스 쪽 에러는 errors.tolerance 와 로그로만 다룹니다.


# ============================================================================
# 정리
# ============================================================================
hr "정리"
for c in q1-source q2-source q4-source q4-sink q5-source q6-sink; do
  curl -s -X DELETE "$CONNECT/connectors/$c" >/dev/null 2>&1 || true
done
sleep 3
for t in $(KT --list | grep -E '^s12ex_' || true); do KT --delete --topic "$t" || true; done
for g in $(KCG --list | grep -E '^connect-q' || true); do KCG --delete --group "$g" || true; done
rm -f "$DATA_DIR"/q[0-9]*.txt "$DATA_DIR"/q[0-9]*-out.txt
echo "정리 완료"