[Kafka Connect] Docker로 Kafka Connect 사용하기

주문 서비스에서 DB로 저장하는 부분을 Kafka Connect를 이용해서 구현해야 하는데 미리 로컬에서 설정하고 테스트했던 내용을 Docker 컨테이너로 만들 때 옮겨야 한다는 문제가 발생했다. 그래서 여러 가지 시행착오를 겪으면서 Dockerfile, 이미지 빌드, 실행까지 진행해 보았다.

 

---

 

Dockerfile 작성하기

 

FROM confluentinc/cp-kafka-connect:latest
COPY ./lib /usr/share/java/kafka-connect-jdbc

 

위는 내가 작성한 Dockerfile이다. 기본 베이스 이미지로 confluent의 Kafka Connect를 가져오고 현재 폴더에 있는 내용을 /usr/share/java/kafka-connect-jdbc로 복사한다. 참고로 현재 폴더에 있는 내용은 다음과 같다.

 

 

Kafka Connect를 스프링에서 이용하기 위해서는 크게 두 가지 파일이 필요하다. kafka-connect-jdbc.jar 파일과 자신의 DB에 맞는 드라이버 파일이다.

 

내가 진행한 방법은 두 파일 모두 미리 받아두고 docker image로 만드는 것이다.

 

---

 

kafka-connect-jdbc 다운로드

 

kafka-connect-jdbc.jar는 아래 사이트에서 다운로드할 수 있다.

 

https://www.confluent.io/hub/confluentinc/kafka-connect-jdbc

JDBC Connector (Source and Sink)

 

 

 

Self-Hosted에서 다운로드를 한다. 다운로드한 파일은 zip파일인데 압축을 풀고 /lib을 보면 여러 jar파일이 있는데 그중에 kafka-connect-jdbc-[버전].jar이 있는 것을 볼 수 있다. 이 파일을 복사해서 Dockerfile이 있는 경로의 폴더에 넣었다.

 

---

 

드라이버 파일 가져오기

 

DB 드라이버 파일은 gradle 기준으로 

 

C:\Users\[사용자명]\.gradle\caches\modules-2\files-2.1\com.mysql\mysql-connector-j\8.3.0

 

위 경로에서 가져올 수 있다. 나는 mysql을 사용할 것이기 때문에 com.mysql 이후 경로까지 들어갔다.

 

이제 두 파일을 Dockerfile이 있는 경로의 폴더에 이동시킨다. 나는 다른 DB도 나중에 필요할 수도 있을 것 같아서  kafka-connect-jdbc를 다운로드할 때 포함되어 있던 /lib의 파일을 전부 가져왔다. 그래서 다음과 같은 상태가 되면 준비 완료이다.

 

 

 

---

 

다른 방법

 

FROM confluentinc/cp-kafka-connect:latest

RUN confluent-hub install --no-prompt confluentinc/kafka-connect-jdbc:latest

COPY ./lib /usr/share/java/kafka-connect-jdbc

 

이 방법은 직접 해보지는 않아서 되는지는 모르겠지만 kafka-connect-jdbc 이미지를 가져와서 설치하는 방법이다. 이렇게 하면 드라이버 파일만 복사해 주면 된다.

 

https://docs.confluent.io/kafka-connectors/jdbc/current/jdbc-drivers.html

JDBC Drivers | Confluent Documentation

 

필요한 내용이 있다면 공식 문서를 참고하면 도움이 된다.

 

---

 

이미지 빌드하기

 

 

Dockerfile이 있는 경로에서 Shift + 마우스 우클릭으로 PowerShell을 연다.

 

docker build -t [DockerHub계정명]/[이미지명]:[태그명] .

 

위 명령어로 빌드를 진행하고

 

docker push [DockerHub계정명]/[이미지명]:[태그명]

 

다시 위 명령어로 DockerHub에 push를 한다. 주의해야 할 점은 해당 계정명으로 Docker Desktop에 로그인되어 있어야 한다는 점이다.

이렇게까지 했으면 DockerHub에 다음과 같이 이미지가 잘 저장되어 있을 것이다.

 

 

---

 

GCP 인스턴스에서 실행하기

 

나는 docker-compose를 이용하여 Zookeeper와 Kafka, Kafka Connect를 한 번에 실행하도록 설정했다.

 

version: '2'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    container_name: zookeeper
    hostname: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
    networks:
      my-network:
        ipv4_address: 172.18.0.100
  kafka:
    image: confluentinc/cp-kafka:latest
    container_name: kafka
    hostname: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: 172.18.0.101
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://api.delibird.store:9092
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
    depends_on:
      - zookeeper
    networks:
      my-network:
        ipv4_address: 172.18.0.101

  connect:
    image: ness727/my-custom-kafka-connect:latest
    container_name: connect
    hostname: connect
    ports:
      - 8083:8083
    depends_on:
      - kafka
    environment:
      CONNECT_BOOTSTRAP_SERVERS: kafka:9092
      CONNECT_REST_PORT: 8083
      CONNECT_GROUP_ID: "quickstart-avro"
      CONNECT_CONFIG_STORAGE_TOPIC: "quickstart-avro-config"
      CONNECT_OFFSET_STORAGE_TOPIC: "quickstart-avro-offsets"
      CONNECT_STATUS_STORAGE_TOPIC: "quickstart-avro-status"
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_REST_ADVERTISED_HOST_NAME: "api.delibird.store"
      CONNECT_LOG4J_ROOT_LOGLEVEL: WARN
      CONNECT_PLUGIN_PATH: "/usr/share/java"
    networks:
      my-network:
        ipv4_address: 172.18.0.102

networks:
  my-network:
    external: true
    name: delibird-network

 

여기서 아까 파일을 복사하고 했던 경로가 CONNECT_PLUGIN_PATH 이 부분이다. 복사했던 파일들이 해당 경로 내에 있어야 플러그인들이 인식된다.

여기서 또 주의할 점은 Kafka Connect의 이미지를 직접 만들었는데 실수로 confluent의 Kafka Connect 이미지를 가져오면 안 된다. 내가 이거 때문에 왜 안 되지 하면서 시간을 많이 낭비했다...

직접 만든 이미지명을 지정하도록 한다.

 

docker-compose -f docker-compose-single-broker.yaml up -d

 

compose yaml 파일을 저장했으면 위 명령어로 실행하면 된다. docker-compose-single-broker는 각자 이름이 다를 수 있으므로 자신이 작성한 파일명을 지정해서 실행하면 된다.

 

docker ps -a

 

위 명령어로 시간이 지나도 Up 상태가 제대로 유지되는지 확인하자. 만약 실행 도중에 문제가 생겼으면 Exited 상태가 될 텐데

 

docker logs [컨테이너명 또는 컨테이너ID]

 

그때는 위 명령어로 로그를 확인하고 원인을 파악하면 된다.

 

 

파일 복사 잘 됐는지 확인하기

 

docker exec -it [컨테이너명 또는 컨테이너ID] /bin/bash
cd /usr/share/java/
ls

 

 

위 명령어를 실행해서 컨테이너 내의 usr/share/java에 Dockerfile에서 복사한 파일이 제대로 있는지 확인할 수 있다.

나는 Dockerfile에서 실수로 .로 모든 파일을 지정해서 Dockerfile까지 복사됐는데 딱히 상관은 없다.

 

 

Docker Network 설정하기

 

docker network create --gateway 172.18.0.1 --subnet 172.18.0.0/16 delibird-network

 

컨테이너끼리 네트워크를 구성한다.

 

---

 

테스트하기

 

 

http://api.delibird.store:8083/connector-plugins

 

위 경로로 GET 요청을 보내면 사용되고 있는 플러그인 목록을 볼 수 있다. 주소는 자신의 Kafka Connect가 실행되고 있는 곳을 지정하면 된다. JdbcSinkConnector와 JdbcSourceConnector가 있으면 아까 복사했던 파일이 제대로 작동하고 있다는 것이다.

 

{
    "name": "delibird-sink-connect",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "connection.url": "jdbc:mysql://[DB주소]:3306/delibird",
        "connection.user": "megamaker",
        "connection.password": "DB 사용자 비번",
        "auto.create": "true",
        "auto.evolve": "true",
        "delete.enabled": "false",
        "tasks.max": "1",
        "topics": "orders"
    }
}

 

 

http://api.delibird.store:8083/connectors

 

/connectors로 POST 요청을 보낸다. Body에 JSON 타입으로 위 정보를 보내면 sink가 정상적으로 등록된 것을 알 수 있다.

 

 

같은 요청을 GET으로 보내면 등록된 내용을 확인할 수 있다.

 

 

이런식으로 현재 상태도 확인할 수 있다.