[Kafka] CLI

개요

Kafka CLI는 사용자가 Kafka 설치를 관리하고 애플리케이션을 구축할 수 있도록 돕는 대화형 셸 환경입니다. 이 도구를 통해 Kafka의 topic, connector, consumer 및 producer의 기본 기능을 호출하고 관리할 수 있습니다.

 

Kafka command path를 지정하려면 다음을 참조하세요.

2024.09.16 - [제품/Kafka] - [Kafka] .bashrc에 Kafka CLI command path 지정하기

 

1. topic 관리 

$KAFKA_HOME/bin/kafka-topics.sh

(1) topic 생성

# 'MyTopic' topic 생성
kafka-topics.sh --bootstrap-server 172.16.4.203:9092 --create --topic MyTopic 

# partition=5인 'MyTopic2' topic 생성
kafka-topics.sh --bootstrap-server 172.16.4.203:9092 --create  --topic MyTopic2 --partitions 5

# replication factor가 있는 'MyTopic3' topic 생성
kafka-topics.sh --bootstrap-server 172.16.4.203:9092 --create  --topic MyTopic3 --partitions 5 --replication-factor 2

 

(2) topic 나열

# 토픽 나열하기
kafka-topics.sh --bootstrap-server 172.16.4.203:9092 --list

 

(3) topic 설명

# 토픽을 자세히 보려면
kafka-topics.sh --bootstrap-server 172.16.4.203:9092 --topic MyTopic2 --describe

 

(4) topic 삭제

#토픽 지우기
kafka-topics.sh --bootstrap-server 172.16.4.203:9092 --topic MyTopic2 --delete

 

[note]

server.config에 delete.topic.enable=true를 설정하고, 위의 커맨드로 토픽이 삭제되지 않는다면: 

우선  메지지가 comsuming 되고 있는지 확인하세요.  comsuming 하고 있을되는  토픽이 삭제될 수 없습니다.

 

>> 토픽을 삭제하고, 같은 이름을 가진 토픽을 생성 -> 기존의 메시지를 볼수 있는가? 아니오

>> 기존의 메시지를 지우세요. 

 

2. 이벤트 게시하기

$KAFKA_HOME/bin/kafka-console-producer.sh 

(1) 이벤트 게시하기

#이벤트 게시하기
kafka-console-producer.sh --bootstrap-server 172.16.4.203:9092 --topic MyTopic
> start producing message
> stop producing message
# 
> ctrl + c

 

위의 커맨드는 메시지를 게시할 수 있는 프롬프트를 제공합니다.

 

(2)  속성을 설정하고 이벤트 게시하기

# broker에게서 ack 메시지를 받도록 설정
# producer-property: acks=all
kafka-console-producer.sh --bootstrap-server 172.16.4.203:9092 --topic MyTopic --producer-property acks=all

 

 

[Note]

Key가 없는 경우(key가 null인 경우): 데이터를 모든 파티션에 분산

Key가 있는 경우: 같은 키를 가진 데이터는 항상 같은 파티션으로 할당

3. 이벤트 읽어오기

$KAFKA_HOME/bin/kafka-console-consumer.sh 

consumer은 partition에서 data를 순서대로 읽습니다.

(1) 이벤트 읽어오기

#새 이벤트를 읽어옵니다. (tail)
kafka-console-consumer.sh --bootstrap-server 172.16.4.203:9092 --topic MyTopic

#이벤트를 처음부터 순서대로 읽어옵니다. 
kafka-console-consumer.sh --bootstrap-server 172.16.4.203:9092 --topic MyTopic --from-beginning

 

4. consumer 그룹

$KAFKA_HOME/bin/kafka-consumer-groups.sh

메시지를 소비하는 consumer 그룹

# Consumer 그룹 나열하기
kafka-consumer-groups.sh --bootstrap-server 172.16.4.203:9092 --list

 

# 특정 Consumer 그룹 상세보기
kafka-consumer-groups.sh --bootstrap-server 172.16.4.203:9092 --describe --group es_logstash

 

# 특정 Consumer 그룹 생성하기


# 특정 Consumer 그룹 삭제하기
./kafka-consumer-groups.sh --bootstrap-server 192.168.xx.xx:9092,192.168.xx.xx:9092,192.168.xx.xx:9092 --delete --group group-01

 

 

# offset은 특정 지점까지 소비된 메시지의 개수입니다. 
# offset을 유지하면 consumer가 중복 없이 원활하게 소비를 재시작할 수 있습니다. 
# 소비 상태와 관계없이 consumer 그룹이 전체 파티션을 처리해야 하는 경우 오프셋을 재설정함으로써 가능합니다.
# offset을 재설정하고 가장 오래된 메시지부터 시작하는 명령어는 아래와 같습니다.

kafka-consumer-groups.sh --bootstrap-server 172.16.4.203:9092 --reset-offsets --group es_logstash --topic estest --to-earliest  --execute

kafka-consumer-groups.sh --bootstrap-server 172.16.4.203:9092 --reset-offsets --group es_logstash --topic estest --to-earliest --dry-run

 

 

[참조]

https://www.redpanda.com/guides/kafka-tutorial-kafka-cheat-sheet

Kafka Cheat Sheet