Apache Kafka CLI 기본 사용법

1. 다운로드 받기

• https://kafka.apache.org/downloads
• zookeeper 와 kafka 바이너리 파일이 포함되어 있다.
• zookeeper 와 kafka 는 모두 JVM 위에서 돌아가기 때문에 반드시 자바가 설치 되어야 한다.

---

2. 압축 해제 후 폴더 구조

• bin : 실행할 쉘 스크립트가 있는 디렉토리
• config : 설정 파일들이 있는 디렉토리
• lib : 실행하는 데 필요한 라이브러리들이 있는 디렉토리

---

3. 설정

vi config/server.properties

• 실행할 브로커 id 는 0
• listeners, advertised.listener 는 주석을 해제하고 호스트 설정

• 로그가 저장될 디렉토리를 지정

---

4. 실행

🍣 카프카 실행 전 zookeeper 먼저 실행

• 윈도우의 경우 bin/windows/ 아래에 있는 배치파일을 실행

sh bin/zookeeper-server-start.sh config/zookeeper.properties

🍣 카프카 실행

sh bin/kafka-server-start.sh config/server.properties

• 현재 실행되고 있는 카프카 확인

sh bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092

• 현재 카프카 브로커의 설정값이 출력

---

5. kafka-topics.sh

• 토픽에 관련된 작업을 할 수 있는 스크립트

🍣 토픽 생성하기

• test-topic 이름으로 토픽을 생성

sh bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --topic test-topic

• 기본설정이 아닌 원하는 설정을 입력하여 생성 가능

sh bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --partitions 10 --replication-factor 1 --topic test-topic2 --config retention.ms=172800000

🍣 토픽을 설정 정보 확인하기

sh bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic test-topic --describe

🍣 토픽 설정 변경하기

sh bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic test-topic --alter --partitions 3

주의!
파티션 개수는 한번 늘리면 줄일 수 없으므로 신중하게 늘려야 된다.

---

6. kafka-configs.sh

• 일부 옵션들을 설정하는 스크립트

🍣 토픽 설정 변경하기

sh bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config min.insync.replicas=2 --topic test-topic

🍣 카프카 브로커 설정 확인하기

sh bin/kafka-configs.sh --bootstrap-server localhost:9092 --broker 0 --all --describe

---

7. kafka-console-producer.sh

• 카프카 브로커에 메시지 전송 하는 스크립트

🍣 메시지 키 설정 X

• 키 설정을 하지 않으면 키는 null 로 설정되고 값만 전송
• 만약 파티션이 여러개 일 경우 각 데이터는 라운드로빈으로 파티션에 전송

sh bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test-topic

🍣 메시지 키 설정 O

• 키 설정을 하려면 다음 옵션을 추가
• 키 설정을 할 경우 같은 키로 전송된 데이터는 같은 파티션으로 전송

sh bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test-topic --property "parse.key=true" --property "key.separator=:"

---

8. kafka-console-consumer.sh

• 카프카에 적재된 데이터를 가져오는 스크립트

🍣 토픽에 저장된 데이터 값 가져오기

• --from-beginnig 옵션은 토픽에 저장된 가장 처음 데이터 부터 출력

sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
--topic test-topic \
--from-beginning

🍣 토픽에 저장된 데이터 키, 값 가져오기

• print.key=true 옵션 추가

sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
--topic test-topic \
--property print.key=true \
--property key.seperator=: \
--from-beginning

🍣 데이터 개수 지정해서 가져오기

• --max-messages 옵션을 추가하여 데이터 개수를 지정

sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
--topic test-topic \
--from-beginning \
--max-messages 3

🍣 특정 파티션에 있는 데이터만 가져오기

• --partition 옵션을 추가하여 파티션 지정

sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
--topic test-topic \
--from-beginning \
--partition 0

🍣 그룹 지정해서 데이터 가져오기

• 그룹을 지정하면 읽은 데이터를 커밋하여 다음에 해당 그룹으로 데이터를 읽으면 커밋된 데이터 이후의 데이터부터 가져온다.
• 해당 커밋내용은 __consumer_offsets 토픽에 저장된다.

sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
--topic test-topic \
--group test-group \
--from-beginning \
--max-messages 1

• __consumer_offsets 토픽

sh bin/kafka-topics.sh --bootstrap-server localhost:9092 --list

🍣 producer -> consumer

• producer 로 데이터를 보내면 consumer 로 받는다.

---

9. kafka-consumer-group.sh

• 컨슈머 그룹을 다루기 위한 스크립트

🍣 컨슈머 그룹 확인하기

• 전체 리스트 보기

sh bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

• 상세 보기

sh bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test-group --describe

• CURRENT-OFFSET : 현재까지 가져간 레코드의 오프셋 위치
• LOG-END-OFFSET : 마지막 레코드의 오프셋 위치
• LAG : LOG-END-OFFSET - CURRENT-OFFSET = 지연된 레코드 수

🍣 오프셋 리셋하기

• --reset-offsets 옵션을 추가하여 오프셋을 리셋 할 수 있다.

오프셋 리셋 옵션 종류

• --to-earliest : 가장 처음 오프셋으로 리셋
• --to-latest : 가장 마지막 오프셋으로 리셋
• --to-current : 현 시점 기준 오프셋으로 리셋
• --to-datetime {YYYY-MM-DDTHH:mmSS.sss} : 특정 일시로 오프셋 리셋
• --to-offset {long} : 특정 오프셋으로 리셋
• --shift-by {+/- long} : 현재 컨슈머 오프셋에서 앞 뒤로 옮겨서 리셋

sh bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
--group test-group \
--topic test-topic \
--reset-offsets --to-earliest \
--execute

---

10. kafka-producer-perf-test.sh

• 카프카 프로듀서로 퍼포먼스를 측정할 때 사용

---

11. kafka-reassign-partitions.sh

• 리더 파티션에 데이터가 집중적으로 모이는 경우 리더 파티션과 팔로워 파티션의 위치를 변경 할 수 있다.

---

12. kafka-delete-record.sh

• 해당 오프셋 이전의 레코드를 지울 때 사용

---

13. kafka-dump-log.sh

• 상세한 로그를 조회할 때 사용