4.5.2. コネクター設定の追加

手順

1. Debezium MongoDB コネクターアーカイブを展開して、コネクタープラグインのディレクトリー構造を作成します。以下に例を示します。

   tree ./my-plugins/
   ./my-plugins/
   ├── debezium-connector-mongodb
   │   ├── ...

2. Debezium コネクターを実行するためのカスタムイメージを作成し、パブリッシュします。

   1. registry.redhat.io/amq7/amq-streams-kafka-26-rhel7:1.6.0 をベースイメージとして使用して、新規の Dockerfile を作成します。以下の例の my-plugins は、プラグインディレクトリーの名前に置き換えます。

      FROM registry.redhat.io/amq7/amq-streams-kafka-26-rhel7:1.6.0
      USER root:root
      COPY ./my-plugins/ /opt/kafka/plugins/
      USER 1001

      Kafka Connect は、コネクターの実行を開始する前に、/opt/kafka/plugins ディレクトリーにあるサードパーティープラグインをロードします。

   2. コンテナーイメージをビルドします。たとえば、前のステップで作成した Dockerfile を debezium-container-for-mongodb として保存し、Dockerfile が現在のディレクトリーにある場合は、以下のコマンドを実行します。

      podman build -t debezium-container-for-mongodb:latest .

   3. カスタムイメージをコンテナーレジストリーにプッシュします。例を以下に示します。

      podman push debezium-container-for-mongodb:latest

   4. 新しいコンテナーイメージを示します。次のいずれかを行います。

      • KafkaConnector カスタムリソースの spec.image プロパティーを編集します。このプロパティーが設定されていると、クラスターオペレータの STRIMZI_DEFAULT_KAFKA_CONNECT_IMAGE 変数がオーバーライドされます。以下に例を示します。

        apiVersion: kafka.strimzi.io/v1beta1
        kind: KafkaConnector
        metadata:
          name: my-connect-cluster
        spec:
          #...
          image: debezium-container-for-mongodb

      • install/cluster-operator/050-Deployment-strimzi-cluster-operator.yaml ファイルの STRIMZI_DEFAULT_KAFKA_CONNECT_IMAGE 変数を編集し、新しいコンテナーイメージを示すようにした後、Cluster Operator を再インストールします。このファイルを編集する場合は、OpenShift クラスターに適用する必要があります。
3. Debezium MongoDB コネクターインスタンスを定義する KafkaConnector カスタムリソースを作成します。コネクター設定の例 を参照してください。

4. コネクターインスタンスを適用します。以下に例を示します。

   oc apply -f inventory-connector.yaml

   これにより inventory-connector が登録され、コネクターが inventory データベースに対して実行されるようになります。

5. コネクターが作成され、指定されたデータベース内の変更のキャプチャーを開始したことを確認します。例えば inventory-connector が起動したときの Kafka Connect のログ出力を見ることで、コネクターのインスタンスを確認することができます。

   1. Kafka Connect のログ出力を表示します。

      oc logs $(oc get pods -o name -l strimzi.io/name=my-connect-cluster-connect)

   2. ログの出力を確認し、初回のスナップショットが実行されたことを確認します。以下のような行が表示されるはずです。

      ... INFO Starting snapshot for ...
      ... INFO Snapshot is using user 'debezium' ...