10.9. Kafka Connect コネクターの設定

手順

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

   oc get KafkaConnector

   Copy to Clipboard Toggle word wrap

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

   Copy to Clipboard Toggle word wrap

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

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

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

手順

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 を検索します。

   oc describe KafkaConnector <kafka_connector_name>

   Copy to Clipboard Toggle word wrap

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

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 リソースを、listOffsets 設定が含まれるように編集します。

   オフセットをリストするための設定例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-source-connector
     labels:
       strimzi.io/cluster: my-connect-cluster
   spec:
     listOffsets:
       toConfigMap: 

   1

   
         name: my-connector-offsets 

   2

   
     # ...

   Copy to Clipboard Toggle word wrap

   1

   オフセットのリストが書き込まれる config map への参照。

   2

   config map の名前。この例では my-connector-offsets という名前です。

2. KafkaConnector リソースにアノテーションを付けて、リストを config map に書き込むコマンドを実行します。

   oc annotate kafkaconnector my-source-connector strimzi.io/connector-offsets=list -n <namespace>

   Copy to Clipboard Toggle word wrap

   アノテーションは、リスト操作が成功するか、リソースから手動で削除されるまで残ります。

3. KafkaConnector リソースを更新した後に、次のコマンドを使用して、オフセットを含む config map が作成されたか確認します。

   oc get configmap my-connector-offsets -n <namespace>

   Copy to Clipboard Toggle word wrap

4. config map の内容を調べて、オフセットがリストされていることを確認します。

   oc describe configmap my-connector-offsets -n <namespace>

   Copy to Clipboard Toggle word wrap

   Streams for Apache Kafka は、オフセット情報を offsets.json プロパティーに置きます。既存 config map の更新時に他のプロパティーは上書きされません。

   ソースコネクターオフセットリストの例

   apiVersion: v1
   kind: ConfigMap
   metadata:
     # ...
     ownerReferences: 

   1

   
     - apiVersion: kafka.strimzi.io/v1beta2
       blockOwnerDeletion: false
       controller: false
       kind: KafkaConnector
       name: my-source-connector
       uid: 637e3be7-bd96-43ab-abde-c55b4c4550e0
     resourceVersion: "66951"
     uid: 641d60a9-36eb-4f29-9895-8f2c1eb9638e
   data:
     offsets.json: |-
       {
         "offsets" : [ {
           "partition" : {
             "filename" : "/data/myfile.txt" 

   2

   
           },
           "offset" : {
             "position" : 15295 

   3

   
           }
         } ]
       }

   Copy to Clipboard Toggle word wrap

   1

   ソースコネクターの KafkaConnector リソースを指す所有者参照。カスタムの所有者参照を指定するには、事前に config map を作成して所有者参照を設定します。

   2

   この例でファイルベースのコネクターのファイル名 /data/myfile.txt で表されるソースパーティション。

   3

   ソースパーティション内の最後に処理されたオフセットの位置。

   シンクコネクターオフセットリストの例

   apiVersion: v1
   kind: ConfigMap
   metadata:
     # ...
     ownerReferences: 

   1

   
     - apiVersion: kafka.strimzi.io/v1beta2
       blockOwnerDeletion: false
       controller: false
       kind: KafkaConnector
       name: my-sink-connector
       uid: 84a29d7f-77e6-43ac-bfbb-719f9b9a4b3b
     resourceVersion: "79241"
     uid: 721e30bc-23df-41a2-9b48-fb2b7d9b042c
   data:
     offsets.json: |-
       {
         "offsets": [
           {
             "partition": {
               "kafka_topic": "my-topic", 

   2

   
               "kafka_partition": 2 

   3

   
             },
             "offset": {
               "kafka_offset": 4 

   4

   
             }
           }
         ]
       }

   Copy to Clipboard Toggle word wrap

   1

   シンクコネクターの KafkaConnector リソースを指す所有者参照。

   2

   シンクコネクターが使用する Kafka トピック。

   3

   Kafka トピックのパーティション。

   4

   このトピックとパーティションの、最後にコミットされた Kafka オフセット。

手順

1. KafkaConnector リソースを編集してコネクターを停止し、alterOffsets 設定を追加します。

   コネクターを停止してオフセットを変更する設定例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-source-connector
     labels:
       strimzi.io/cluster: my-connect-cluster
   spec:
     state: stopped 

   1

   
     alterOffsets:
       fromConfigMap: 

   2

   
         name: my-connector-offsets 

   3

   
     # ...

   Copy to Clipboard Toggle word wrap

   1

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

   2

   更新を提供する config map への参照。

   3

   config map の名前。この例では my-connector-offsets という名前です。

2. 変更を加えるために、config map を編集します。

   この例では、ソースコネクターのオフセット位置を 15000 にリセットします。

   ソースコネクターオフセットリストの設定例

   apiVersion: v1
   kind: ConfigMap
   metadata:
     # ...
   data:
     offsets.json: |- 

   1

   
       {
         "offsets" : [ {
           "partition" : {
             "filename" : "/data/myfile.txt"
           },
           "offset" : {
             "position" : 15000 

   2

   
           }
         } ]
       }

   Copy to Clipboard Toggle word wrap

   1

   offsets.json プロパティー内で編集する必要があります。

   2

   ソースパーティション内の更新されたオフセット位置。

3. KafkaConnector リソースにアノテーションを付けて、オフセット位置を更新するコマンドを実行します。

   oc annotate kafkaconnector my-source-connector strimzi.io/connector-offsets=alter -n <namespace>

   Copy to Clipboard Toggle word wrap

   アノテーションは、更新操作が成功するか、リソースから手動で削除されるまで残ります。

4. コネクターオフセットのリスト表示 の手順を使用して変更を確認します。

5. 状態を running に変更してコネクターを再起動します。

   コネクターを起動するための設定例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-source-connector
     labels:
       strimzi.io/cluster: my-connect-cluster
   spec:
     state: running
     # ...

   Copy to Clipboard Toggle word wrap

手順

1. コネクターを停止するように、KafkaConnector リソースを編集します。

   コネクターを停止するための設定例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-source-connector
     labels:
       strimzi.io/cluster: my-connect-cluster
   spec:
     # ...
     state: stopped 

   1

   
     # ...

   Copy to Clipboard Toggle word wrap

   1

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

2. KafkaConnector リソースにアノテーションを付けて、オフセット位置をリセットするコマンドを実行します。

   oc annotate kafkaconnector my-source-connector strimzi.io/connector-offsets=reset -n <namespace>

   Copy to Clipboard Toggle word wrap

   アノテーションは、リセット操作が成功するか、リソースから手動で削除されるまで残ります。

3. コネクターオフセットのリスト表示 の手順を使用して変更を確認します。

   リセット後、offsets.json プロパティーは空になります。

   ソースコネクターオフセットリストの例

   apiVersion: v1
   kind: ConfigMap
   metadata:
     # ...
   data:
     offsets.json: |-
       {
         "offsets" : []
       }

   Copy to Clipboard Toggle word wrap

4. 状態を running に変更してコネクターを再起動します。

   コネクターを起動するための設定例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-source-connector
     labels:
       strimzi.io/cluster: my-connect-cluster
   spec:
     state: running
     # ...

   Copy to Clipboard Toggle word wrap