Kafka 2. CLI

kafka local에서 시작하기(Mac)

• java 11 설치
• kafka 설치

  brew install kafka

• zookeeper를 이용한 실행

  zookeeper-server-start /opt/homebrew/etc/kafka/zookeeper.properties
  kafka-server-start /opt/homebrew/etc/kafka/server.properties

• kraft를 이용한 실행

  kafka-storage random-uuid # 새로운 uuid 생성
  kafka-storage format -t <uuid> -c /opt/homebrew/etc/kafka/kraft/server.properties
  kafka-server-start.sh /opt/homebrew/etc/kafka/kraft/server.properties

kafka-topic

• topic에 대한 명령어
• --bootstrap-server : 연결하고자하는 부트스트랩 서버
• --command-config : config file
• --create --topic {topic_name} : topic 생성
  • --partitions : partition 수 지정
    • server.properties 의 num.properties의 값을 조정해서 기본값 설정 가능
    • 초기 상태라면 1로 설정되어있음
  • --replication-factor : repica factor 수 지정, broker 수보다 크면안됨
• --list : topic list
• --describe --topic {topic_name} : topic에 대한 상세 내용
• --delete --topic {topic_name} : topic 삭제

kafka-topics --bootstrap-server localhost:9092 (+)
## list
 --list 
## create
 --topic first_topic --create
 --topic second_topic --create --partitions 3
 --topic third_topic --create --partitions 3 --replication-factor 2
 --topic third_topic --create --partitions 3 --replication-factor 1
## describe
 --topic first_topic --describe
## delete
 --topic first_topic --delete

kafka-console-producer

• producer 실행에 대한 명령어
• 존재하지 않는 토픽에 메시지를 생성할때 자동으로 토픽이 생성되게 할 수 있음
  • server.properties의 auto.create.topics.enable=true
  • 하지만 프로덕션 환경에서는 이 값을 false하는 것이 권장됨
• --bootstrap-server : 연결하고자하는 부트스트랩 서버
• --topic : 메시지가 저장된 topic
• --producer-property
  • acks=0/1/all : acks 값 설정
  • partitioner.class=org.apache.kafka.clients.producer.RoundRobinPartitioner
    • RoundRobin방식으로 설정, 메시지를 보낼때마다 partition을 변경, 프로덕션 환경에서는 비효율적이여서 쓰지 않음
    • 이 방식을 설정하지 않으면 message의 크기가 16KB가 될때까지 같은 partition에 메시지를 보냄
• key값 지정하도록 설정
  • --property parse.key=true
  • --property key.separator=: : key와 value의 구분자를 : 사용

kafka-console-producer --bootstrap-server localhost:9092 (+)
 --topic first_topic 
 --topic first_topic --producer-property acks=all
 --topic first_topic --property parse.key=true --property key.separator=:

kafka-console-consumer

• consumer 실행에 관한 명령어
• --from-beginning : consumer의 실행 시점에 관계없이 해당 토픽의 첫 메시지부터 소비
• --formatter : 포맷 지정
• --property
  • print.timestamp=true : message를 받은 시간 표시
  • print.key=true : key 출력
  • print.value=true : value 출력
  • print.partition=true : Partition number 출력
• consumer의 timestamp는 partition 내의 메시지 저장 순서를 보장함

kafka-console-consumer --bootstrap-server localhost:9092 (+) 
 --topic second_topic
 --topic second_topic --from-beginning
 --topic second_topic --formatter kafka.tools.DefaultMessageFormatter --property print.timestamp=true --property print.key=true --property print.value=true --property print.partition=true --from-beginning

consumer group

• consumer group을 만들어 partition을 group내에서 나눠서 담당
• group내에 consumer 가 없는 inactive 상태가 offsets.retention.minutes 로 설정된 시간만큼 지속되면 topic에 저장되어 있던 offset이 날라간다.

kafka-console-consumer --bootstrap-server localhost:9092 (+) 
 --topic third_topic --group my-first-application
 --topic third_topic --group my-first-application
 --topic third_topic --group my-second-application --from-beginning

• --describe : consumer group의 정보를 알 수 있음
  • current_offset : group이 소비한 메시지 수
  • log_end_offset : topic에 발행된 메시지 수
  • lag : log_end_offset - current_offset
• 지속적으로 사용할 consumer라면 consumer group을 생성하는 것을 추천

kafka-consumer-groups --bootstrap-server localhost:9092 (+) 
 --describe --group my-second-application

• --reset-offsets : offset을 재설정

  • to-earliest : offset을 0으로 초기화

• --dry-run : 시험적으로 실행해봄. 결과에는 반영되지않음

• --execute : 실제로 실행함

• offset을 재설정하기 위해서는 consumer group이 inactive한 상태여야함

• --dry-run/--execute 두 옵션 모두 주지 않으면 --dry-run

  • 추후 버전에서는 --dry-run이 실행되지 않고 --execute 할 건지 사용자에게 물어보는 형태로 변경될 예정

  kafka-consumer-groups --bootstrap-server localhost:9092 (+) 
    --group my-second-application --reset-offsets to-earliest --topic second_topic --execute