[Kafka] Docker Compose - Single Broker 사용으로 통신해보기

준비사항

docker-compose 버전은 2 이상이어야 한다.

$ docker-compose version
 
Docker Compose version v2.13.0

 

 

 

---

 

 

 

Docker compose 설정파일 작성하기

* docker compose 를 이용하기 위해서는 docker-compose.yml 파일을 작성해야 한다.

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

* docker-compose-single.yml 파일을 생성하고 다음과 같이 작성한다.

 

주석 있는 버전

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

 

 

주석 없는 버전

version: '2'
 
services:
  zookeeper:
    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"
 
  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    ports:
      - "29092:29092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0

 

 

 

 

---

 

 

 

docker-compose 실행하기

$ docker-compose -f docker-compose.yml up -d
 
[+] Running 3/3
 ⠿ Network handson_01_default        Created                                             0.0s
 ⠿ Container handson_01-zookeeper-1  Started                                             0.9s
 ⠿ Container handson_01-kafka-1      Started                                             1.7s

* -f <설정파일>을 통해서 작성한 설정으로 docker-compose를 실행한다.

* up 옵션을 통해 docker-compos 를 실행한다

* -d 옵션은 detach 모드로 컨테이너를 백그라운드로 실행하게 해준다.

 

 

 

---

 

 

 

docker 상태 로그 확인하기

$ docker ps
 
CONTAINER ID   IMAGE                              COMMAND                  CREATED         STATUS              PORTS                                         NAMES
5a0bbef060c8   confluentinc/cp-kafka:latest       "/etc/confluent/dock…"   4 minutes ago   Up About a minute   9092/tcp, 0.0.0.0:29092->29092/tcp            kafka-docker-kafka-1
d369bb6e4621   confluentinc/cp-zookeeper:latest   "/etc/confluent/dock…"   4 minutes ago   Up 4 minutes        2888/tcp, 3888/tcp, 0.0.0.0:22181->2181/tcp   kafka-docker-zookeeper-1

 

* 위 결과에서 컨테이너 id들로 로그를 확인할 수 있다.

$ docker logs 5a0bbef060c8
 
$ docker logs d369bb6e4621

 

* 주키퍼 로그와 카프카 로그를 확인해서 오류가 없이 정상 수행되는지 확인할 수 있다.

 

 

 

---

 

 

 

topic 생성하기

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

 

* docker-compose: 명령어를 수행한다.

* exec: 컨테이너 내에서 커맨드를 수행하도록 한다.

* kafka: 설정으로 생성한 브로커(서비스) 이름이다.

* kafka-topics: 카프카 토픽에 대한 명령을 실행한다.

* --create: 토픽 생성한다.

* --topic: 생성할 토픽 이름을 지정한다.

* --bootstrap-server service:port: bootstrap-server는 kafka 브로커 서비스를 나타낸다. 이때 서비스:포트 로 지정하여 접근할 수 있다.

* --replication-factor 1: 1) 복제 계수를 지정한다. 2) 여기서는 1로 지정했다.

* --partition: 토픽내에 파티션 개수를 지정한다.

 

 

 

---

 

 

 

생성된 토픽 확인하기

$ docker-compose exec kafka kafka-topics --describe --topic my-topic --bootstrap-server kafka:9092
 
Topic: my-topic    TopicId: nRlJJQWLTlCj4xOQQllbcQ    PartitionCount: 1    ReplicationFactor: 1    Configs:
    Topic: my-topic    Partition: 0    Leader: 1    Replicas: 1    Isr: 1

* docker-compose: 명령어를 수행한다.

* exec: 컨테이너 내에서 커맨드를 수행하도록 한다.

* kafka: 우리가 설정으로 생성한 브로커(서비스) 이름이다.

* kafka-topics: 카프카 토픽에 대한 명령을 실행한다.

* --describe: 생성된 토픽에 대한 상세 설명을 보여달라는 옵션이다

* --topic: 생성한 토픽 이름을 지정한다.

* --bootstrap-server service:port: bootstrap-server는 kafka 브로커 서비스를 나타낸다. 이때 서비스:포트 로 지정하여 접근할 수 있다.

* 결과로 토픽이름, 아이디, 복제계수, 파티션, 리더, 복제정보, isr 등을 확인할 수 있다.

 

 

 

---

 

 

 

컨슈머 실행하기

브로커를 생성하였으니 메시지를 수신할 준비가 필요하다.

컨슈머를 먼저 실행하는 이유는 일반적으로 컨슈머가 메시지를 컨슘하려고 대기하고 있고, 송신자가 메시지를 생성해서 보내기 때문이다.

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

 

docker-compose exec kafka bash 를 사용해서 컨테이너 내부의 shell로 접속한다. 

접속 후에는 kafka-console-consumer 를 이용하여 컨슘한다. 컨슘할 때 컨슘할 토픽을 지정하고, 브로커를 지정하기 위해 --bootstrap-server 를 이용했다.

 

 

 

---

 

 

 

프로듀서 실행하기

컨슈머는 메시지 수신대기 상태이므로 프로듀서로 메시지를 전송해본다.

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

 

kafka-console-producer --topic <토픽 이름> --broker-list kafka:9092 명령어를 사용하면 보로커의 토픽에 접근하여 메시지를 생성한다.

 

 

 

---

 

 

 

통신 결과

'>' 표시가 되면 메시지를 작성할 수 있다.

 

터미널의 아랫 부분은 생성자(Producer)이기 때문에 '>' 표시가 되고 메시지를 작성할 수 있다.

터미널의 윗부분은 소비자(Consumer)이기 때문에 컨슘이 되고 있다.

 

 

 

---

 

 

 

docker-compose 컨테이너 내리기

$ docker-compose down
 
[+] Running 3/3
 ⠿ Container kafka-docker-kafka-1      Removed                                                                          0.9s
 ⠿ Container kafka-docker-zookeeper-1  Removed                                                                          0.6s
 ⠿ Network kafka-docker_default        Removed                                                                          0.1s

 

 

 

 

 

 

 

 

reference

* kido, [Kafka] Docker Compose 를 이용하여 Single Broker 구성하기