카프카 잘 알고 사용하기 - 카프카 클러스터 구축하기(with Docker, docker-compose)

개요

과거 프로젝트 진행했을 때 카프카를 단일 브로커를 사용해서 구축하였다. 이는 카프카가 제공하는 고가용성의 장점을 충분히 활용하지 못한 사례로, 단일 브로커에 장애가 발생할 경우 전체 서비스에 심각한 영향을 미칠수 있는 구조였다.

 

이러한 문제를 개선하기 위해 이번에는 과거 kafka설정을 기반으로 재구성하여 kafka클러스터를 단일 브로커가 아닌 다중 브로커(3개) 환경으로 구축하여 리플리케이션이 가능하도록 구축해보겠다.

 

docker-compose.yml

version: '3.8'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_SERVER_ID: 1
      ZOOKEEPER_TICK_TIME: 2000
      ZOOKEEPER_CLIENT_PORT: 2181
    networks:
      - kafka-network

  kafka1:
    image: confluentinc/cp-kafka:latest
    container_name: kafka1
    ports:
      - "9092:9092"   # 클라이언트 통신 포트
      - "19092:19092" # 브로커 간 통신 포트
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT, EXTERNAL:PLAINTEXT    # 리스너 이름:PLAINTEXT(평문 - 보안없이)
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka1:19092, EXTERNAL://localhost:9092  # 리스너 주소(카프카 내부 브로커간 통신 kafka1:19092, 외부 접근 localhost:9092)
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL                                      # 카프카 브로커간 통신에 사용
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
    depends_on:
      - zookeeper
    networks:
      - kafka-network

  kafka2:
    image: confluentinc/cp-kafka:latest
    container_name: kafka2
    ports:
      - "9093:9093"   # 클라이언트 통신 포트
      - "19093:19093" # 브로커 간 통신 포트
    environment:
      KAFKA_BROKER_ID: 2
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT, EXTERNAL:PLAINTEXT    # 리스너 이름:PLAINTEXT(평문 - 보안없이)
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka2:19093, EXTERNAL://localhost:9093  # 리스너 주소(카프카 내부 브로커간 통신 kafka1:19093, 외부 접근 localhost:9093)
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL                                      # 카프카 브로커간 통신에 사용
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
    depends_on:
      - zookeeper
    networks:
      - kafka-network

  kafka3:
    image: confluentinc/cp-kafka:latest
    container_name: kafka3
    ports:
      - "9094:9094"   # 클라이언트 통신 포트
      - "19094:19094" # 브로커 간 통신 포트
    environment:
      KAFKA_BROKER_ID: 3
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT, EXTERNAL:PLAINTEXT    # 리스너 이름:PLAINTEXT(평문 - 보안없이)
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka3:19094, EXTERNAL://localhost:9094  # 리스너 주소(카프카 내부 브로커간 통신 kafka1:19094, 외부 접근 localhost:9094)
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL                                      # 카프카 브로커간 통신에 사용
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
    depends_on:
      - zookeeper
    networks:
      - kafka-network

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    container_name: kafka-ui
    ports:
      - 8081:8080
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka1:19092, kafka2:19093, kafka3:19094
      KAFKA_CLUSTERS_0_ZOOKEEPER: zookeeper:2181
    networks:
      - kafka-network


networks:
  kafka-network:    # 동일한 네트워크 구성
    driver: bridge

 

 

카프카 브로커는 총 3개 구성하였으며 클라이언트 통신용포트(9092, 9093, 9094)와 브로커간 통신 포트(19092, 19093, 19094)로 구성하였다.

 

kafka-ui는 kafka클러스터의 상태를 web UI(접속 localhost:8081)로 쉽게 모니터링하기 위해 추가하였다. 필요없다면 해당 부분은 제거해도 좋다.

 

실행

docker-compose up -d

 

총 5개의 컨테이너가 정상작동해야한다.

 

테스트(카프카 CLI)

호스트 머신에서 토픽생성이나 메시지를 보내려면 Kafka CLI를 설치해야한다.

 

Kafka CLI로 테스트 하기

Apache Kafka

해당 링크에 접속하여 Binary download에서 다운받아 압축을 해제한다.

 

Kafka CLI 경로 설정(Linux/Mac)

kafka CLI 도구를 간편하게 사용하기 위해 압축푼 디렉토리의 bin경로를 시스템 환경 변수에 추가한다.

export PATH=$PATH:/path/to/kafka/bin   # 자신의 디렉토리의 bin 경로로 맞게 변경

 

토픽 생성 및 메시지 테스트

실습용 테스트 토픽을 생성한다. 이름은 test-topic 파티션은 1개, 리플리케이션 팩터 수는 3개로 설정

kafka-topics.sh --create \
  --bootstrap-server localhost:9092 \
  --replication-factor 3 \
  --partitions 1 \
  --topic test-topic

 

잘 생성 됬는지 확인하려면 다음 명령어로 토픽 상세정보를 확인할 수 있다.

[토픽 상세정보 확인]
--describe --bootstrap-server localhost:9092 --topic test-topic

 

[출력결과]
Topic: test-topic	TopicId: nKfLDWTvQaW9TrxUnlAjEg	PartitionCount: 1	ReplicationFactor: 3	
Configs:
	Topic: test-topic	Partition: 0	Leader: 2	Replicas: 2,1,3	Isr: 2,1,3

Topic	토픽 이름
TopicId	현재 토픽 고유ID
ReplicationFactor	리플리케이션 팩터 수(3)
Configs:Partition	파티션 번호(0)
Configs:Leader	해당 파티션(0)의 리더로 할당된 브로커
Configs:Replicas	파티션(0)의 복제본이 있는 브로커 ID
Configs:Isr(In-Sync Replicas)	리더 브로커와 동기화 상태를 유지하고 있는 브로커 ID

 

[토픽 컨슈머 생성]
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic

[토픽 메시지 전송]
kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test-topic
> 안녕하세요!

프로듀서가 "안녕하세요" 메시지를 보내면 컨슈머는 메시지를 받을 수 있게된다.

 

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic
> 안녕하세요!

 

 

Kafka 브로커 장애 시 리더 재선출 과정 확인

해당 메시지가 정상적으로 주고 받는지는 확인하였으니, 현재 test-topic의 파티션0의 리더인 브로커 2가 장애가 발생했을 경우에도 새로운 리더를 선출해 정상적으로 동작하는지 확인하기 위해 의도적인 장애 상황 연출을 위해 브로커 2를 종료해보겠다.

현재 test-topic의 파티션0의 리더 종료

 

 

여태 배운 내용을 돌이켜 보면 리더가 장애가 발생했으므로 ISR(InSyncReplica)그룹에 속한 팔로워 중 하나가 리더로 선출될 것이라고 예상할 수 있다.

 

[토픽 상세정보 확인]
--describe --bootstrap-server localhost:9092 --topic test-topic

[출력결과]
Topic: test-topic	TopicId: nKfLDWTvQaW9TrxUnlAjEg	PartitionCount: 1	ReplicationFactor: 3	
Configs:
	Topic: test-topic	Partition: 0	Leader: 1	Replicas: 2,1,3	Isr: 1,3

ISR내에 새로운 리더로 브로커 1이 선택되었고, 브로커 2는 ISR 그룹에서 추방되었음을 확인할 수 있다.

 

[토픽 메시지 전송]
kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test-topic
> 새로운 리더 선출 후 메시지 송신

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic
> 새로운 리더 선출 후 메시지 송신

리더 변경 후에도 메시지가 정상적으로 송수신되었음을 확인할 수 있다. 이는 Kafka 클러스터의 고가용성(High Availability)과 장애 복구(Leader Election) 메커니즘이 올바르게 작동하고 있음을 의미한다.

 

 

 

마무리

Kafka 클러스터를 Docker를 활용해 구축하고, 의도적으로 장애 상황을 연출하여 Kafka의 고가용성과 장애 복구 메커니즘이 효과적으로 작동함을 확인했다. 이를 통해 Kafka가 데이터 손실과 서비스 중단을 방지하는 데 얼마나 중요한 역할을 하는지 실질적으로 이해할 수 있었다.