Kafka CLI 명령어 살펴보기

Docker container 접속

docker exec -it [docker-container-name] /bin/sh

 

토픽 관련 콘솔

# 토픽 생성
$ /bin/kafka-topics \
--create \
--bootstrap-server <host>:<port> \
--topic <topic-name>

# 파티션 갯수, 복제 갯수, 토픽 데이터 유지 기간 옵션 지정 토픽 생성
$ /bin/kafka-topics \
--create \
--bootstrap-server <host:port> \
--partitions 10 \
--replication-factor 1 \
--topic <topic-name> \
--config retention.ms=172800000

# 특정 토픽 정보 확인 (파티션 번호, 리더 브로커 번호, 복제 계수, ISR)
$ /bin/kafka-topics \
--bootstrap-server <host:port> \
--describe --topic <topic-name>

# 토픽 목록 확인
$ /bin/kafka-topics \
--bootstrap-server=<host:port> \
--list

# 토픽 삭제
$ /bin/kafka-topics \
--bootstrap-server <host:port> \
--delete \
--topic <topic-name>

 

발행 관련 콘솔

# 특정 토픽에 레코드 발행하기
$ /bin/kafka-console-producer \
--bootstrap-server <host:port> \
--topic <topic-name>

# 특정 토픽에 메시지 키 포함 레코드 발행하기
$ /bin/kafka-console-producer \
--bootstrap-server <host:port> \
--topic hello.kafka \
--property "parse.key=true" \
--property "key.separator=:" # default : Tab Delimiter(`t)

 

소비 관련 콘솔

# 특정 토픽 소비하기
$ /bin/kafka-console-consumer \
--bootstrap-server <host:port> \
--topic <topic-name> \
--from-beginning # 토픽에 저장된 가장 처음 데이터 출력

# 레코드 메시지 키, 메시지 값 확인
$ /bin/kafka-console-consumer \
--bootstrap-server <host:port> \
--topic <topic-name> \
--property print.key=true \
--property key.separator=":" \
--from-beginning

# 최대 컨슘 메시지 갯수 설정 : --max-messages [number]
$ /bin/kafka-console-consumer \
--bootstrap-server <host:port> \
--topic hello.kafka \
--from-beginning \
--max-messages 1

# 특정 파티션 컨슘 : --partition [partition-num]
$ /bin/kafka-console-consumer \
--bootstrap-server <host:port> \
--topic <topic-name> \
--partition 2 \
--from-beginning \

# 그룹 기반 컨슘하기
$ /bin/kafka-console-consumer \
--bootstrap-server <host:name> \
--topic <topic-name> \
--group hello-group
--from-beginning

# 특정 파티션, 오프셋 기준 소비
$ /bin/kafka-console-consumer \\
--bootstrap-server <host:port> \\
--topic <topic-name> \\
--partition <partition-number> \\
--offset <offset-number> # earlist : 파티션 가장 처음 메시지 부터 일기 시작, latest : 파티션 가장 최신 메시지 이후 읽기 시작

 

Reference

https://docs.confluent.io/kafka/operations-tools/kafka-tools.html