[Kafka] Docker Compose로 멀티 클러스터 설치하기

Install Kafka Multi Brokers with Docker compose

• Single Kafka Broker를 생성했었다.
• 이제는 Multi Kafka Broker를 생성해보자.
• 멀티 브로커 구성은 3대의 브로커로 구성해 볼 것이다.
• 이를 위해서 zookeeper 3개 인스턴스, kafka broker 3개 인스턴스를 올릴 것이다.

사전준비사항

• 참고로 설치는 mac에서 수행할 것이다.
• 사전 준비사항은 다음과 같다.

1. Docker Desktop 설치 : https://docs.docker.com/desktop/mac/install/ 에서 docker를 설치하자.
2. Docker dekstop 설치후 다음과 같이 버젼을 확인하자.

$ docker-compose version
Docker Compose version v2.2.3

• 예에서는 docker-compose버젼이 2 이상이어야 한다.

Docker compose 설정파일 작성하기.

• docker compose 를 이용하기 위해서는 docker-compose.yml 파일을 작성해야한다.
• docker-compose를 이용하여 실행하기 위해서는 서비스를 구성하는 설정파일을 작성해야하며, 해당 설정파일이 존재하는 디렉토리에서 docker-compose를 수행해야한다.
• docker-compose.yml 파일을 생성하고 다음과 같이 작성하자.

---
version: '2'
services:
  zookeeper-1:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_SERVER_ID: 1
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
      ZOOKEEPER_INIT_LIMIT: 5
      ZOOKEEPER_SYNC_LIMIT: 2
    ports:
      - "22181:2181"

  zookeeper-2:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_SERVER_ID: 2
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
      ZOOKEEPER_INIT_LIMIT: 5
      ZOOKEEPER_SYNC_LIMIT: 2
    ports:
      - "32181:2181"

  zookeeper-3:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_SERVER_ID: 3
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
      ZOOKEEPER_INIT_LIMIT: 5
      ZOOKEEPER_SYNC_LIMIT: 2
    ports:
      - "42181:2181"
  
  kafka-1:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper-1
      - zookeeper-2
      - zookeeper-3
    ports:
      - 29092:29092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-1:9092,PLAINTEXT_HOST://localhost:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      
  kafka-2:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper-1
      - zookeeper-2
      - zookeeper-3
    ports:
      - "39092:39092"
    environment:
      KAFKA_BROKER_ID: 2
      KAFKA_ZOOKEEPER_CONNECT: zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-2:9092,PLAINTEXT_HOST://localhost:39092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1

  kafka-3:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper-1
      - zookeeper-2
      - zookeeper-3
    ports:
      - "49092:49092"
    environment:
      KAFKA_BROKER_ID: 3
      KAFKA_ZOOKEEPER_CONNECT: zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-3:9092,PLAINTEXT_HOST://localhost:49092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1

• 위 내용을 보면 zookeeper 서비스 3개, kafka 서비스 3개를 지정했다.
• zookeeper-1, zookeeper-2, zookeeper-3 으로 서비스를 설정했다.
• kafka-1, kafka-2, kafka-3 으로 서비스를 설정했다.
• version:
  • docker-compose 버젼을 지정한다. 여기서는 2 라고 기술했다.
• services:
  • docker-compose의 경우 docker 컨테이너로 수행될 서비스들은 services 하위에 기술한다.
• zookeeper:
  • 서비스 이름을 zookeeper-1, 2, 3 로 작성했다.
  • service 하위에 작성하면 서비스 이름으로 동작한다.
• image:
  • 우리는 여기서 confluentinc/cp-zookeeper:latest 를 이용할 것이다.
  • 참고로 실전에서 사용하려면 latest 라는 태그를 사용하지 말고, 정확히 원하는 버젼을 기술해서 사용하길 추천한다.
  • latest라고 태그를 지정하면, 매번 컨테이너를 실행할때마다 최신버젼을 다운받아 실행하므로 변경된 버젼으로 인해 원하지 않는 결과를 볼 수 있다. (주의 !!!)
• environment:
  • confluentinc 는 몇가지 환경 변수를 설정할 수 있다.
  • environment 하위에 필요한 환경을 작성하자.
• ZOOKEEPER_SERVER_ID:
  • zookeeper 클러스터에서 유일하게 주키퍼를 식별할 아이디이다.
  • 동일 클러스터 내에서 이 값은 중복되면 안된다. 단일 브로커이기 때문에 이 값은 의미가 없다.
• ZOOKEEPER_CLIENT_PORT:
  • zookeeper_client_port를 지정한다. 여기서는 기본 주키퍼의 포트인 2181로 지정한다.
  • 즉 컨테이너 내부에서 주키퍼는 2181로 실행된다.
• ZOOKEEPER_TICK_TIME:
  • zookeeper이 클러스터를 구성할때 동기화를 위한 기본 틱 타임을 지정한다.
  • millisecond로 지정할 수 있으며 여기서는 2000으로 설정했으니 2초가 된다.
• ZOOKEEPER_INIT_LIMIT:
  • 주키퍼 초기화를 위한 제한 시간을 설정한다.
  • 주키퍼 클러스터는 쿼럼이라는 과정을 통해서 마스터를 선출하게 된다. 이때 주키퍼들이 리더에게 커넥션을 맺을때 지정할 초기 타임아웃 시간이다.
  • 타임아웃 시간은 이전에 지정한 ZOOKEEPER_TICK_TIME 단위로 설정된다.
  • 우리는 ZOOKEEPER_TICK_TIME을 2000으로 지정했고, ZOOKEEPER_INIT_LIMIT을 5로 잡았으므로 2000 * 5 = 10000 밀리세컨이 된다. 즉, 10초가 된다.
  • 이 옵션은 멀티 브로커에서 유효한 속성이다.
• ZOOKEEPER_SYNC_LIMIT:
  • 이 시간은 주키퍼 리더와 나머지 서버들의 싱크 타임이다.
  • 이 시간내 싱크응답이 들어오는 경우 클러스터가 정상으로 구성되어 있늠을 확인하는 시간이다.
  • 여기서 2로 잡았으므로 2000 * 2 = 4000 으로 4초가 된다.
  • 이 옵션은 멀티 브로커에서 유효한 속성이다.
• kafka
  • kafka 브로커 이름을 지정한다.
• image:
  • kafka 브로커는 confluentinc/cp-kafka:latest 를 이용하였다.
  • 역시 태그는 latest보다는 지정된 버젼을 사용하는것을 추천한다.
• depends_on:
  • docker-compose 에서는 서비스들의 우선순위를 지정해 주기 위해서 depends_on 을 이용한다.
  • zookeeper 라고 지정하였으므로, kafka는 zookeeper이 먼저 실행되어 있어야 컨테이너가 올라오게 된다.
• ports:
  • kafka 브로커의 포트를 의미한다.
  • 외부포트:컨테이너내부포트 형식으로 지정한다.
• environment:
  • kafka 브로커를 위한 환경 변수를 지정한다.
• KAFKA_BROKER_ID:
  • kafka 브로커 아이디를 지정한다. 유니크해야하며 지금 예제는 단일 브로커기 때문에 없어도 무방하다.
• KAFKA_ZOOKEEPER_CONNECT:
  • kafka가 zookeeper에 커넥션하기 위한 대상을 지정한다.
  • 여기서는 zookeeper(서비스이름):2181(컨테이너내부포트) 로 대상을 지정했다.
• KAFKA_ADVERTISED_LISTENERS:
  • 외부에서 접속하기 위한 리스너 설정을 한다.
• KAFKA_LISTENER_SECURITY_PROTOCOL_MAP:
  • 보안을 위한 프로토콜 매핑이디. 이 설정값은 KAFKA_ADVERTISED_LISTENERS 과 함께 key/value로 매핑된다.
• KAFKA_INTER_BROKER_LISTENER_NAME:
  • 도커 내부에서 사용할 리스너 이름을 지정한다.
  • 이전에 매핑된 PLAINTEXT가 사용되었다.
• KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR:
  • 이 값은 트랜잭션 상태에서 복제 계수를 지정한다. 우리는 단순하게 작업하기 위해서 복제 계수를 1로 설정했다.
• KAFKA_TRANSACTION_STATE_LOG_MIN_ISR:
  • 트랜잭션 최소 ISR(InSyncReplicas 설정) 을 지정하는 것으로 우리는 단순하게 작업하기 위해서 복제 계수를 1로 설정했다.

docker-compose 실행하기.

• 이제 docker-compose 명령을 통해 컨테이너를 실행한다.

$ docker-compose -f docker-compose.yaml up -d      

[+] Running 7/7
 ⠿ Network handson_01_default          Created                                                                                                            0.0s
 ⠿ Container handson_01-zookeeper-3-1  Started                                                                                                            1.1s
 ⠿ Container handson_01-zookeeper-2-1  Started                                                                                                            1.0s
 ⠿ Container handson_01-zookeeper-1-1  Started                                                                                                            1.1s
 ⠿ Container handson_01-kafka-3-1      Started                                                                                                            2.2s
 ⠿ Container handson_01-kafka-1-1      Started                                                                                                            2.1s
 ⠿ Container handson_01-kafka-2-1      Started         

• -f <설정파일>을 통해서 우리가 작성한 설정으로 docker-compose를 실행한다.
• up 옵션을 통해 docker-compos 를 실행한다.
• -d 옵션은 detach 모드로 컨테이너를 백그라운드로 실행하게 해준다.

docker 상태 로그 확인하가.

• 정상적으로 실행되었는지 확인을 위해서 다음과 같이 로그를 살펴 보자.

$ docker ps 

CONTAINER ID   IMAGE                              COMMAND                  CREATED              STATUS              PORTS                                         NAMES
571d813de396   confluentinc/cp-kafka:latest       "/etc/confluent/dock…"   About a minute ago   Up About a minute   9092/tcp, 0.0.0.0:29092->29092/tcp            handson_01-kafka-1-1
be0abe8e7cfc   confluentinc/cp-kafka:latest       "/etc/confluent/dock…"   About a minute ago   Up About a minute   9092/tcp, 0.0.0.0:49092->49092/tcp            handson_01-kafka-3-1
b4cff8e22cfe   confluentinc/cp-kafka:latest       "/etc/confluent/dock…"   About a minute ago   Up About a minute   9092/tcp, 0.0.0.0:39092->39092/tcp            handson_01-kafka-2-1
1a1a33b85834   confluentinc/cp-zookeeper:latest   "/etc/confluent/dock…"   About a minute ago   Up About a minute   2888/tcp, 3888/tcp, 0.0.0.0:42181->2181/tcp   handson_01-zookeeper-3-1
e5c0c44467c3   confluentinc/cp-zookeeper:latest   "/etc/confluent/dock…"   About a minute ago   Up About a minute   2888/tcp, 3888/tcp, 0.0.0.0:32181->2181/tcp   handson_01-zookeeper-2-1
41d38fe8b03c   confluentinc/cp-zookeeper:latest   "/etc/confluent/dock…"   About a minute ago   Up About a minute   2888/tcp, 3888/tcp, 0.0.0.0:22181->2181/tcp   handson_01-zookeeper-1-1

• 위 결과에서 컨테이너 id로 로그를 확인한다.

$ docker logs 41d38fe8b03c

• 주키퍼 로그를 확인해서 오류가 없이 정상 수행했는지 확인할 수 있다.
• 역시 이어서 kafka도 로그를 확인할 수 있다.

topic 생성하기.

• 이제 topic을 생성해보자.

$ docker-compose exec kafka-1 kafka-topics --create --topic my-topic --bootstrap-server kafka-1:9092 --replication-factor 3 --partitions 2

Created topic my-topic.

• docker-compose:
  • 명령어를 수행한다.
• exec:
  • 컨테이너 내에서 커맨드를 수행하도록 한다.
• kafka-1:
  • 우리가 설정으로 생성한 브로커(서비스) 이름이다.
• kafka-topics:
  • 카프카 토픽에 대한 명령을 실행한다.
• --create:
  • 토픽을 생성하겠다는 의미이다.
• --topic :
  • 생성할 토픽 이름을 지정한다.
• --bootstrap-server service:port
  • bootstrap-server는 kafka-1 브로커 서비스를 나타낸다. 이때 서비스:포트 로 지정하여 접근할 수 있다.
• --replication-factor 1:
  • 복제 계수를 지정한다.
  • 여기서는 1로 지정했다.
• --partition:
  • 토픽내에 파티션 개수를 지정한다.

생성된 토픽 확인하기.

$ docker-compose exec kafka-1 kafka-topics --describe --topic my-topic --bootstrap-server kafka-1:9092 

Topic: my-topic TopicId: 3McNxmotQaqMwLZrig73Yg PartitionCount: 2       ReplicationFactor: 3    Configs: 
        Topic: my-topic Partition: 0    Leader: 3       Replicas: 3,1,2 Isr: 3,1,2
        Topic: my-topic Partition: 1    Leader: 1       Replicas: 1,2,3 Isr: 1,2,3

• docker-compose:
  • 명령어를 수행한다.
• exec:
  • 컨테이너 내에서 커맨드를 수행하도록 한다.
• kafka-1:
  • 우리가 설정으로 생성한 브로커(서비스) 이름이다.
• kafka-topics:
  • 카프카 토픽에 대한 명령을 실행한다.
• --describe:
  • 생성된 토픽에 대한 상세 설명을 보여달라는 옵션이다.
• --topic :
  • 생성한 토픽 이름을 지정한다.
• --bootstrap-server service:port
  • bootstrap-server는 kafak-1 브로커 서비스를 나타낸다. 이때 서비스:포트 로 지정하여 접근할 수 있다.
• 결과로 kafka-1브로커의 토픽이름, 아이디, 복제계수, 파티션, 리더, 복제정보, isr 등을 확인할 수 있다.
• 복제 계수는 3으로 지정되었다.
• 그리고 파티션은 2개로 0, 1 이 있다.
• Leader의 결과를 통해서 해당 파티션의 리더가 어떤 브로커인지 알려준다.
• Replicas 는 데이터 복제에 대해서 알려준다.
• Isr: In sync replica 에 대해서 알려준다. (동기화된 복제본임을 알려준다.)

컨슈머 실행하기.

• 이제 브로커를 만들었으니 브로커로 부터 메시지를 수신해 보자.
• 컨슈머를 먼저 실행하는 이유는, 일반적으로 컨슈머가 메시지를 컨슘하려고 대기하고 있고, 송신자가 메시지를 생성해서 보내기 때문이다.

$ docker-compose exec kafka-1 bash
[appuser@571d813de396 ~]$ kafka-console-consumer --topic my-topic --bootstrap-server kafka-1:9092

• 우선 docker-compose exec kafka-1 bash 를 통해서 컨테이너 내부의 쉘로 접속한다.
• 이후 kafka-console-consumer 를 이용하여 컨슘한다.
• 역시 컨슘할 토픽을 지정하고, 브로커를 지정하기 위해서 --bootstrap-server 를 이용했다.

프로듀서 실행하기

• 이제 컨슈머가 메시지를 수신대기 하므로, 프로듀서를 통해 메시지를 전송하자.

$ docker-compose exec kafka-1 bash 
[appuser@571d813de396 ~]$ kafka-console-producer --topic my-topic --broker-list kafka-1:9092

>hello
>this is producer

컨슈머 결과 보기

• 컨슈머 결과 타이핑한 메시지가 전달됨을 확인할 수 있다.

hello
this is producer

docker-compose 컨테이너 내리기

$ docker-compose down

• 위 명령을 통해서 docker-compose 컨테이너 인스턴스를 내릴 수 있다.

WrapUp

• 지금까지 kafka multi broker 모드로 docker-compose를 이용하여 실행해 보았다.
• kafka는 zookeeper를 통해서 상태와 브로커의 오케스트레이션을 수행한다.
• 그리고 kafka-1, 2, 3라는 서비스 이름으로 브로커를 생성했다.
• 메시지를 주고 받기 위해서 토픽을 생성했고, 프로듀서와 컨슈머를 활용하여 메시지를 전송/수신해 보았다.
• 멀티 브로커를 이용하여 클러스터를 구성했다. 이는 일반적인 멀티 브로커 클러스터로 개발을 진행할때 docker-compose 를 활용하는 방안은 클러스터로 프로젝트를 수행해볼 수 있는 좋은 가이드라인이 될 것이다.