[Kafka] Kafka Command-Line Tool(2) - kafka.topics.sh

kafka-topics.sh

이 커맨드 라인 툴을 통해 토픽(topic)과 관련된 명령을 실행할 수 있다. 토픽이란 카프카에서 데이터를 구분하는 가장 기본적인 개념이다. 마치 RDBMS에서 사용하는 테이블과 유사하다고 볼 수 있다. 카프카 클러스터에 토픽은 여러 개 존재할 수 있다.

토픽에는 파티션(partition)이 존재하는데 파티션의 개수는 최소 1개부터 시작한다. 파티션은 카프카에서 토픽을 구성하는 데에 아주 중요한 요소이다. 파티션을 통해 한 번에 처리할 수 있는 데이터양을 늘릴 수 있고 토픽 내부에서도 파티션을 통해 데이터의 종류를 나누어 처리할 수 있기 때문이다.

 

 

토픽을 생성하는 상황은 크게 2가지가 있다.

 

1. 카프카 컨슈머 또는 프로듀서가 카프카 브로커에 생성되지 않은 토픽에 대해 데이터를 요청할 때

2. 커맨드 라인 툴로 명시적으로 토픽을 생성할 때

 

 

 

 

토픽을 효과적으로 유지보수하기 위해서는 토픽을 명시적으로 생성하는 것을 추천한다. 토픽마다 처리되어야 하는 데이터의 특성이 다르기 때문에 토픽에 들어오는 데이터양과 병렬로 처리되어야 하는 용량을 잘 파악하여 생성하는 것이 중요하다.

 

 

 

 

토픽 생성

 

 

kafka-topics.sh를 통해 토픽 관련 명령을 실행할 수 있다.

--create 옵션을 사용해 hello.kafka라는 이름을 가진 토픽을 생성할 수 있다.

 

 

$ bin/kafka-topics.sh --create --bootstrap-server my-kafka:9092 --topic hello.kafka
WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.
Created topic hello.kafka.

 

1. --create 옵션 : 토픽을 생성하는 명령어임을 명시

 

2. --bootstrap-server : 토픽을 생성할 카프카 클러스터를 구성하는 브로커들의 IP와 port를 적는다.
=> 필자는 2022.10.26 - [Message Queue/Kafka] - [Kafka] Kafka Command-Line Tool(1) - 설정 에서 브로커 IP를 hosts에 설정해주었다.

 

3. --topic : 토픽 이름을 작성한다. 

 

 

 

 

토픽 생성 시 클러스터 정보와 토픽 이름은 토픽을 만들기 위한 필수 값이다. 이렇게 만들어진 토픽은 파티션 개수, 복제 개수 등과 같이 다양한 옵션이 포함되어 있지만 모두 브로커에 설정된 기본값으로 설정되었다. 만약 파티션 개수, 복제 개수, 토픽 데이터 유지 기간 옵션들을 지정하여 토픽을 생성하고 싶다면 다음과 같이 명령을 실행하면 된다.

 

 

$ bin/kafka-topics.sh --create --bootstrap-server my-kafka:9092 \
--partitions 3 \
--replication-factor 1 \
--config retention.ms=172800000 \
--topic hello.kafka.2
WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.
Created topic hello.kafka.2.

 

1. --partitions : 파티션 개수를 지정할 수 있다. 파티션 최소 개수는 1개이다. 만약 이 옵션을 사용하지 않으면 카프카 브로커 설정파일(config/server.properties)에 있는 num.partitions 옵션값을 따라 생성된다.

 

2. --replication-factor : 토픽의 파티션을 복제할 복제 개수를 적는다.

1은 복제를 하지 않고 사용한다는 의미다.

2이면 1개의 복제본을 사용하겠다는 의미이다. 파티션의 데이터는 각 브로커마다 저장된다. 한 개의 브로커에 장애가 발생하더라도 나머지 한 개 브로커에 저장된 데이터를 사용하여 안전하게 데이터 처리를 지속적으로 할 수 있다. 복제 개수 최소 설정은 1이고 최대 설정은 통신하는 카프카 클러스터의 브로커 개수이다.

현재 실습하고 있는 실습용 카프카의 브로커 개수는 1개이므로 설정할 수 있는 최대 복제 설정은 1이다. 이 설정을 명시적으로 하지 않으면 카프카 브로커 설정에 있는 default.replication.factor 옵션값에 따라서 생성된다.

 

3. --config : kafka-topics.sh 명령에 포함되지 않은 추가 설정을 할 수 있다. retention.ms는 토픽의 데이터를 유지하는 기간을 뜻한다. 172800000ms는 2일을 밀리세컨드 단위로 나타낸 것이다. 2일이 지난 토픽의 데이터는 삭제된다.

 

 

토픽 리스트 조회

 

 

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

 

--list : 해당 옵션을 사용하여 카프카 클러스터에 생성된 토픽들의 이름을 확인할 수 있다.

위에서 생성한 토픽 2개가 출력이 되었다. 카프카에 내부 관리를 위한 인터널 토픽(internal topic)이 존재하는데, 실질적으로 운영하는 데에 사용되지 않으므로 --exclude -internal 옵션을 추가하여 조회 시 목록에서 제외할 수도 있다.

 

 

 

토픽 상세 조회

 

 

bin/kafka-topics.sh --bootstrap-server my-kafka:9092 --describe --topic hello.kafka.2
Topic: hello.kafka.2    TopicId: gjREMXUCTQ20QATeDrmmsA    PartitionCount: 3    ReplicationFactor: 1    Configs: segment.bytes=1073741824,retention.ms=172800000
    Topic: hello.kafka.2    Partition: 0    Leader: 0    Replicas: 0    Isr: 0
    Topic: hello.kafka.2    Partition: 1    Leader: 0    Replicas: 0    Isr: 0
    Topic: hello.kafka.2    Partition: 2    Leader: 0    Replicas: 0    Isr: 0

 

 

이미 생성된 토픽의 상태를 --describe 옵션을 사용하여 확인할 수 있다. 파티션 개수가 몇 개인지, 복제된 파티션이 위치한 브로커의 번호, 기타 토픽을 구성하는 설정들을 출력한다.

kafka.2 토픽의 정보를 보면 3개 파티션이 모두 Leader 가 0으로 표시되어 있는데 0번 브로커에 위치하고 있음을 뜻한다. 여러 대의 브로커로 카프카 클러스터를 운영할 때 토픽의 리더 파티션이 일부 브로커에 몰려있을 수 있는데, 이를 확인하기 위해 --describe 옵션을 사용할 수 있다. 리더 파티션이 일부 브로커에 몰려있는 경우 카프카 클러스터 부하가 특정 브로커들로 몰릴 수 있다. 부하가 분산되지 못하면 데이터 통신 쏠림 현상으로 인해 네트워크 대역의 이슈가 생길 수 있다. 카프카 클러스터의 성능이 생각보다 좋지 않다면 토픽 상세 조회 명령을 통해 토픽의 리더 파티션 쏠림 현상을 확인하는 것도 좋은 방법이다.

 

 

 

토픽 옵션 수정

 

 

토픽에 설정된 옵션을 변경하기 위해서는 kafka-topics.sh 또는 kafka-configs.sh 두 개를 사용해야 한다.

파티션 개수를 변경하려면 kafka-topics.sh를 사용해야 하고 토픽 삭제 정책인 리텐션 기간을 변경하려면 kafka-configs.sh를 사용해야 한다.

이와 같이 토픽 설정 옵션이 파편화된 이유는 토픽에 대한 정보를 관리하는 일부 로직이 다른 명령어로 넘어갔기 때문이다. 

명령어를 통해 파티션 개수를 3개에서 4개로 늘리고, 리텐션 기간은 17280000ms에서 8640000ms(1일)로 변경해보자.

 

 

 

$ bin/kafka-topics.sh --bootstrap-server my-kafka:9092 --topic hello.kafka --alter --partitions 4
 
$ bin/kafka-topics.sh --bootstrap-server my-kafka:9092 --topic hello.kafka --describe
Topic: hello.kafka    TopicId: WOA3uds1Smm7jdSXN_UiIQ    PartitionCount: 4    ReplicationFactor: 1    Configs: segment.bytes=1073741824
    Topic: hello.kafka    Partition: 0    Leader: 0    Replicas: 0    Isr: 0
    Topic: hello.kafka    Partition: 1    Leader: 0    Replicas: 0    Isr: 0
    Topic: hello.kafka    Partition: 2    Leader: 0    Replicas: 0    Isr: 0
    Topic: hello.kafka    Partition: 3    Leader: 0    Replicas: 0    Isr: 0
 
$ bin/kafka-configs.sh --bootstrap-server my-kafka:9092 \
--entity-type topics --entity-name hello.kafka \
--alter --add-config retention.ms=86400000
Completed updating config for topic hello.kafka.
 
$ bin/kafka-configs.sh --bootstrap-server my-kafka:9092 \
--entity-type topics --entity-name hello.kafka --describe
Dynamic configs for topic hello.kafka are:
  retention.ms=86400000 sensitive=false synonyms={DYNAMIC_TOPIC_CONFIG:retention.ms=86400000}

 

1. --alter + --partitions : 파티션 개수를 변경한다. 토픽의 파티션을 늘릴 수 있지만 줄일 수는 없다.

2. 0번부터 시작하는 파티션이 4개로 변경되어 3번 파티션이 생성되었다. 파티션 번호는 0번부터 시작하고 1씩 늘어난다.

3. retention.ms 를 수정하기 위해서 kafka-configs.sh와 --alter, --add-config 옵션을 사용했다.