2.3. Red Hat Enterprise Linux での AMQ Streams を使用した Debezium のデプロイ

手順

1. 使用する Debezium コネクターを Red Hat Integration ダウンロードサイト からダウンロードします。たとえば、Debezium を MySQL データベースで使用するには、Debezium 1.9.5 MySQL Connector をダウンロードします。

2. AMQ Streams をデプロイした Red Hat Enterprise Linux ホストで、ターミナルウィンドウを開き、/opt/kafka に connector-plugins ディレクトリーを作成します (まだ存在しない場合)。

   $ sudo mkdir /opt/kafka/connector-plugins

3. 次のコマンドを入力して、/opt/kafka/connector-plugins ディレクトリーにダウンロードした Debezium コネクターアーカイブの内容を抽出します。

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

4. インストールするコネクターごとに、手順 1 - 3 を繰り返します。

5. ターミナルウィンドウから、kafka ユーザーとしてサインインします。

   $ su - kafka
   $ Password:

6. Kafka Connect プロセスが実行中の場合は、停止します。

   1. 次のコマンドを入力して、Kafka Connect が分散モードで実行されているかどうかを確認します。

      $ jcmd | grep ConnectDistributed

      プロセスが実行中の場合、コマンドはプロセス ID を返します。次に例を示します。

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

   2. プロセス ID を指定して kill コマンドを入力し、プロセスを停止します。次に例を示します。

      $ kill 18514

7. /opt/kafka/config/ にある connect-distributed.properties ファイルを編集し、plugin.path の値を Debezium コネクタープラグインの親ディレクトリーの場所に設定します。

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

8. 分散モードで Kafka Connect を起動します。

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

9. Kafka Connect の実行後、Kafka Connect API を使用してコネクターを登録します。
   curl コマンドを入力して、「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", \
   "database.server.name": "dbserver1", \
   "database.include.list": "public.inventory", \
   "database.history.kafka.bootstrap.servers": "kafka:9092", \
   "database.history.kafka.topic": "dbhistory.inventory" } }'

   複数のコネクターを登録するには、コネクターごとに個別のリクエストを送信します。

10. Kafka Connect を再起動して、変更を実装します。

    Kafka Connect が起動すると、connector-plugins ディレクトリーから、設定済みの Debezium コネクターをロードします。

    設定が完了すると、デプロイされたコネクターはソースデータベースに接続し、挿入、更新、または削除された行またはドキュメントごとにイベントを生成します。

11. 各 Kafka Connect ワーカーノードに対して、ステップ 5 - 10 を繰り返します。