6.5. Kafka Connect コネクターの追加

手順

1. .spec.build.output でコンテナーレジストリーを、.spec.build.plugins で追加のコネクターを指定して、KafkaConnect カスタムリソースを設定します。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect
   metadata:
     name: my-connect-cluster
   spec: 1
     #...
     build:
       output: 2
         type: docker
         image: my-registry.io/my-org/my-connect-cluster:latest
         pushSecret: my-registry-credentials
       plugins: 3
         - name: connector-1
           artifacts:
             - type: tgz
               url: <url_to_download_connector_1_artifact>
               sha512sum: <SHA-512_checksum_of_connector_1_artifact>
         - name: connector-2
           artifacts:
             - type: jar
               url: <url_to_download_connector_2_artifact>
               sha512sum: <SHA-512_checksum_of_connector_2_artifact>
     #...

   1

   Kafka Connect クラスターの仕様。

   2

   (必須) 新しいイメージがプッシュされるコンテナーレジストリーの設定。

   3

   (必須) 新しいコンテナーイメージに追加するコネクタープラグインとそれらのアーティファクトのリスト。各プラグインは、1 つ以上の artifact を使用して設定する必要があります。

2. リソースを作成または更新します。

   $ oc apply -f <kafka_connect_configuration_file>

3. 新しいコンテナーイメージがビルドされ、Kafka Connect クラスターがデプロイされるまで待ちます。
4. Kafka Connect REST API または KafkaConnector カスタムリソースを使用して、追加したコネクタープラグインを使用します。

手順

1. ベースイメージとして registry.redhat.io/amq-streams/kafka-37-rhel9:2.7.0 を使用して、新しい Dockerfile を作成します。

   FROM registry.redhat.io/amq-streams/kafka-37-rhel9:2.7.0
   USER root:root
   COPY ./my-plugins/ /opt/kafka/plugins/
   USER 1001

   プラグインファイルの例

   $ tree ./my-plugins/
   ./my-plugins/
   ├── debezium-connector-mongodb
   │   ├── bson-<version>.jar
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mongodb-<version>.jar
   │   ├── debezium-core-<version>.jar
   │   ├── LICENSE.txt
   │   ├── mongodb-driver-core-<version>.jar
   │   ├── README.md
   │   └── # ...
   ├── debezium-connector-mysql
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mysql-<version>.jar
   │   ├── debezium-core-<version>.jar
   │   ├── LICENSE.txt
   │   ├── mysql-binlog-connector-java-<version>.jar
   │   ├── mysql-connector-java-<version>.jar
   │   ├── README.md
   │   └── # ...
   └── debezium-connector-postgres
       ├── CHANGELOG.md
       ├── CONTRIBUTE.md
       ├── COPYRIGHT.txt
       ├── debezium-connector-postgres-<version>.jar
       ├── debezium-core-<version>.jar
       ├── LICENSE.txt
       ├── postgresql-<version>.jar
       ├── protobuf-java-<version>.jar
       ├── README.md
       └── # ...

   COPY コマンドは、コンテナーイメージにコピーするプラグインファイルを指します。

   この例では、Debezium コネクター (MongoDB、MySQL、および PostgreSQL) のプラグインを追加しますが、簡潔にするためにすべてのファイルがリストされているわけではありません。Kafka Connect で実行されている Debezium は、他の Kafka Connect タスクと同じように表示されます。

2. コンテナーイメージをビルドします。
3. カスタムイメージをコンテナーレジストリーにプッシュします。

4. 新しいコンテナーイメージを示します。

   次のいずれかの方法でイメージを指定できます。

   • KafkaConnect カスタムリソースの KafkaConnect.spec.image プロパティーを編集します。

     設定されている場合、このプロパティーは Cluster Operator の STRIMZI_KAFKA_CONNECT_IMAGES 環境変数をオーバーライドします。

     apiVersion: kafka.strimzi.io/v1beta2
     kind: KafkaConnect
     metadata:
       name: my-connect-cluster
     spec: 1
       #...
       image: my-new-container-image 2
       config: 3
         #...

     1

     Kafka Connect クラスターの仕様。

     2

     Kafka Connect Pod の Docker イメージ。

     3

     Kafka Connect ワーカー (コネクターではない) の設定。

   • install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml ファイルの STRIMZI_KAFKA_CONNECT_IMAGES 環境変数を編集して、新しいコンテナーイメージを指すようにし、Cluster Operator を再インストールします。

手順

1. 次のいずれかの方法で、FileStreamSourceConnector および FileStreamSinkConnector プラグインを Kafka Connect に追加します。

   • プラグインを使用して新しいコンテナーイメージを自動的にビルドするように Kafka Connect を設定する
   • ベースの Kafka Connect イメージから Docker イメージを作成します (手動または継続的インテグレーションを使用)

2. Kafka Connect 設定で strimzi.io/use-connector-resources annotation を true に設定します。

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

   KafkaConnector リソースを有効にすると、Cluster Operator はそれらを監視します。

3. examples/connect/source-connector.yaml ファイルを編集します。

   KafkaConnector ソースコネクター設定の例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-source-connector  1
     labels:
       strimzi.io/cluster: my-connect-cluster 2
   spec:
     class: org.apache.kafka.connect.file.FileStreamSourceConnector 3
     tasksMax: 2 4
     autoRestart: 5
       enabled: true
     config: 6
       file: "/opt/kafka/LICENSE" 7
       topic: my-topic 8
       # ...

   1

   コネクターの名前として使用される KafkaConnector リソースの名前。OpenShift リソースで有効な名前を使用します。

   2

   コネクターインスタンスを作成する Kafka Connect クラスターの名前。コネクターは、リンク先の Kafka Connect クラスターと同じ namespace にデプロイする必要があります。

   3

   コネクタークラスのフルネーム。これは、Kafka Connect クラスターによって使用されているイメージに存在するはずです。

   4

   コネクターが作成できる Kafka Connect タスクの最大数。

   5

   失敗したコネクターとタスクの自動再起動を有効にします。デフォルトでは、再起動の回数は無制限ですが、maxRestarts プロパティーを使用して自動再起動の最大回数を設定できます。

   6

   キーと値のペア形式の コネクター設定。

   7

   外部データファイルの場所。この例では、/opt/kafka/LICENSE ファイルから読み取るように FileStreamSourceConnector を設定しています。

   8

   ソースデータのパブリッシュ先となる Kafka トピック。

4. OpenShift クラスターでソース KafkaConnector を作成します。

   oc apply -f examples/connect/source-connector.yaml

5. examples/connect/sink-connector.yaml ファイルを作成します。

   touch examples/connect/sink-connector.yaml

6. 以下の YAML を sink-connector.yaml ファイルに貼り付けます。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-sink-connector
     labels:
       strimzi.io/cluster: my-connect
   spec:
     class: org.apache.kafka.connect.file.FileStreamSinkConnector 1
     tasksMax: 2
     config: 2
       file: "/tmp/my-file" 3
       topics: my-topic 4

   1

   コネクタークラスのフルネームまたはエイリアス。これは、Kafka Connect クラスターによって使用されているイメージに存在するはずです。

   2

   キーと値のペア形式の コネクター設定。

   3

   ソースデータのパブリッシュ先となる一時ファイル。

   4

   ソースデータの読み取り元となる Kafka トピック。

7. OpenShift クラスターにシンク KafkaConnector を作成します。

   oc apply -f examples/connect/sink-connector.yaml

8. コネクターリソースが作成されたことを確認します。

   oc get kctr --selector strimzi.io/cluster=<my_connect_cluster> -o name
   
   my-source-connector
   my-sink-connector

   <my_connect_cluster> を Kafka Connect クラスターの名前に置き換えます。

9. コンテナーで、kafka -console-consumer.sh を実行して、ソースコネクターによってトピックに書き込まれたメッセージを読み取ります。

   oc exec <my_kafka_cluster>-kafka-0 -i -t -- bin/kafka-console-consumer.sh --bootstrap-server <my_kafka_cluster>-kafka-bootstrap.NAMESPACE.svc:9092 --topic my-topic --from-beginning

   <my_kafka_cluster> を Kafka クラスターの名前に置き換えます。

1. 設定で KafkaConnector リソースをデプロイし、コネクターインスタンスを作成します。
2. strimzi.io/use-connector-resources アノテーションを true に設定して、Kafka Connect 設定で KafkaConnector リソースを有効にします。