2.3. Red Hat Enterprise Linux에서 AMQ Streams를 사용하여 Debezium 배포

절차

1. Red Hat Integration 다운로드 사이트에서 사용하려는 Debezium 커넥터 또는 커넥터를 다운로드합니다. 예를 들어 MySQL 데이터베이스와 함께 Debezium을 사용하려면 Debezium 2.1.4 MySQL Connector 를 다운로드합니다.

2. AMQ Streams를 배포한 Red Hat Enterprise Linux 호스트에서 터미널 창을 열고 /opt/kafka 에 connector-plugins 디렉터리를 생성합니다.

   $ sudo mkdir /opt/kafka/connector-plugins

   Copy to Clipboard Toggle word wrap

3. 다음 명령을 입력하여 /opt/kafka/connector-plugins 디렉터리에 다운로드한 Debezium 커넥터 아카이브의 콘텐츠를 추출합니다.

   $ sudo unzip debezium-connector-mysql-2.1.4.Final.zip -d /opt/kafka/connector-plugins

   Copy to Clipboard Toggle word wrap
4. 설치하려는 각 커넥터에 대해 1~3단계를 반복합니다.

5. 터미널 창에서 kafka 사용자로 로그인합니다.

   $ su - kafka
   $ Password:

   Copy to Clipboard Toggle word wrap

6. Kafka Connect 프로세스가 실행 중인 경우 중지합니다.

   1. 다음 명령을 입력하여 Kafka Connect가 분산 모드에서 실행 중인지 확인합니다.

      $ jcmd | grep ConnectDistributed

      Copy to Clipboard Toggle word wrap

      프로세스가 실행 중인 경우 명령에서 프로세스 ID를 반환합니다. 예를 들면 다음과 같습니다.

      18514 org.apache.kafka.connect.cli.ConnectDistributed /opt/kafka/config/connect-distributed.properties

      Copy to Clipboard Toggle word wrap

   2. 프로세스 ID로 kill 명령을 입력하여 프로세스를 중지합니다. 예를 들면 다음과 같습니다.

      $ kill 18514

      Copy to Clipboard Toggle word wrap

7. /opt/kafka/config/ 에서 connect-distributed.properties 파일을 편집하고 Debezium 커넥터 플러그인의 상위 디렉터리 위치로 plugin.path 값을 설정합니다.

   plugin.path=/opt/kafka/connector-plugins

   Copy to Clipboard Toggle word wrap

8. 분산 모드에서 Kafka Connect를 시작합니다.

   $ /opt/kafka/bin/connect-distributed.sh /opt/kafka/config/connect-distributed.properties

   Copy to Clipboard Toggle word wrap

9. Kafka Connect가 실행된 후 Kafka Connect API를 사용하여 커넥터를 등록합니다.
   curl 명령을 입력하여 2.2절. “Debezium 커넥터 구성 계획” 에 지정한 커넥터 구성 JSON을 localhost:8083/connectors 의 Kafka Connect REST API 엔드포인트에 전송하는 POST 요청을 제출합니다.
   예를 들면 다음과 같습니다.

   curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ \
   -d '{"name": "inventory-connector", "config": \
   { "connector.class": "io.debezium.connector.mysql.MySqlConnector", \
   "tasks.max": "1", \
   "database.hostname": "mysql", \
   "database.port": "3306", \
   "database.user": "debezium", \
   "database.password": "dbz", \
   "database.server.id": "184054", \
   "topic.prefix": "dbserver1", \
   "table.include.list": "public.inventory", \
   "schema.history.internal.kafka.bootstrap.servers": "kafka:9092", \
   "schema.history.internal.kafka.topic": "dbhistory.inventory" } }'

   Copy to Clipboard Toggle word wrap

   여러 커넥터를 등록하려면 각각에 대해 별도의 요청을 제출합니다.

10. Kafka Connect를 다시 시작하여 변경 사항을 구현합니다.

    Kafka Connect가 시작되면 connector-plugins 디렉터리에서 구성된 Debezium 커넥터를 로드합니다.

    구성을 완료한 후 배포된 커넥터는 소스 데이터베이스에 연결되고 삽입, 업데이트 또는 삭제된 행 또는 문서에 대한 이벤트를 생성합니다.

11. 각 Kafka Connect 작업자 노드에 대해 5-10단계를 반복합니다.