Kafka Kafka명령어, local 환경에 Kafka Cluster 구축

Kafka 설치

아래 링크에 들어가 Binary downloads를 하면됩니다. Version은 자신의 환경 및 학습하고자 하는 버전에 맞게 다운받으시면 됩니다. (저는 기존에 깔아둔 2.8.2 version 입니다.)

https://kafka.apache.org/downloads

Apache Kafka

 

Broker 3대 실행하기

Broker를 실행해주기 위해 설정파일을 작성해줍니다. 설치한 Kafka 폴더안에 config 폴더로 이동하여 server.properties 설정파일을 작성해줍니다. 

 

설정 파일 작성

server.properties는 Broker에 대한 설정 파일입니다.
우리는 Broker 3대를 만들기 때문에 server.properties 파일을 복사하여 파일을 두 개 더 만들도록 하겠습니다.
(파일의 이름을 server1.properties, server2.properties로 바꿔주었습니다.)

server.properties

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=0

#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://:9092

# A comma separated list of directories under which to store log files
log.dirs=/tmp/kafka-logs

 

server1.properties

broker.id=1
listeners=PLAINTEXT://:9093
log.dirs=/tmp/kafka-logs1

server2.properties

broker.id=2
listeners=PLAINTEXT://:9094
log.dirs=/tmp/kafka-logs2

 

 

실행(작동)

zookeeper와 Broker들을 각각 shell에서 실행시켜 줍니다.

💁Windows, Mac 환경에서 Kafka 명령어??
실행하기에 앞서 아래 예제는 Windows 환경입니다.
Windows 환경에선  \bin\windows\zookeeper-server-start.bat 이지만
Mac 환경에서는 \bin\zookeeper-server.start.sh 입니다.

 

zookeeper 실행

D:\tool\kafka> .\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties

 

Broker 실행

D:\tool\kafka> .\bin\windows\kafka-server-start.bat .\config\server.properties
D:\tool\kafka> .\bin\windows\kafka-server-start.bat .\config\server1.properties
D:\tool\kafka> .\bin\windows\kafka-server-start.bat .\config\server2.properties

 

Topic 생성

D:\tool\kafka> .\bin\windows\kafka-topics.bat --create --topic topic1 --partitions 3 --replication-factor 3 --bootstrap-server localhost:9093

• Topic 생성을 할 때 port 9093로 작성하여 Broker 1을 지정하였습니다. 사실 Broker 1을 연결하여도, Broker 0을 선택하여도, Broker 0,1,2를 모두 선택하여도 상관이 없습니다. Kafka는 Cluster로 구성되어 있고, zookeeper에 의해  관리되고 있기 때문에 어떤 Broker에 Topic 생성을 요청해도 Cluster에 의해 Topic 생성이 관리됩니다.
• --create: topic 생성 위한 옵션
• --topic topic1: topic1라는 이름의 주제(topic)
• --partitions: 생성할 Topic의 Partition 개수를 지정하는 옵션
• --replication-factor: 생성할 Topic의 Replication Factor를 지정합니다. 이 경우, 각 파티션에 3개의 복제본을 생성합니다.
• --bootstrap-server: Kafka 클러스터에 연결하기 위한 부트스트랩 서버를 지정합니다. 옛날에는 -zookeeper 옵션을 통해 zookeeper 서버의 주소를 지정하여서 Kafka 클러스터에 연결하였으나, Kafka 2.8.0 버전부터 zookeeper에 대한 의존성이 감소하고, 대부분의 메타데이터 관리가 브로커 자체에 내장됨에 따라 --bootstrap-server를 사용하여 Kafka 클러스터에 연결하는 것이 권장됩니다. 

 

Topic 확인

.\bin\windows\kafka-topics.bat --topic topic1 --bootstrap-server localhost:9094 --describe

• 위와 같은 이유로 어떤 Broker를 지정하여도 같은 결과가 나옵니다.

 

broker 서버의 Topic List 확인

D:\tool\kafka> .\bin\windows\kafka-topics.bat --list --bootstrap-server localhost:9094

 

 

Producer 실행

D:\tool\kafka> .\bin\windows\kafka-console-producer.bat --topic topic1 --bootstrap-server localhost:9092,localhost:9093,localhost:9094

• 위와 같은 이유로 어떤 Broker를 지정하여도 같은 결과가 나옵니다. 위 예제 코드에서는 모든 브로커를 지정해준 코드입니다.

 

Consumer 실행, Message 읽기

D:\tool\kafka> .\bin\windows\kafka-console-consumer.bat --topic topic1 --from-beginning --group first --bootstrap-server localhost:9092,localhost:9093,localhost:9094

• 위와 같은 이유로 어떤 Broker를 지정하여도 같은 결과가 나옵니다.
• --group first:
  Consumer 그룹 이름을 first로 지정하였습니다. Consumer에 그룹을 지정함으로써 동일한 Topic의 Partition을 여러 Consumer가 분산해서 소비할 수 있습니다. 이렇게 그룹으로 지정한 Consumer들은 Partition에서 메시지를 읽을 때 Consumer가 읽은 Offset 정보를 추적합니다. 이를 current offset이라 합니다. 같은 Consumer 그룹 내의 Consumer들은 동일한 Topic의 Partition에 대해 동일한 current offset을 공유하게 되어 동일한 메시지를 중복없이 효율적으로 처리합니다.
• --from-beginning:
  Topic이 갖고 있는 첫 레코드부터 읽는다는 의미입니다. 해당 옵션이 없는 경우 consumer가 실행된 이후 생성된 메시지만 읽습니다. (group이 지정되어 있지 않다면 항상 첫 레코드부터 읽습니다. current offset은 group명으로 관리되기 때문에 group을 지정하지 않으면 offset 정보가 없게되고 첫 레코드부터 읽게 됩니다.) 

 

그 외 Kafka 명령어들

Topic의 Partition 개수 변경

D:\tool\kafka> .\bin\windows\kafka-topics.bat --alter --topic topic1 --partitions 4 --bootstrap-server localhost:9093

• --alter: 변경 옵션
• --partitions 개수는 기존 partition의 개수보다 많게만 변경할 수 있습니다.

💁 partition 추가 시 주의해야할 점!!
서비스 운영중인 Kafka Topic이라면 partition을 추가하는 것을 신중하게 결정해야합니다. 
partition이 추가되면 메시지 재배치가 발생하게 됩니다. 이는 네트워크 사용량을 증가시켜 성능에 영향을 끼칠 수 있습니다. 또한 partition을 저장하는 방식이 메시지 Key 기반인 경우 메시지 유실 가능성도 있습니다.
따라서, Topic 생성 시 처음부터, partition 개수를 여유있게 설정하는 것이 바람직합니다.

 

 

Topic 삭제

D:\tool\kafka> .\bin\windows\kafka-topics.bat --delete --topic topic1 --bootstrap-server localhost:9093

• --delete: 삭제 옵션
• Topic을 삭제하기 위해서 server.properties 파일에서 delete.topic.enable=true 설정을 해주고 서버 재시작 해줘야합니다.

 

다양한 엔티티(Topic, Broker, Consumer Group 등) 설정 변경

D:\tool\kafka> .\bin\windows\kafka-configs.bat --alter --entity-type topics --entity-name topic1 --add-config retention.ms=86400000 --bootstrap-server localhost:9093

• --entity-type topics: 설정을 변경할 엔티티의 타입을 지정합니다. 여기서는 토픽(topics)을 지정하였습니다.
• --entity-name topic1: 설정을 변경할 대상의 이름을 지정합니다. 여기서는 topic1이라는 이름의 토픽을 지정하였습니다.
• --entity-type topics --entity-name topic1 대신 --topic topic1로 대체할 수 있습니다. 토픽 설정 변경 시 이를 통해 간편하고 직관적이게 명령어를 작성할 수 있습니다.
• --add-config retention.ms=86400000: 설정 변경하는 옵션입니다. 여기서는 메시지의 보존 시간을 24시간으로 설정합니다.

Producer 옵션

D:\tool\kafka> .\bin\windows\kafka-console-producer.bat --topic topic1 --request-required-acks 1 --bootstrap-server localhost:9092,localhost:9093,localhost:9094

--request-required-acks:

• 0 (No acknowledgment)
  프로듀서가 메시지를 전송한 후 브로커로부터 어떤 확인도 받지 않습니다. 이 설정은 가장 빠른 속도로 메시지를 전송할 수 있지만, 브로커에 의해 처리되지 않은 메시지의 손실이 발생할 수 있습니다.
• 1 (Leader acknowledgment)
  프로듀서가 메시지를 전송한 후 메시지를 처리한 리더 파티션에게 확인을 받습니다. 이 경우 리더 파티션만 확인을 보내기 때문에 전송 속도는 빠르지만, 메시지 손실은 아직 발생할 수 있습니다.
• -1 (All acknowledgment) 또는 all
  프로듀서가 메시지를 전송한 후 모든 ISR(In-Sync Replicas) 복제본에게 확인을 받을 때까지 대기합니다. 모든 ISR 복제본이 메시지를 확인하고 처리했을 때만 성공적으로 확인을 받습니다. 이 설정은 가장 높은 신뢰성을 제공하지만, 전송 시간이 다소 지연될 수 있습니다.

--message-send-max-retries:

Kafka 프로듀서가 메시지를 전송하는 동안 재시도할 최대 횟수를 지정합니다.

 

--max-messages:

Kafka 프로듀서가 전송할 최대 메시지 수입니다. 메시지 수를 제한하려는 경우 사용합니다.\

 

 

Consumer 옵션

D:\tool\kafka> .\bin\windows\kafka-console-consumer.bat --topic topic1 --property print.key=true  --property key.separator="-" --bootstrap-server localhost:9092 --from-beginning --group first

• --property print.key=true:
  이 옵션은 메시지의 키(key)를 출력할지 여부를 설정합니다. 여기서는 true로 설정되어 있으므로 메시지의 키가 출력됩니다.
• --property key.separator="-":
  출력되는 메시지의 키와 값 사이의 구분자를 지정합니다. 여기서는 "-"로 설정되어 있으므로 키와 값 사이에 "-"가 사용됩니다.