6.4. 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: debezium-postgres-connector
           artifacts:
             - type: tgz
               url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/2.1.3.Final/debezium-connector-postgres-2.1.3.Final-plugin.tar.gz
               sha512sum: c4ddc97846de561755dc0b021a62aba656098829c70eb3ade3b817ce06d852ca12ae50c0281cc791a5a131cb7fc21fb15f4b8ee76c6cae5dd07f9c11cb7c6e79
         - name: camel-telegram
           artifacts:
             - type: tgz
               url: https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/camel-telegram-kafka-connector/0.11.5/camel-telegram-kafka-connector-0.11.5-package.tar.gz
               sha512sum: d6d9f45e0d1dbfcc9f6d1c7ca2046168c764389c78bc4b867dab32d24f710bb74ccf2a007d7d7a8af2dfca09d9a52ccbc2831fc715c195a3634cca055185bd91
     #...

   Copy to Clipboard Toggle word wrap

   1

   Kafka Connect クラスターの仕様。

   2

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

   3

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

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

   $ oc apply -f <kafka_connect_configuration_file>

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

手順

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

   FROM registry.redhat.io/amq-streams/kafka-35-rhel8:2.5.1
   USER root:root
   COPY ./my-plugins/ /opt/kafka/plugins/
   USER 1001

   Copy to Clipboard Toggle word wrap

   プラグインファイルの例

   $ 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 to Clipboard Toggle word wrap

   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

     
         #...

     Copy to Clipboard Toggle word wrap

     1

     Kafka Connect クラスターの仕様。

     2

     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:
       # ...

   Copy to Clipboard Toggle word wrap

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

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

   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

   
       # ...

   Copy to Clipboard Toggle word wrap

   1

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

   2

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

   3

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

   4

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

   5

   失敗したコネクターとタスクの自動再起動を有効にします。

   6

   キーと値のペアとしての コネクター設定。

   7

   このサンプルソースコネクター設定では、/opt/kafka/LICENSE ファイルからデータが読み取られます。

   8

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

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

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

   Copy to Clipboard Toggle word wrap

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

   touch examples/connect/sink-connector.yaml

   Copy to Clipboard Toggle word wrap

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

   Copy to Clipboard Toggle word wrap

   1

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

   2

   キーと値のペアとしての コネクター設定。

   3

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

   4

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

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

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

   Copy to Clipboard Toggle word wrap

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

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

   Copy to Clipboard Toggle word wrap

   <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

   Copy to Clipboard Toggle word wrap

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

手順

1. 再起動する Kafka コネクターを制御する KafkaConnector カスタムリソースの名前を見つけます。

   oc get KafkaConnector

   Copy to Clipboard Toggle word wrap

2. OpenShift で KafkaConnector リソースにアノテーションを付けて、コネクターを再起動します。

   oc annotate KafkaConnector <kafka_connector_name> strimzi.io/restart=true

   Copy to Clipboard Toggle word wrap

   restart アノテーションは true に設定されています。

3. 次の調整が発生するまで待ちます (デフォルトでは 2 分ごとです)。

   アノテーションが調整プロセスで検出されれば、Kafka コネクターは再起動されます。Kafka Connect が再起動リクエストを受け入れると、アノテーションは KafkaConnector カスタムリソースから削除されます。

手順

1. 再起動する Kafka コネクタータスクを制御する KafkaConnector カスタムリソースの名前を見つけます。

   oc get KafkaConnector

   Copy to Clipboard Toggle word wrap

2. KafkaConnector カスタムリソースから再起動するタスクの ID を検索します。タスク ID は、0 から始まる非負の整数です。

   oc describe KafkaConnector <kafka_connector_name>

   Copy to Clipboard Toggle word wrap

3. OpenShift で KafkaConnector リソースにアノテーションを付けて、ID を使用してコネクタータスクを再開します。

   oc annotate KafkaConnector <kafka_connector_name> strimzi.io/restart-task=0

   Copy to Clipboard Toggle word wrap

   この例では、タスク 0 が再起動されます。

4. 次の調整が発生するまで待ちます (デフォルトでは 2 分ごとです)。

   アノテーションが調整プロセスで検出されれば、Kafka コネクタータスクは再起動されます。Kafka Connect が再起動リクエストを受け入れると、アノテーションは KafkaConnector カスタムリソースから削除されます。

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