kafka command line tool

[yeomko22]

카프카 토픽

• 카프카에서 데이터를 구분하는 가장 기본적인 개념, RDBMS의 테이블과 유사
• 하나의 토픽은 여러 파티션을 가질 수 있다. 토픽 내부에서도 파티션을 통해 데이터의 종류를 나누어 처리할 수 있다.

토픽 생성 2가지 방법

• 카프카 컨슈머 또는 프로듀서가 카프카 브로커에 생성되지 않은 토픽에 대해 데이터를 요청할 때
• 커맨드 라인 툴로 명시적으로 토픽 생성 (추천함. 왜? 토픽마다 처리되어야 하는 데이터의 특성이 다르기 때문)

토픽 생성시 주의할 점

• 동시 데이터 처리량이 많은 토픽의 경우 파티션을 크게 설정할 것
• 단기간 데이터 처리만 필요할 경우 보관 기간을 짧게 설정할 것
• 그러므로 들어오는 데이터의 양과 병렬 처리 용량을 잘 파악해서 파티션 크기를 정할 것

[yeomko22]

토픽 생성

$ bin/kafka-topics.sh \
-- create \
--bootstrap-server my-kafka:9092 \
--topic hello.kafka

$ bin/kafka-topics.sh \
--create \
--bootstrap-server my-kafka:9092 \
--partitions 3 \
--replication-factor 1 \
--config retention.ms=172800000 \
--topic hello.kafka.2

• replication factor는 토픽의 파티션을 복제할 복제 개수이다.
• 즉, 하나의 파티션을 여러개의 브로커에 복제해서 저장한다는 의미이다.
• 그러므로 replication-factor의 최대 값은 브로커의 개수와 동일하다.

생성된 토픽 조회

$ bin/kafka-topics.sh --bootstrap-server my-kafka:9092 --describe --topic hello.kafka.2

토픽 옵션 수정

$ bin/kafka-topics.sh --bootstrap-server my-kafka:9092 \
--topic hello.kafka \
--alter \
--partitions 4

• 토픽의 파티션은 늘릴 수는 있어도 줄일 수는 없다.

$ bin/kafka-configs.sh --bootstrap-server my-kafka:9092 \
> --entity-type topics \
> --entity-name hello.kafka \
> --alter --add-config retention.ms=865400000

• dynamic topic 옵션들은 kafka-config를 통해서 변경할 수 있다.

[yeomko22]

kafka-console-producer

$  bin/kafka-console-producer.sh --bootstrap-server my-kafka:9092 \
> --topic hello.kafka
> --property "parse.key=true" \
> --property "key.separator=:"

• 해당 명령어 실행하면 key value 값을 넣을 수 있는 콘솔이 실행된다.
• parse.key 옵션을 추가해야만 메시지에 키를 추가할 수 있다.
• key.separator를 직접 지정하지 않으면 기본 값은 \t 이다.
• 키의 해쉬값에 의해 파티션이 결정됨. 즉, 동일한 키를 가지면 동일한 파티션에 배정됨. 하나의 키더라도 여러 벨류값을 가질 수 있다. 즉, 키 벨류 1:1 매핑이 아니다.
• 만일 파티션을 추가한다면 파티션과 메시지 키와의 일관성은 보장되지 않는다.

[yeomko22]

kafka-console-consumer

bin/kafka-console-consumer.sh --bootstrap-server my-kafka:9092 \
> --topic hello.kafka \
> --property print.key=true \
> --property key.separator="-" \
> --group hello-group \
> --from-beginning

• --group 옵션을 통해서 컨슈머 그룹을 생성함, 컨슈머 그룹은 1개 이상의 컨슈머로 이루어져있음.
• 컨슈머 그룹을 통해 가져간 토픽의 메시지는 가져간 메시지에 대해 커밋한다. 이 떄 커밋이란 컨슈머가 특정 레코드까지 처리를 완료했다고 레코드의 오프셋 번호를 카프카 브로커에 저장하는 것이다. 이 커밋 정보는 __consumer_offsets 이름의 내부 토픽에 저장된다.
• 출력 순서가 입력 순서와 다른 것은 컨슈머가 토픽에서 데이러를 읽어올 떄 모든 파티션에서 동시에 읽기 때문

[yeomko22]

kafka-consumer-groups.sh

bin/kafka-consumer-groups.sh --bootstrap-server my-kafka:9092 \
> --group hello-group \
> --describe

Consumer group 'hello-group' has no active members.

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
hello-group     hello.kafka     3          1               1               0               -               -               -
hello-group     hello.kafka     2          3               3               0               -               -               -
hello-group     hello.kafka     1          2               2               0               -               -               -
hello-group     hello.kafka     0          4               4               0               -               -               -

• 특정 컨슈머 그룹이 각 파티션에서 어디까지 읽었는 지를 저장한다.
• CURRENT_OFFSET은 해당 컨슈머 그룹이 마지막으로 커밋한 오프셋이며, LOG-ENG-OFFSET은 파티션의 마자막 로그 오프셋이다.
• LAG는 LOG-END-OFFSET - CURRENT-OFFSET이다.

[yeomko22]

kafka-verifiable-producer, consumer

• verifiable을 사용하면 String 타입 메시지 값을 코드 없이 주고 받을 수 있다.
• 카프카 클러스터 설치 완료 후 토픽에 데이터를 전송하여 간단한 네트워크 통신 테스트 시에 유용하다.

$  bin/kafka-verifiable-producer.sh --bootstrap-server my-kafka:9092 \
> --max-messages 10 \
> --topic verify-test
{"timestamp":1631862276987,"name":"startup_complete"}
[2021-09-17 16:04:37,147] WARN [Producer clientId=producer-1] Error while fetching metadata with correlation id 1 : {verify-test=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
{"timestamp":1631862277283,"name":"producer_send_success","key":null,"value":"0","topic":"verify-test","partition":0,"offset":0}
{"timestamp":1631862277285,"name":"producer_send_success","key":null,"value":"1","topic":"verify-test","partition":0,"offset":1}
{"timestamp":1631862277285,"name":"producer_send_success","key":null,"value":"2","topic":"verify-test","partition":0,"offset":2}
{"timestamp":1631862277286,"name":"producer_send_success","key":null,"value":"3","topic":"verify-test","partition":0,"offset":3}
{"timestamp":1631862277286,"name":"producer_send_success","key":null,"value":"4","topic":"verify-test","partition":0,"offset":4}
{"timestamp":1631862277286,"name":"producer_send_success","key":null,"value":"5","topic":"verify-test","partition":0,"offset":5}
{"timestamp":1631862277286,"name":"producer_send_success","key":null,"value":"6","topic":"verify-test","partition":0,"offset":6}
{"timestamp":1631862277286,"name":"producer_send_success","key":null,"value":"7","topic":"verify-test","partition":0,"offset":7}
{"timestamp":1631862277287,"name":"producer_send_success","key":null,"value":"8","topic":"verify-test","partition":0,"offset":8}
{"timestamp":1631862277287,"name":"producer_send_success","key":null,"value":"9","topic":"verify-test","partition":0,"offset":9}
{"timestamp":1631862277294,"name":"shutdown_complete"}

$ bin/kafka-verifiable-consumer.sh --bootstrap-server my-kafka:9092 \
> --topic verify-test \
> --group-id test-group
{"timestamp":1631862349679,"name":"startup_complete"}
{"timestamp":1631862350018,"name":"partitions_assigned","partitions":[{"topic":"verify-test","partition":0}]}
{"timestamp":1631862350096,"name":"records_consumed","count":10,"partitions":[{"topic":"verify-test","partition":0,"count":10,"minOffset":0,"maxOffset":9}]}
{"timestamp":1631862350108,"name":"offsets_committed","offsets":[{"topic":"verify-test","partition":0,"offset":10}],"success":true}

• consumer 자체도 파티션에 할당된다.
• 한번에 메세지를 읽어와서 처리한다. 그 후 커밋했는지 여부도 확인 가능하다.