3.6. Debezium Db2 コネクターのデプロイ

手順

1. Kafka Connect の Debezium Db2 コンテナーを作成します。

   1. Debezium Db2 コネクターアーカイブ をダウンロードします。

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

      ./my-plugins/
      ├── debezium-connector-db2
      │   ├── ...

      Copy to Clipboard Toggle word wrap

   3. registry.redhat.io/amq7/amq-streams-kafka-28-rhel8:1.8.0 をベースイメージとして使用する Docker ファイルを作成します。たとえば、ターミナルウィンドウに以下のコマンドを入力します。my -plugins はプラグインディレクトリーの名前に置き換えます。

      cat <<EOF >debezium-container-for-db2.yaml 

      1

      
      FROM registry.redhat.io/amq7/amq-streams-kafka-28-rhel8:1.8.0
      USER root:root
      COPY ./<my-plugins>/ /opt/kafka/plugins/ 

      2

      
      USER 1001
      EOF

      Copy to Clipboard Toggle word wrap

      1 1 1 1 1 1 1

      任意のファイル名を指定できます。

      2 2 2 2 2 2 2

      my-plugins は、プラグインディレクトリーの名前に置き換えます。

      このコマンドは、現在のディレクトリーに debezium-container-for-db2.yaml という名前の Docker ファイルを作成します。

   4. 前の手順で作成した debezium-container-for-db2.yaml Docker ファイルからコンテナーイメージをビルドします。ファイルを含むディレクトリーから、ターミナルウィンドウを開き、以下のコマンドのいずれかを入力します。

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

      Copy to Clipboard Toggle word wrap

      docker build -t debezium-container-for-db2:latest .

      Copy to Clipboard Toggle word wrap

      上記のコマンドは、debezium-container-for-db2 という名前のコンテナーイメージを構築します。

   5. カスタムイメージを quay.io などのコンテナーレジストリーまたは内部のコンテナーレジストリーにプッシュします。コンテナーレジストリーは、イメージをデプロイする OpenShift インスタンスで利用できる必要があります。以下のいずれかのコマンドを実行します。

      podman push <myregistry.io>/debezium-container-for-db2:latest

      Copy to Clipboard Toggle word wrap

      docker push <myregistry.io>/debezium-container-for-db2:latest

      Copy to Clipboard Toggle word wrap

   6. 新しい Debezium Db2 KafkaConnect カスタムリソース(CR)を作成します。たとえば、以下の例のように アノテーション およびイメージ プロパティーを指定する dbz-connect.yaml という名前の KafkaConnect CR を作成します。

      apiVersion: kafka.strimzi.io/v1beta2
      kind: KafkaConnect
      metadata:
        name: my-connect-cluster
        annotations:
          strimzi.io/use-connector-resources: "true" 

      1

      
      spec:
        #...
        image: debezium-container-for-db2  

      2

      Copy to Clipboard Toggle word wrap

      1

      metadata.annotations は、KafkaConnectors リソースがこの Kafka Connect クラスターでコネクター を設定するために使用される ことを Cluster Operator に示します。

      2

      spec.image は、Debezium コネクターを実行するために作成したイメージの名前を指定します。このプロパティーは、Cluster Operator の STRIMZI_DEFAULT_KAFKA_CONNECT_IMAGE 変数を上書きします。

   7. 以下のコマンドを入力して、KafkaConnect CR を OpenShift Kafka Connect 環境に適用します。

      oc create -f dbz-connect.yaml

      Copy to Clipboard Toggle word wrap

      このコマンドは、Debezium コネクターを実行するために作成したイメージの名前を指定する Kafka Connect インスタンスを追加します。

2. Debezium Db2 コネクターインスタンスを設定する KafkaConnector カスタムリソースを作成します。

   コネクターの設定プロパティーを指定する a .yaml ファイルで Debezium Db2 コネクターを設定します。コネクター設定は、Debezium に対して、スキーマおよびテーブルのサブセットにイベントを生成するよう指示する可能性があり、または機密性の高い、大きすぎる、または不必要な指定のコラムで Debezium が値を無視、マスク、または切り捨てするようにプロパティーを設定する可能性もあります。

   以下の例では、ポート 50000 で Db2 サーバーホスト 192.168.99.100 に接続する Debezium コネクターを設定します。このホストには、mydatabase という名前のデータベースがあり、名前 インベントリーが 含まれるテーブルはサーバーの論理名 です。

   Db2 inventory-connector.yaml

   apiVersion: kafka.strimzi.io/v1beta2
     kind: KafkaConnector
     metadata:
       name: inventory-connector  

   1

   
       labels:
         strimzi.io/cluster: my-connect-cluster
       annotations:
         strimzi.io/use-connector-resources: 'true'
     spec:
       class: io.debezium.connector.db2.Db2Connector 

   2

   
       tasksMax: 1  

   3

   
       config:  

   4

   
         database.hostname: 192.168.99.100   

   5

   
         database.port: 50000 

   6

   
         database.user: db2inst1 

   7

   
         database.password: Password! 

   8

   
         database.dbname: mydatabase 

   9

   
         database.server.name: fullfillment   

   10

   
         database.include.list: public.inventory   

   11

   Copy to Clipboard Toggle word wrap

   Expand

   表3.11 コネクター設定の説明

   項目	説明
   1	Kafka Connect クラスターに登録する場合のコネクターの名前。
   2	この Db2 コネクタークラスの名前。
   3	1 度に 1 つのタスクのみが動作する必要があります。
   4	コネクターの設定。
   5	Db2 インスタンスのアドレスであるデータベースホスト。
   6	Db2 インスタンスのポート番号。
   7	Db2 ユーザーの名前。
   8	Db2 ユーザーのパスワード。
   9	変更をキャプチャーするデータベースの名前。
   10	namespace を形成する Db2 インスタンス/クラスターの論理名で、コネクターが書き込む Kafka トピックの名前、Kafka Connect スキーマ名、および Arvo コネクターが使用される場合に対応する Avro スキーマの namespace で使用 さ れます。
   11	Debezium が変更をキャプチャーする必要があるすべてのテーブルのリスト。

   Show more

3. Kafka Connect でコネクターインスタンスを作成します。たとえば、KafkaConnector リソースを inventory-connector.yaml ファイルに保存した場合、以下のコマンドを実行します。

   oc apply -f inventory-connector.yaml

   Copy to Clipboard Toggle word wrap

   上記のコマンドは inventory-connector を登録し、コネクターは KafkaConnector CR に定義されている mydatabase データベースに対して実行を開始します。

4. コネクターが作成され、起動されたことを確認します。

   1. Kafka Connect ログ出力を表示して、コネクターが作成され、指定データベースの変更のキャプチャーが開始されたことを確認します。

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

      Copy to Clipboard Toggle word wrap

   2. ログの出力を確認し、Debezium が初回のスナップショットを実行することを確認します。ログには、以下のメッセージと同様の出力が表示されます。

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

      Copy to Clipboard Toggle word wrap

      コネクターがエラーがなく正常に起動すると、コネクターが変更をキャプチャーする各テーブルのトピックが作成されます。CR のサンプルでは、include.list プロパティーに指定されたテーブルのトピックがあります。ダウンストリームアプリケーションは、これらのトピックをサブスクライブできます。

   3. 以下のコマンドを実行して、コネクターによってトピックが作成されたことを検証します。

      oc get kafkatopics

      Copy to Clipboard Toggle word wrap