Kafka Docker Compose로 설치 및 실습해보기

이번에 프로젝트에서 MSA를 적용하게 되면서 비동기 통신 작업을 위해 Kafka를 학습하였고 환경설정을 하게 되었다.

1. kafka, zookeeper Docker Compose로 설치

version: '3.8'
services:
  zookeeper:
    image: wurstmeister/zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
  kafka:
    image: wurstmeister/kafka
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: 127.0.0.1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_CREATE_TOPICS: "test:1:1"
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
    depends_on: 
      - zookeeper

2. Kafka 테스트

(1) 토픽생성하기

kafka-topics.sh --bootstrap-server localhost:9092 --create --topic test-events --partitions 1

(2) Producer 실행

./opt/kafka_2.13-2.8.1/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test-events

(3) Consumer 실행

./opt/kafka_2.13-2.8.1/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-events \ --from-beginning

(4) 테스트 결과

Producer가 토픽에 메시지를 보내면 해당 토픽을 구독하고 있던 Consumer가 데이터를 읽어오는 것을 볼 수 있다.

3. Kafka Connect 설치 

(1) kafka connect 설치

curl -O http://packages.confluent.io/archive/6.1/confluent-community-6.1.0.tar.gz

(1) kafka connect 실행 (zookeeper와 kafka가 실행중일 때 Mac기준)

./bin/connect-distributed ./etc/kafka/connect-distributed.properties

실행후에 토픽을 조회해 보면 기존에 존재하지 않던 것들이 자동으로 생긴다.

4. Kafka Connect JDBC 설치 및 Kafka Connect 설정

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

JDBC Connector (Source and Sink)

위 URL로 접근하면 JDBC Connector를 설치할 수 있다.

그다음에 Kafka Connect 설치한 폴더에 들어가서 connect-distributed.properties 파일을 열어서 방금 설치한 Kafka Connect JDBC의 confluentic-kafka-connect-jdbc-10.7.2/lib 폴더 내의 kafka-connect-jdbc-10.7.2.jar를 사용하도록 연결해야 한다.

plugin.path를 원하는 jar파일이 있는 path로 변경해준다.

5. Kafka Conenct 폴더 내로 mariadb-java-client.jar 파일 복사

 ~/.m2/repository/org/mariadb/jdbc/mariadb-java-client/2.7.2  경로로 이동

mariadb-java-client-2.7.2.jar 파일을  ~/xxx/confluent-6.1.0/share/java/kafka 폴더 내로 복사

cp ./mariadb-java-client-2.7.2.jar ~/xxx/confluent-6.1.0/share/java/kafka

6. Kafka Source Connect 등록

Postman을 실행하여 http://127.0.0.1:8083/connectors로 아래 json을 body에 포함하여 POST 요청한다. 

{
    "name" : "my-source-connect",
    "config" : {
        "connector.class" : "io.confluent.connect.jdbc.JdbcSourceConnector",
        "connection.url":"jdbc:mariadb://localhost:3305/mydb",
        "connection.user":"root",
        "connection.password":"비밀번호",
        "mode": "incrementing",
        "incrementing.column.name" : "id", //자동 증가시킬 컬럼
        "table.whitelist":"users", //감지할 테이블 이름
        "topic.prefix" : "my_topic_",
        "tasks.max" : "1"
    }
}

만약 요청을 잘못 보내서 삭제해야 하는 경우 http://127.0.0.1:8083/connectors/소스커넥터이름 DELETE 요청을 한다.

그러고 나서 감지할 테이블 즉, 여기서는 users 테이블에 row를 추가해 본다.
그러면 감지되어 아래와 같이 토픽이 생성되고 컨슈머를 실행하면 값을 볼 수 있다.

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "int32",
        "optional": false,
        "field": "id"
      },
      {
        "type": "string",
        "optional": true,
        "field": "user_id"
      },
      {
        "type": "string",
        "optional": true,
        "field": "pwd"
      },
      {
        "type": "string",
        "optional": true,
        "field": "name"
      },
      {
        "type": "int64",
        "optional": true,
        "name": "org.apache.kafka.connect.data.Timestamp",
        "version": 1,
        "field": "created_at"
      }
    ],
    "optional": false,
    "name": "users"
  },
  "payload": {
    "id": 1,
    "user_id": "user1",
    "pwd": "test111",
    "name": "username",
    "created_at": 1685872665000
  }
}

7. Kafka Sink Connect 등록

Kafka Source Connector와 똑같이 Postman을 통해 sink connector를 등록한다.

{
    "name":"my-sink-connect",
    "config":{
        "connector.class":"io.confluent.connect.jdbc.JdbcSinkConnector",
        "connection.url":"jdbc:mysql://localhost:3305/mydb",
        "connection.user":"root",
        "connection.password":"love1994",
        "auto.create":"true",
        "auto.evolve":"true",
        "delete.enabled":"false",
        "tasks.max":"1",
        "topics":"my_topic_users"
    }
}

8. Kafka Source Connector, Sink Connector 등록 확인

http://127.0.0.1:8083/connectors Get 호출하여 커넥터가 제대로 등록되었는지 확인한다.

sink connector 등록 시에 auto.create=ture 설정을 줬으므로 위와 같이 my_topic_users 테이블이 등록되었다.
또한 이전에 source connector 등록 후에 users테이블에 insert 되었던 2개의 user row가 들어가 있는 것도 확인할 수 있다. 

 

 

아직 이 구조가 익숙하지 않아 kafka source connector, sink connector는 컨테이너 환경이 아닌 로컬에서 동작하도록 하였는데 추후에 익숙해지면 도전해봐야겠다..