Connect 치트시트

export CW=http://localhost:8083

# 1) 전체 커넥터의 상태를 한 번에 (가장 유용한 한 줄)
curl -s "$CW/connectors?expand=status" | jq -r '
  to_entries[] | "\(.value.status.connector.state)\t\(.key)\t태스크: " +
  ([.value.status.tasks[].state] | join(","))'

# 출력 예
# RUNNING   mysql-source     태스크: RUNNING,RUNNING
# FAILED    s3-sink          태스크: FAILED,RUNNING

# 2) 실패한 커넥터만 골라내기
curl -s "$CW/connectors?expand=status" | jq -r '
  to_entries[]
  | select(.value.status.connector.state == "FAILED"
           or any(.value.status.tasks[]; .state == "FAILED"))
  | .key'

# 3) 실패 원인 확인 — trace 에 스택트레이스가 들어 있습니다
curl -s "$CW/connectors/s3-sink/status" | jq -r '.tasks[] | select(.state=="FAILED") | .trace'

# 4) 실패한 태스크만 재시작 (Connect 는 자동 재시작하지 않습니다)
curl -s -X POST "$CW/connectors/s3-sink/restart?includeTasks=true&onlyFailed=true"

# 배포 전 검증 — 실제로 만들지 않고 오류만 확인합니다
curl -s -X PUT "$CW/connector-plugins/FileStreamSinkConnector/config/validate" \
  -H "Content-Type: application/json" \
  -d '{
    "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
    "tasks.max": "1",
    "topics": "orders",
    "file": "/tmp/orders.sink.txt"
  }' | jq '.error_count, [.configs[] | select(.value.errors | length > 0) | {name: .value.name, errors: .value.errors}]'

# 생성 (POST) — name 과 config 를 감싼 형태
curl -s -X POST "$CW/connectors" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "orders-file-sink",
    "config": {
      "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
      "tasks.max": "1",
      "topics": "orders",
      "file": "/tmp/orders.sink.txt"
    }
  }'

# 멱등 생성/갱신 (PUT) — config 만 보냅니다. GitOps 에 적합합니다.
# 주의: 설정 "전체"를 보내야 합니다. 빠진 키는 기본값으로 돌아갑니다.
curl -s -X PUT "$CW/connectors/orders-file-sink/config" \
  -H "Content-Type: application/json" \
  -d '{
    "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
    "tasks.max": "2",
    "topics": "orders",
    "file": "/tmp/orders.sink.txt"
  }'

# 일부만 바꾸기 (PATCH) — 나머지는 보존됩니다
curl -s -X PATCH "$CW/connectors/orders-file-sink/config" \
  -H "Content-Type: application/json" \
  -d '{ "tasks.max": "4" }'

# 1) 반드시 stop (pause 로는 안 됩니다)
curl -s -X PUT "$CW/connectors/mysql-source/stop"

# 2) STOPPED 확인
curl -s "$CW/connectors/mysql-source/status" | jq -r '.connector.state'

# 3) 현재 오프셋 확인 — 되돌릴 수 있게 저장해 두세요
curl -s "$CW/connectors/mysql-source/offsets" | tee mysql-source-offsets.bak.json | jq

# 4) 초기화 (되돌릴 수 없습니다)
curl -s -X DELETE "$CW/connectors/mysql-source/offsets"

# 5) 재개
curl -s -X PUT "$CW/connectors/mysql-source/resume"

curl -s -X PUT "$CW/admin/loggers/org.apache.kafka.connect.runtime.WorkerSinkTask" \
  -H "Content-Type: application/json" -d '{"level": "DEBUG"}'

# 현재 명시적으로 설정된 로거 확인
curl -s "$CW/admin/loggers" | jq

# 조사 후 되돌리기
curl -s -X PUT "$CW/admin/loggers/org.apache.kafka.connect.runtime.WorkerSinkTask" \
  -H "Content-Type: application/json" -d '{"level": "INFO"}'

#!/usr/bin/env bash
set -euo pipefail
CW="${CW:-http://localhost:8083}"
CONNECTOR="$1"

for attempt in 1 2 3 4 5; do
  code=$(curl -s -o /dev/null -w '%{http_code}' -X POST \
    "$CW/connectors/$CONNECTOR/restart?includeTasks=true&onlyFailed=true")

  if [ "$code" = "200" ] || [ "$code" = "204" ]; then
    echo "재시작 요청 성공"
    break
  fi

  if [ "$code" = "409" ]; then
    # 리밸런스 중입니다. 리밸런스가 이미 태스크를 재시작했을 수 있으므로
    # 재시도 전에 상태를 먼저 확인합니다.
    state=$(curl -s "$CW/connectors/$CONNECTOR/status" | jq -r '.connector.state')
    if [ "$state" = "RUNNING" ]; then
      echo "리밸런스가 이미 복구했습니다 — 재시작 불필요"
      exit 0
    fi
    echo "409 (리밸런스 중) — ${attempt}회, 대기 후 재시도"
    sleep $((attempt * 10))
    continue
  fi

  echo "예상치 못한 응답: $code" >&2
  exit 1
done

# 최종 상태 확인
curl -s "$CW/connectors/$CONNECTOR/status" | jq '{connector: .connector.state, tasks: [.tasks[].state]}'

{
  "connector.class": "io.example.MySinkConnector",
  "tasks.max": "2",
  "topics.regex": "prod\\.orders\\..*",

  "//": "transforms 에 나열한 별칭 순서가 곧 적용 순서입니다",
  "transforms": "extractKey,maskPii,addSource,route",

  "//1": "1) 값의 orderId 필드로 키를 만듭니다 (파티셔닝 근거 확보)",
  "transforms.extractKey.type": "org.apache.kafka.connect.transforms.ValueToKey",
  "transforms.extractKey.fields": "orderId",

  "//2": "2) 개인정보 필드를 마스킹합니다",
  "transforms.maskPii.type": "org.apache.kafka.connect.transforms.MaskField$Value",
  "transforms.maskPii.fields": "email,phone,ssn",
  "transforms.maskPii.replacement": "REDACTED",

  "//3": "3) 원본 위치를 필드로 남깁니다 (사후 추적용)",
  "transforms.addSource.type": "org.apache.kafka.connect.transforms.InsertField$Value",
  "transforms.addSource.topic.field": "_src_topic",
  "transforms.addSource.partition.field": "_src_partition",
  "transforms.addSource.offset.field": "_src_offset",

  "//4": "4) prod. 접두어를 떼어 대상 토픽명을 바꿉니다",
  "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.route.regex": "prod\\.(.*)",
  "transforms.route.replacement": "$1"
}

{
  "transforms": "dropTombstones",
  "transforms.dropTombstones.type": "org.apache.kafka.connect.transforms.Filter",
  "transforms.dropTombstones.predicate": "isTombstone",

  "predicates": "isTombstone",
  "predicates.isTombstone.type": "org.apache.kafka.connect.transforms.predicates.RecordIsTombstone",

  "//": "negate 를 true 로 두면 반대로 tombstone 만 남깁니다",
  "//negate": "transforms.dropTombstones.negate: true"
}

export BS=localhost:9092

# config: 파티션 1개 필수, compact
bin/kafka-topics.sh --bootstrap-server $BS --create --if-not-exists \
  --topic connect-configs --partitions 1 --replication-factor 3 \
  --config cleanup.policy=compact

# offsets: source 커넥터 오프셋
bin/kafka-topics.sh --bootstrap-server $BS --create --if-not-exists \
  --topic connect-offsets --partitions 25 --replication-factor 3 \
  --config cleanup.policy=compact

# status: 커넥터·태스크 상태
bin/kafka-topics.sh --bootstrap-server $BS --create --if-not-exists \
  --topic connect-status --partitions 5 --replication-factor 3 \
  --config cleanup.policy=compact

# 확인 — cleanup.policy 가 compact 인지 반드시 봅니다
for t in connect-configs connect-offsets connect-status; do
  echo "--- $t"
  bin/kafka-topics.sh --bootstrap-server $BS --describe --topic "$t"
done

bootstrap.servers=localhost:9092

# 워커 그룹 ID. 컨슈머 그룹 ID 와 겹치지 않게 하세요.
group.id=connect-cluster

# 컨버터 — 스키마를 쓰지 않으면 schemas.enable 를 false 로
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false

# 내부 토픽
config.storage.topic=connect-configs
config.storage.replication.factor=3
offset.storage.topic=connect-offsets
offset.storage.partitions=25
offset.storage.replication.factor=3
status.storage.topic=connect-status
status.storage.partitions=5
status.storage.replication.factor=3

# 플러그인 격리 — 여기에 없으면 커넥터가 보이지 않습니다
plugin.path=/opt/connectors

# REST
listeners=http://:8083
# 컨테이너에서는 다른 워커가 도달할 수 있는 주소를 알려야 합니다
#rest.advertised.host.name=connect-1.internal
#rest.advertised.port=8083

# 롤링 재시작 중 불필요한 태스크 재배치를 미룹니다
scheduled.rebalance.max.delay.ms=300000

{
  "connector.class": "io.example.MySinkConnector",
  "topics": "orders",

  "//": "이 네 개는 함께 설정해야 의미가 있습니다",
  "errors.tolerance": "all",
  "errors.deadletterqueue.topic.name": "orders-sink-dlq",
  "errors.deadletterqueue.topic.replication.factor": "3",
  "errors.deadletterqueue.context.headers.enable": "true",

  "//log": "DLQ 와 별개로 로그도 남깁니다. 다만 레코드 내용은 개인정보 위험이 있습니다",
  "errors.log.enable": "true",
  "errors.log.include.messages": "false",

  "//retry": "일시적 오류는 재시도로 흡수합니다 (기본값 0 은 재시도 없음)",
  "errors.retry.timeout": "60000",
  "errors.retry.delay.max.ms": "10000"
}

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic orders-sink-dlq --from-beginning --max-messages 5 \
  --property print.headers=true \
  --property print.key=true \
  --property print.offset=true

# 실패 단계와 예외만 빠르게 집계하려면 (stage 로 원인 분류)
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic orders-sink-dlq --from-beginning --timeout-ms 10000 \
  --property print.headers=true --property print.value=false \
  | grep -oE '__connect\.errors\.(stage|exception\.class\.name):[^,]*' \
  | sort | uniq -c | sort -rn

# sink 커넥터는 __consumer_offsets 를 쓰므로 일반 컨슈머 도구가 통합니다
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list \
  | grep '^connect-'

bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group connect-orders-file-sink

# source 커넥터는 위 방법으로 볼 수 없습니다 — REST 를 씁니다
curl -s "$CW/connectors/mysql-source/offsets" | jq