안녕하세요. Kafka 한국 사용자 모임인 kafka KRU 에서 진행한 스터디 모임에서 'kafkaAdminClient API를 이용한 kafka Managing 서비스 구축’이란 주제로 프로젝트를 진행한 이진환, 이상헌 입니다.

기존에는 Kafka CLI를 이용하여 kafka Object에 관한 정보를 조회하고 만들었는데요. 입력 인자 중 Kafka cluster host:port 를 입력해 줘야 하는 등 불편한 점이 있었습니다.

저희는 kafka Object를 매니징 할 수 있는 KafkaAdminClient API를 이용해 사용자가 간편하게 kafka 관련 인자를 조회하고 관리할 수 있는 웹 애플리케이션을 제작했습니다.

1. KafkaAdminClient란?

Kafka Object(topics, brokers, acls, consumers 등)의 정보를 조회하고 매니징하는데 사용할 수 있는 class입니다.

Topic을 예로 들면, 아래와 같은 methods를 사용해서 Topic을 관리할 수 있습니다.

• createTopics // 토픽 생성

• deleteTopics // 토픽 삭제

• listTopics // 토픽 목록조회

• describeTopics // 토픽 상세조회

[공식문서 url]: https://kafka.apache.org/33/javadoc/index.html?org/apache/kafka/clients/admin/AdminClient.html

2. 프로젝트 상세

• 주제 : kafkaAdminClient API를 이용한 kafka Managing 서비스

• Github : https://github.com/jinhwan2/kafka-admin-server

• 사용한 프레임워크 :

[Kafka]  
kafkaClientsVersion=3.3.1
https://github.com/conduktor/kafka-stack-docker-compose 
(Using Multiple Zookeeper / Multiple Kafka case)

[Spring Boot]
SpringBootVersion=2.7.5

[lombok]
lombokVersion=1.18.24

[swagger]
swaggerVersion=2.9.2

• Controller : Client의 요청을 받았을 때 그 요청에 대해 실제 동작을 수행하는 Service를 호출합니다. 해당 요청에 대한 URI 경로를 Mapping 하여 broker에 관한 요청인지, topic에 관한 요청인지 구분할 수 있게 해줍니다.

• Service : Controller의 요청을 받아, 알맞은 정보를 가공하여 Controller에게 재전달합니다. kafkaAdminClient API를 사용하여 데이터를 가공하는 모듈입니다.

• DTO : 계층 간의 데이터 교환을 위한 객체입니다. Kafka에서 사용하는 Node, Topic, ConsumerGroup과 같은 자료구조를 담고 있습니다.

2-1. 사용한 API methods

<각 모듈 별로 사용한 API 목록>

• broker-controller

  • describeConfigs(config) // Config 상세 조회

  • incrementalAlterConfigs(config) // Config 업데이트

• cluster-controller

  • describeCluster( ) // Cluster 상세 조회

    • authorizedOperations( ) // ACL 정보

    • clusterId( ) // Cluster Id 정보

    • nodes( ) // node 정보 (org.apache.kafka.common.Node)

    • controller( ) // Controller node 정보

• consumer-group-controller

  • listConsumerGroups( ) // Consumer groups list 조회

  • describeConsumerGroups(Collection<String> groupId) // Consumer groups 상세 조회

  • listConsumerGroupOffsets(String groupId) // Consumer groups list offset 조회

  • alterConsumerGroupOffsets(String groupId, Map<TopicPartition,OffsetAndMetadata> Offsets) // Consumer groups list offset 변경

• topic-controller

  • createTopics(Collection newTopics) // topic 생성

  • listTopics( ) // topic list 조회

  • describeTopics(Collection topicNames) // topic 상세 조회

  • deleteTopics(Collection topics) // topic 삭제

  • describeConfigs(config) // Config 상세 조회

  • incrementalAlterConfigs(config) // Config 업데이트

[About Configs() methods]

매개변수가 config인 methods에 주목할 필요가 있습니다.

broker-controller 모듈과 topic-controller 모듈에서 describeConfigs와 incrementalAlterConfigs method를 사용하는데요.

현재 Kafka(3.3.1 version) 에서의 describeConfigs mothods 상세는 아래와 같습니다.

DescribeConfigsResult **describeConfigs**(Collection<ConfigResource> resources,
DescribeConfigsOptions options)
---------------------------------------------------------------------------
Get the configuration for the specified resources.

**Parameters:**
resources - The resources (topic and broker resource types are currently supported)
options - The options to use when describing configs

**Returns:**
The DescribeConfigsResult

매개변수 Collection<>의 데이터 타입인 ConfigResource class 내부에는 ConfigResource.Type 이라는 Enum 이 정의되어 있습니다. [Enum에 대하여]

Enum ConfigResource.Type에는 BROKER, BROKER_LOGGER, TOPIC, UNKNOWN이 정의되어있는데, Broker와 Topic에 관심이 있는 저희는 이 두 개의 ConfigResource.Type을 이용하여 정보를 얻어오겠습니다.

<사용 Example>

package kafkakru.admin.service;

@Service
public class TopicService extends AbstractKafkaAdminClientService{
    private final ConfigService configService;
    private static final ConfigResource.TypeRESOURCE_TYPE= ConfigResource.Type.TOPIC;
...
}

2-2. 비교해보기(CRUD Example)

2-2-1. Create Topic

<Kafka CLI 사용해서 직접 topic 만드는 Example>

$ bin/kafka-topics.sh --create \
    --bootstrap-server <hostname>:<port> \
    --topic <topic-name> \
    --partitions <number-of-partitions> \
    --replication-factor <number-of-replicating-servers>

<REST API 이용해서 topic 만드는 Example>

• 아래와 같은 POST request를 보냅니다.

curl -X POST "http://localhost:3000/v1/topics?numPartitions=<number-of-partitions>&replicationFactor=<number-of-replicating-servers>&topicName=<topic-name>"

• Topic 생성 성공 시, 아래 메시지를 응답으로 받습니다.

{"data":true,"status":200,"failMessage":null}

• Swagger 웹 화면에서 사용 예시

2-2-2. Read configuration

<Kafka CLI 사용해서 직접 topic configuration 조회하는 Example>

> bin/kafka-configs.sh --describe \
   --bootstrap-server <hostname>:<port> \
   --entity-type <entity-type> \
   --entity-name <entity-name>

<REST API 이용해서 topic configuration 조회하는 Example>

• 아래와 같은 POST request를 보냅니다.

curl -X GET "http://localhost:3000/v1/topic-configs/<topic-name>"

Topic 생성 성공 시, 아래 메시지를 응답으로 받습니다.

• 예시

  {
    "data": [
      {
        "key": "compression.type",
        "value": "producer"
      },
    ** < .. 생략 ..>**
      {
        "key": "delete.retention.ms",
        "value": "86400000"
      },
      {
        "key": "segment.ms",
        "value": "604800000"
      }
    ],
    "status": 200,
    "failMessage": null
  }

→ retention 기간, cleanup.policy 등을 간편히 조회할 수 있습니다.

• Swagger 웹 화면에서 사용 예시

2-2-3. Update Topic

<Kafka CLI 사용해서 직접 topic config 수정하는 Example>

• 방법 1) server.properties에서 직접 설정 값을 바꿔준다.

• 방법 2) 아래 명령어를 입력한다.

./bin/kafka-topics.sh --alter \
--bootstrap-server <hostname>:<port> \ 
‑‑topic <topic-name> \
‑‑config <key>=<value>

<REST API 이용해서 topic config 수정하는 Example>

• getTopicConfig를 이용해서 아래 메시지를 응답으로 받습니다.

• 예시

  {
    "data": [
  
      {
        "key": "retention.ms",
        "value": "604800000"
      },
      {
        "key": "flush.messages",
        "value": "9223372036854775807"
      },
  
           ** <.. 생략 .. >**
      {
        "key": "message.format.version",
        "value": "3.0-IV1"
      },
      {
        "key": "min.compaction.lag.ms",
        "value": "0"
      } 
    ],
    "status": 200
  }

• 만약, 토픽의 데이터를 유지하는 시간을 더 길게 조정하고 싶다면, 아래 파라미터를 바꿔야 합니다.

{
      "key": "retention.ms",
      "value": "604800000"
}

• 현재 값은 1주(604800000 ms)인데, 이를 2배 늘려서 2주(1209600000 ms)로 변경하고자 합니다.

• 아래와 같은 PUT request를 보냅니다.

curl -X PUT "http://localhost:3000/v1/topics/<topic-name>" -H "accept: */*" -H "Content-Type: application/json" -d "{ \"config\": { \"retention.ms\": \"1209600000\" }}"

• topic config 업데이트 후

{
      "key": "retention.ms",
      "value": "1209600000"
}

값이 바뀐 것을 확인할 수 있습니다. :)

• Swagger 웹 화면에서 사용 예시

2-2-4. Delete Topic

<Kafka CLI 사용해서 직접 topic 삭제하는 Example>

• 사전 설정

  server.properties에서 delete.topic.enable = true 로 설정 값을 바꿔줍니다.

> bin/kafka-topics.sh --delete \
   --bootstrap-server <hostname>:<port> \
   --topic <topic-name> \

<REST API 이용해서 topic 삭제하는 Example>

• GetAllTopicNames 결과 값 - Topic lists

{"data":["123","test3","test","exam-topic","hello2","test-00","test1"],"status":200}

• 아래와 같은 Delete request를 보냅니다.

curl -X DELETE "http://localhost:3000/v1/topics/123"

• ‘123’ topic 삭제 후, GetAllTopicNames 결과 값 - Topic lists

{"data":["test3","test","exam-topic","hello2","test-00","test1"],"status":200}

• Swagger 웹 화면에서 사용 예시

3. 마무리

KafkaAdminClient API와 Spring을 이용하여 Kafka Object(topic, cluster, broker 등)를 managing하는 Web 어플리케이션을 제작했습니다.

bash terminal에서 Kafka CLI를 사용하여 managing하는 방법과 Web UI system에서 managing하는 방법을 비교해봤습니다.

bootstrap-server ip를 입력하지 않아도 되고, 버튼 클릭으로 작동한다는 점이 CLI보다 훨씬 간편하다고 생각합니다.

Kafka cluster를 운영하다보면, 일부 config 값을 이용하여 특정 동작을 수행하고 싶을 때가 있을 것이고, Kafka CLI에서 지원을 하지 않은 정보들을 확인하고 싶을 때가 있을텐데,

(예시. Kafka 특정 토픽이 비었는지 확인하기, 출처 : https://honeyinfo7.tistory.com/300)

이런 상황에서 KafkaAdminClient API를 사용하여 원하는 code를 작성하려 한다면, Kafka 초급자에서 중급자로 한 단계 성장 중이실 겁니다! :D

4주간 Kafka KRU 스터디를 통해,

1. Kafka의 탄생부터 기초까지 + AWS에서 Kafka Cluster 구축

2. 어디서도 잘 알려주지 않는 zookeeper에 대해서

3. Kafka의 내부 동작원리(Producer, Broker, Consumer, ConsumerGroup)

4. 다양한 Replication Case 고려해보기

5. 운영을 위한 kafka WEB UI 어플리케이션 및 monitoring Tool

등등 kafka 전반을 다뤄보았습니다.

해당 스터디를 기획해주시고 이끌어주신 고승범님과 해당 프로젝트 개발을 진행해주신 이진환님께 매우 감사드립니다.

다른 Kafka KRU 맴버들도 4주간 고생하셨습니다.

감사합니다.