9.6. Kafka Connect の設定

手順

1. KafkaUser リソースの authorization プロパティーを編集し、アクセス権限をユーザーに付与します。

   アクセス権は、literal 名の値を使用して、Kafka Connect トピックとクラスターグループに対して設定されます。次の表に、トピックとクラスターグループ ID に設定されたデフォルトの名前を示します。

   表9.2 アクセス権設定の名前

   プロパティー	名前
   offset.storage.topic	connect-cluster-offsets
   status.storage.topic	connect-cluster-status
   config.storage.topic	connect-cluster-configs
   group	connect-cluster

   この設定例では、デフォルト名を使用してアクセス権を指定します。Kafka Connect インスタンスに別の名前を使用している場合は、ACL 設定でそれらの名前を使用します。

   簡易認可の設定例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaUser
   metadata:
     name: my-user
     labels:
       strimzi.io/cluster: my-cluster
   spec:
     # ...
     authorization:
       type: simple
       acls:
         # access to offset.storage.topic
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           operations:
             - Create
             - Describe
             - Read
             - Write
           host: "*"
         # access to status.storage.topic
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operations:
             - Create
             - Describe
             - Read
             - Write
           host: "*"
         # access to config.storage.topic
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operations:
             - Create
             - Describe
             - Read
             - Write
           host: "*"
         # cluster group
         - resource:
             type: group
             name: connect-cluster
             patternType: literal
           operations:
             - Read
           host: "*"

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

   oc apply -f KAFKA-USER-CONFIG-FILE

手順

1. 一時停止または停止するコネクターを制御する KafkaConnector カスタムリソースの名前を見つけます。

   oc get KafkaConnector

2. KafkaConnector リソースを編集して、コネクターを停止または一時停止します。

   Kafka Connect コネクターを停止する設定例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-source-connector
     labels:
       strimzi.io/cluster: my-connect-cluster
   spec:
     class: org.apache.kafka.connect.file.FileStreamSourceConnector
     tasksMax: 2
     config:
       file: "/opt/kafka/LICENSE"
       topic: my-topic
     state: stopped
     # ...

   state 設定を stopped または paused に変更します。このプロパティーが設定されていない場合のコネクターのデフォルトの状態は running です。

3. 変更を KafkaConnector 設定に適用します。

   state を running に変更するか、設定を削除して、コネクターを再開できます。

手順

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

   oc get KafkaConnector

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

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

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

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

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

手順

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

   oc get KafkaConnector

2. KafkaConnector カスタムリソースから再起動するタスクの ID を検索します。

   oc describe KafkaConnector <kafka_connector_name>

   タスク ID は 0 から始まる負の値ではない整数です。

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

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

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

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

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