CLI 치트시트

curl -O https://archive.apache.org/dist/kafka/4.3.1/kafka_2.13-4.3.1.tgz
tar -xzf kafka_2.13-4.3.1.tgz
cd kafka_2.13-4.3.1

# 이후 이 페이지의 모든 명령은 이 디렉터리에서 bin/ 아래를 호출합니다
bin/kafka-topics.sh --version

export BS=localhost:9092

# 1. 지금 문제가 있는 파티션이 있는가 (URP)
bin/kafka-topics.sh --bootstrap-server $BS --describe --under-replicated-partitions

# 2. 쓰기가 막힌 파티션이 있는가 (ISR < min.insync.replicas)
bin/kafka-topics.sh --bootstrap-server $BS --describe --under-min-isr-partitions

# 3. 리더가 아예 없는 파티션이 있는가 (offline)
bin/kafka-topics.sh --bootstrap-server $BS --describe --unavailable-partitions

# 4. 컨트롤러 쿼럼은 살아 있는가
bin/kafka-metadata-quorum.sh --bootstrap-server $BS describe --status

# 5. 브로커별 디스크 사용량 (JSON)
bin/kafka-log-dirs.sh --bootstrap-server $BS --describe

# 6. 특정 그룹의 lag
bin/kafka-consumer-groups.sh --bootstrap-server $BS --describe --group my-group

# 7. lag이 큰 파티션만 정렬해서 보기 (LAG 열이 5번째)
bin/kafka-consumer-groups.sh --bootstrap-server $BS --describe --group my-group \
  | awk 'NR>1 && $6 != "-" { print $6, $1, $2, $3 }' | sort -rn | head -20

# 8. 전체 그룹의 lag 합계를 그룹별로
for g in $(bin/kafka-consumer-groups.sh --bootstrap-server $BS --list); do
  total=$(bin/kafka-consumer-groups.sh --bootstrap-server $BS --describe --group "$g" \
    | awk 'NR>1 && $6 ~ /^[0-9]+$/ { s += $6 } END { print s+0 }')
  echo "$total $g"
done | sort -rn

# 9. 누가 어느 파티션을 들고 있는가 (리밸런스 조사)
bin/kafka-consumer-groups.sh --bootstrap-server $BS --describe --group my-group --members --verbose

# 10. 그룹 상태만 (Stable / PreparingRebalance / CompletingRebalance / Empty)
bin/kafka-consumer-groups.sh --bootstrap-server $BS --describe --group my-group --state

# 11. 리셋을 먼저 미리보기 (--execute 를 빼면 dry-run)
bin/kafka-consumer-groups.sh --bootstrap-server $BS --reset-offsets \
  --group my-group --topic orders --to-latest

# 12. 최신으로 리셋 — lag을 즉시 0으로 (미처리 데이터를 버립니다)
bin/kafka-consumer-groups.sh --bootstrap-server $BS --reset-offsets \
  --group my-group --topic orders --to-latest --execute

# 13. 1시간 전으로 되돌려 재처리
bin/kafka-consumer-groups.sh --bootstrap-server $BS --reset-offsets \
  --group my-group --all-topics --by-duration PT1H --execute

# 14. 파티션별 최신 오프셋(LEO)과 최초 오프셋
bin/kafka-get-offsets.sh --bootstrap-server $BS --topic orders --time latest
bin/kafka-get-offsets.sh --bootstrap-server $BS --topic orders --time earliest

# 15. 특정 시각(epoch ms) 기준 오프셋 — 재처리 시작점 찾기
bin/kafka-get-offsets.sh --bootstrap-server $BS --topic orders --time 1767225600000

# 16. 리텐션을 1시간으로 줄여 디스크 확보 (원래 값은 반드시 기록해 두세요)
bin/kafka-configs.sh --bootstrap-server $BS --alter \
  --entity-type topics --entity-name bloated-topic --add-config retention.ms=3600000

# 17. 되돌리기 — 토픽 오버라이드 제거
bin/kafka-configs.sh --bootstrap-server $BS --alter \
  --entity-type topics --entity-name bloated-topic --delete-config retention.ms

# 18. 리더 편중 해소 (재시작 직후에 자주 필요)
bin/kafka-leader-election.sh --bootstrap-server $BS \
  --election-type preferred --all-topic-partitions

# 19. 특정 클래스만 DEBUG로 (재시작 없이, 브로커 1번)
bin/kafka-configs.sh --bootstrap-server $BS --alter \
  --entity-type broker-loggers --entity-name 1 \
  --add-config org.apache.kafka.clients.consumer=DEBUG

# 20. read_committed가 멈췄을 때 — 멈춘 트랜잭션 찾기
bin/kafka-transactions.sh --bootstrap-server $BS find-hanging --topic orders

# SASL_SSL + SCRAM-SHA-512 예시
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
  username="admin" \
  password="admin-secret";
ssl.truststore.location=/var/private/ssl/client.truststore.jks
ssl.truststore.password=changeit

bin/kafka-topics.sh --bootstrap-server broker1:9093 \
  --command-config client.properties --list

# 1. 클러스터 UUID 생성 (한 번만)
KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
echo "$KAFKA_CLUSTER_ID"

# 2. 로그 디렉터리 포맷 — KRaft에서 반드시 필요한 단계
bin/kafka-storage.sh format --standalone -t "$KAFKA_CLUSTER_ID" -c config/server.properties

# 3. 기동
bin/kafka-server-start.sh config/server.properties

# 4. 확인
bin/kafka-metadata-quorum.sh --bootstrap-server localhost:9092 describe --status

# 모든 노드가 동일한 클러스터 ID로 포맷되어야 합니다.
# ID가 다르면 INCONSISTENT_CLUSTER_ID 로 조인이 거부됩니다.
bin/kafka-storage.sh format -t "$KAFKA_CLUSTER_ID" -c config/server.properties

# SASL/SCRAM 을 쓸 경우 브로커 간 인증 자격 증명을 포맷 시점에 심습니다
bin/kafka-storage.sh format -t "$KAFKA_CLUSTER_ID" -c config/server.properties \
  --add-scram 'SCRAM-SHA-256=[name="admin",password="admin-secret"]'

# 1) 새 컨트롤러를 포맷·기동한 뒤 복제가 따라잡았는지 확인
bin/kafka-metadata-quorum.sh --bootstrap-server localhost:9092 describe --replication

# 2) 쿼럼에 추가 (컨트롤러 자신의 설정 파일을 --command-config 로 넘깁니다)
bin/kafka-metadata-quorum.sh --command-config config/controller.properties \
  --bootstrap-server localhost:9092 add-controller

# 3) 제거는 종료 "전에" — 순서를 바꾸면 쿼럼이 흔들립니다
bin/kafka-metadata-quorum.sh --bootstrap-server localhost:9092 \
  remove-controller --controller-id 3 --controller-directory-id <directory-id>

{
  "topics": [
    { "topic": "orders" },
    { "topic": "payments" }
  ],
  "version": 1
}

# 계획 생성: 현재 배치와 제안 배치를 함께 출력합니다.
# "Current partition replica assignment" 출력은 롤백용으로 반드시 파일로 저장하세요.
bin/kafka-reassign-partitions.sh --bootstrap-server $BS \
  --topics-to-move-json-file topics-to-move.json \
  --broker-list "1,2,3,4,5,6" --generate

# 실행: throttle 없이 돌리면 복제 트래픽이 클러스터를 잡아먹습니다 (bytes/sec)
bin/kafka-reassign-partitions.sh --bootstrap-server $BS \
  --reassignment-json-file reassignment.json --execute --throttle 50000000

# 검증: 완료되면 throttle 설정이 자동으로 제거됩니다.
# --verify 를 돌리지 않으면 throttle 이 남아 복제가 계속 느립니다.
bin/kafka-reassign-partitions.sh --bootstrap-server $BS \
  --reassignment-json-file reassignment.json --verify

{
  "version": 1,
  "partitions": [
    { "topic": "orders", "partition": 0, "replicas": [1, 2, 3] },
    { "topic": "orders", "partition": 1, "replicas": [2, 3, 1] },
    { "topic": "orders", "partition": 2, "replicas": [3, 1, 2] }
  ]
}

# 진행 중인 재할당 목록
bin/kafka-reassign-partitions.sh --bootstrap-server $BS --list

# 취소 — 계획 파일이 필요합니다
bin/kafka-reassign-partitions.sh --bootstrap-server $BS \
  --reassignment-json-file reassignment.json --cancel

# 여러 단계로 나눠 진행할 때는 throttle 을 유지합니다
bin/kafka-reassign-partitions.sh --bootstrap-server $BS \
  --reassignment-json-file reassignment.json --verify --preserve-throttles

# 목록·상세·상태
bin/kafka-share-groups.sh --bootstrap-server $BS --list
bin/kafka-share-groups.sh --bootstrap-server $BS --describe --group my-share-group
bin/kafka-share-groups.sh --bootstrap-server $BS --describe --group my-share-group --members
bin/kafka-share-groups.sh --bootstrap-server $BS --describe --group my-share-group --state

# 콘솔로 소비해 보기
bin/kafka-console-share-consumer.sh --bootstrap-server $BS --topic orders

# 그룹 설정을 그룹 단위로 오버라이드 (GROUP 리소스 타입)
bin/kafka-configs.sh --bootstrap-server $BS --alter \
  --entity-type groups --entity-name my-share-group \
  --add-config share.delivery.count.limit=3

# 오프셋 리셋
bin/kafka-share-groups.sh --bootstrap-server $BS --reset-offsets \
  --group my-share-group --topic orders --to-latest --execute