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

手順

1. 使用する Debezium コネクターを Red Hat Integration ダウンロードサイト からダウンロードします。たとえば、Debezium を MySQL データベースで使用するには、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 ファイルを編集し、plugin.path の値を Debezium コネクタープラグインの親ディレクトリーの場所に設定します。

   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 コマンドを入力して、「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 を繰り返します。