3.2. Kafka Connect のデプロイ

手順

1. OpenShift クラスターにログインし、debezium などのプロジェクトを作成または開きます。

2. コネクターの Debezium KafkaConnect カスタムリソース (CR) を作成するか、既存のリソースを変更します。
   以下の例は、KafkaConnect カスタムリソースを記述した dbz-connect.yaml ファイルからの抜粋を示しています。
   metadata.annotations および spec.build プロパティーが必要です。

   例3.1 Debezium コネクターを含む KafkaConnect カスタムリソースを定義した dbz-connect.yaml ファイル

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

   1

   
   spec:
     replicas: 1
     bootstrapServers: my-cluster-kafka-bootstrap:9092
     version: 3.9.0
     build: 

   2

   
       output: 

   3

   
       type: imagestream  

   4

   
       image: debezium-streams-connect:latest
       plugins: 

   5

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

   6

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

   7

   
   
   ...

   Copy to Clipboard Toggle word wrap

   Expand

   表3.1 Kafka Connect 設定の説明

   項目	説明
   1	strimzi.io/use-connector-resources アノテーションを "true" に設定して、クラスター Operator が KafkaConnector リソースを使用してこの Kafka Connect クラスター内のコネクターを設定できるようにします。
   2	spec.build 設定は、ビルドイメージの保存場所を指定し、プラグインアーティファクトの場所とともにイメージに追加するプラグインをリストします。
   3	build.output は、新しくビルドされたイメージを保存するレジストリーを指定します。
   4	イメージ出力の名前およびイメージ名を指定します。output.type の有効な値は、Docker Hub や Quay などのコンテナーレジストリーにプッシュする場合は docker、内部の OpenShift ImageStream にイメージをプッシュする場合は imagestream です。ImageStream を使用するには、ImageStream リソースをクラスターにデプロイする必要があります。KafkaConnect 設定で build.output を指定する方法の詳細は、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 クラスターが指定されたサーバーにアクセスできる必要があります。

   Show more

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

   oc create -f dbz-connect.yaml

   Copy to Clipboard Toggle word wrap

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

4. KafkaConnector リソースを作成して、MySQL コネクターのインスタンスを定義します。
   たとえば、以下の KafkaConnector CR を作成し、debezium-inventory-connector.yaml として保存します。

   例3.2 Debezium コネクターの KafkaConnector カスタムリソースを定義した mysql-inventory-connector.yaml ファイル

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

   1

   
     labels:
           strimzi.io/cluster: my-connect-cluster
   spec:
     class: io.debezium.connector.mysql.MySqlConnector 

   2

   
     tasksMax: 1  

   3

   
     config:  

   4

   
       database.hostname: mysql 

   5

   
       database.port: 3306   

   6

   
       database.user: debezium  

   7

   
       database.password: dbz  

   8

   
       database.server.id: 184054
       topic.prefix: dbserver1  

   9

   
       table.include.list: inventory.*  

   10

   
       schema.history.internal.kafka.bootstrap.servers: 'my-cluster-kafka-bootstrap:9092' 

   11

   
       schema.history.internal.kafka.topic: schema-changes.inventory 

   12

   Copy to Clipboard Toggle word wrap
   Expand

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

   項目	説明
   1	Kafka Connect クラスターに登録するコネクターの名前。
   2	コネクタークラスの名前。
   3	一度に 1 つのタスクのみを実行します。MySQL コネクターは MySQL サーバーの binlog を読み取るため、単一のコネクタータスクを使用して、適切な順序とイベント処理を確保します。Kafka Connect サービスは、コネクターを使用して 1 つ以上のタスクを開始し、作業を完了します。実行中のタスクを Kafka Connect サービスのクラスター全体に自動的に分散します。サービスが停止またはクラッシュした場合、タスクは実行中のサービスに再分散されます。
   4	コネクターの設定。
   5	MySQL データベースインスタンスのホスト名またはアドレス。
   6	データベースインスタンスのポート番号。
   7	Debezium がデータベースに接続するユーザーアカウントの名前。
   8	Debezium がデータベースユーザーアカウントに接続するために使用するパスワード。
   9	MySQL サーバーまたはクラスターのトピック接頭辞。この文字列は、コネクターがイベントレコードを送信する全 Kafka トピックの名前の前に付けます。
   10	コネクターが変更イベントをキャプチャーするテーブルのリスト。コネクターは、inventory テーブルで発生した場合にのみ、変更を検出します。
   11	DDL ステートメントをデータベーススキーマの履歴トピックに書き込み、復元するためにコネクターによって使用される Kafka ブローカーのリスト。これは、コネクターが変更イベントレコードを送信するブローカーと同じです。再起動後、コネクターは、コネクターが読み取りを再開する時点で binlog に存在しているデータベーススキーマを復元します。
   12	データベーススキーマ履歴トピックの名前。このトピックは内部使用のみを目的としており、コンシューマーが使用しないようにしてください。

   Show more

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

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

   Copy to Clipboard Toggle word wrap

   以下に例を示します。

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

   Copy to Clipboard Toggle word wrap

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