第3章 シンクコネクター

手順

1. OpenShift クラスターにログインします。

2. コネクターの Debezium KafkaConnect カスタムリソース (CR) を作成するか、既存のリソースを変更します。
   たとえば、次の抜粋に示すように、annotations と image プロパティーを指定する dbz-jdbc-connect.yaml という名前の KafkaConnect CR を作成します。次の例では、カスタムリソースは、次のアーティファクトをダウンロードするように設定されています。

   • Debezium JDBC コネクターアーカイブ。

   • Oracle または Db2 シンクデータベースへの接続に必要な JDBC ドライバー。他のシンク宛先ではこのエントリーを省略できます。

     Debezium コネクターを含む KafkaConnect カスタムリソースを定義する dbz-jdbc-connect.yaml ファイル

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

     1

     
     spec:
       version: 3.9.0
       replicas: 1
       bootstrapServers: debezium-kafka-cluster-kafka-bootstrap:9093
       build:  

     2

     
         output:  

     3

     
           type: imagestream  

     4

     
           image: debezium-streams-connect:latest
         plugins:   

     5

     
           - name: debezium-jdbc-connector
             artifacts:
               - type: zip  

     6

     
                 url: https://maven.repository.redhat.com/ga/io/debezium/debezium-connector-jdbc/3.0.8.Final-redhat-00004/debezium-connector-jdbc-3.0.8.Final-redhat-00004-plugin.zip  

     7

     
               - type: jar
                 url: https://repo1.maven.org/maven2/com/oracle/ojdbc/ojdbc11/21.15.0.0/ojdbc11-21.15.0.0.jar  

     8

     
       ...

     Copy to Clipboard Toggle word wrap

     Expand

     表3.5 Kafka Connect 設定の説明

     項目	説明
     1	strimzi.io/use-connector-resources アノテーションを "true" に設定して、クラスター Operator が KafkaConnector リソースを使用してこの Kafka Connect クラスター内のコネクターを設定できるようにします。
     2	spec.build 設定は、ビルドイメージの保存場所を指定し、プラグインアーティファクトの場所とともにイメージに追加するプラグインをリストします。
     3	build.output は、新しくビルドされたイメージを保存するレジストリーを指定します。
     4	イメージ出力の名前およびイメージ名を指定します。output.type の有効な値は、Quay などのコンテナーレジストリーにプッシュする docker または、イメージを内部 OpenShift ImageStream にプッシュする imagestream です。ImageStream を使用するには、ImageStream リソースをクラスターにデプロイする必要があります。KafkaConnect 設定で build.output を指定する方法の詳細は、{NameConfiguringStreamsOpenShift} の Streams for Apache Kafka ビルドスキーマリファレンス を参照してください。
     5	plugins 設定は、Kafka Connect イメージに追加するすべてのコネクターをリストします。リストの各エントリーについて、プラグイン name と、コネクターのビルドに必要なアーティファクトに関する情報を指定します。必要に応じて、各コネクタープラグインに対して、コネクターと使用できる他のコンポーネントを含めることができます。たとえば、Service Registry アーティファクトまたは Debezium スクリプトコンポーネントを追加できます。
     6	artifacts.type の値は、artifacts.url で指定するアーティファクトのファイルタイプを指定します。有効なタイプは zip、tgz、または jar です。Debezium コネクターアーカイブは、.zip ファイル形式で提供されます。JDBC ドライバーファイルは .jar 形式です。type の値は、url フィールドで参照されるファイルのタイプと一致させる必要があります。
     7	artifacts.url の値は、コネクターアーティファクトのファイルを格納する Maven リポジトリーなどの HTTP サーバーのアドレスを指定します。OpenShift クラスターが指定されたサーバーにアクセスできる必要があります。
     8	(Db2 または Oracle シンクのみの場合) Maven Central 内の JDBC JDBC ドライバーの場所を指定します。Debezium がこれらのデータベースに接続するために必要なドライバーは、Debezium コネクターアーカイブに含まれていません。 この例では、Oracle Database JDBC ドライバーの Maven URL を示しています。Db2 JDBC ドライバーは、次の Maven の場所で入手できます: https://repo1.maven.org/maven2/com/ibm/db2/jcc/11.5.9.0/jcc-11.5.9.0.jar

     Show more

3. 以下のコマンドを入力して、KafkaConnect ビルド仕様を OpenShift クラスターに適用します。

   oc create -f dbz-jdbc-connect.yaml

   Copy to Clipboard Toggle word wrap

   Streams Operator はカスタムリソースで指定された設定に基づいて、デプロイする Kafka Connect イメージを準備します。
   ビルドが完了すると、Operator はイメージを指定されたレジストリーまたは ImageStream にプッシュし、Kafka Connect クラスターを起動します。設定にリスト表示されているコネクターアーティファクトはクラスターで利用できます。

4. KafkaConnector リソースを作成し、デプロイする各コネクターのインスタンスを定義します。
   たとえば、次の KafkaConnector CR を作成し、orders-to-postgresql-jdbc-connector.yaml として保存します。

   Debezium コネクターの KafkaConnector カスタムリソースを定義する orders-to-postgresql-jdbc-connector.yaml ファイル

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: orders-topic-to-postgresql-via-jdbc-sink-connector 

   1

   
     labels: strimzi.io/cluster: debezium-kafka-connect-cluster
   spec:
     class: io.debezium.connector.jdbc.JdbcSinkConnector  

   2

   
     tasksMax: 1  

   3

   
      config:  

   4

   
       connection.url: jdbc://postgresql://<host>:<port>/<database-name> 

   5

   
       connection.username: <database-user>  

   6

   
       connection.password: <database-pwd>  

   7

   
       insert.mode: upsert
       delete.enabled: true
       topics: orders 

   8

   
       primary.key.mode: record_key
       schema.evolution: basic
       use.time.zone: UTC

   Copy to Clipboard Toggle word wrap

   Expand

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

   項目	説明
   1	Kafka Connect クラスターに登録するコネクターの名前。
   2	コネクタークラスの名前。
   3	同時に動作できるタスクの数。
   4	コネクターの設定。
   5	シンクデータベースの JDBC 接続 URL。URL は、ポート番号と、データベースへの接続に必要な認証プロパティーを指定します。たとえば、jdbc:oracle:thin:@myhost.example.com:1521/myservice などです。
   6	Debezium がデータベースへの接続に使用するアカウントの名前。
   7	Debezium がデータベースユーザーアカウントに接続するために使用するパスワード。
   8	コネクターが読み取る Kafka トピックのコンマ区切りリストを指定します。各トピックからのイベントは、シンクデータベースで同じ名前のテーブルにストリーミングされます。

   Show more

5. 以下のコマンドを実行してコネクターリソースを作成します。

   oc create -n <namespace> -f <kafkaConnector>.yaml

   Copy to Clipboard Toggle word wrap

   以下に例を示します。

   oc create -n debezium -f jdbc-inventory-connector.yaml

   Copy to Clipboard Toggle word wrap

   コネクターは Kafka Connect クラスターに登録され、KafkaConnector CR の spec.config.database.dbname で指定されたデータベースに対して実行を開始します。コネクター Pod の準備ができると、Debezium が実行されます。

手順

1. Kafka Connect に Debezium JDBC コネクターコンテナーを作成します。

   1. registry.redhat.io/amq-streams/kafka-39-rhel9:2.9.0 をベースイメージとして使用する Containerfile を作成します。たとえば、ターミナルウィンドウから、以下のコマンドを入力します。
      

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

   podman build -t debezium-jdbc-connector-container:latest .

   Copy to Clipboard Toggle word wrap

   docker build -t debezium-jdbc-connector-container:latest .

   Copy to Clipboard Toggle word wrap

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

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

   podman push <myregistry.io>/debezium-jdbc-connector-container:latest

   Copy to Clipboard Toggle word wrap

   docker push <myregistry.io>/debezium-jdbc-connector-container:latest

   Copy to Clipboard Toggle word wrap

3. 新しい Debezium Oracle KafkaConnect カスタムリソース (CR) を作成します。たとえば、annotations および image プロパティーを指定する dbz-connect.yaml という名前の KafkaConnect CR を作成します。以下の例は、KafkaConnect カスタムリソースを記述する dbz-connect.yaml ファイルからの抜粋を示しています。
   

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

   1

   
   spec:
     image: debezium-jdbc-connector-container 

   2

   
   
     ...

   Copy to Clipboard Toggle word wrap

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

   oc create -f dbz-connect.yaml

   Copy to Clipboard Toggle word wrap

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

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

      Debezium JDBC コネクターは、コネクターの設定プロパティーを指定する .yaml ファイルで設定します。

      次の例は、KafkaConnect カスタムリソースの主要なプロパティーのいくつかを設定する dbz-connect.yaml ファイルからの抜粋を示しています。
      コネクターは、ポート 5432 上の PostgreSQL サーバーシンクへの JDBC 接続を確立します。
      
      使用可能なコネクタープロパティーの全範囲については、Debezium JDBC コネクター設定プロパティーの説明 を参照してください。

      例3.1 jdbc-connector.yaml

      apiVersion: kafka.strimzi.io/v1beta2
      kind: KafkaConnector
      metadata:
        labels:
        strimzi.io/cluster: my-connect-cluster 

      1

      
        name: orders-topic-to-postgresql-via-jdbc-sink-connector 

      2

      
      ...
      spec:
        class: io.debezium.connector.jdbc.JdbcSinkConnector 

      3

      
      ...
        config:
          connection.url: jdbc:<_jdbc_>://<_database-host_>:5432/<_database-name_>  

      4

      
          connection.username: "<_database-user_>" 

      5

      
          connection.password: "<_database-password_>" 

      6

      
          topics: orders 

      7

      Copy to Clipboard Toggle word wrap

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

   oc apply -f jdbc-connector.yaml

   Copy to Clipboard Toggle word wrap

   上記のコマンドは、orders-topic-to-postgresql-via-jdbc-sink-connector を登録します。コネクターは起動し、KafkaConnector CR で指定されたとおりに、orders トピックからの読み取りを開始します。