第2章 デプロイメント設定

手順

1. Kafka リソースで、Kafka クラスター、ZooKeeper クラスター、またはその両方に割り当てられた永続ボリュームのサイズを増やします。

   • Kafka クラスターに割り当てられたボリュームサイズを増やすには、spec.kafka.storage プロパティーを編集します。

   • ZooKeeper クラスターに割り当てたボリュームサイズを増やすには、spec.zookeeper.storage プロパティーを編集します。

     たとえば、ボリュームサイズを 1000Gi から 2000Gi に増やすには、以下のように編集します。

     apiVersion: kafka.strimzi.io/v1beta1
     kind: Kafka
     metadata:
       name: my-cluster
     spec:
       kafka:
         # ...
         storage:
           type: persistent-claim
           size: 2000Gi
           class: my-storage-class
         # ...
       zookeeper:
         # ...

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

   次のように oc apply を使用します。

   oc apply -f your-file

   OpenShift では、Cluster Operator からの要求に応じて、選択された永続ボリュームの容量が増やされます。サイズ変更が完了すると、サイズ変更された永続ボリュームを使用するすべての Pod が Cluster Operator によって再起動されます。これは自動的に行われます。

手順

1. Kafka リソースの spec.kafka.storage.volumes プロパティーを編集します。新しいボリュームを volumes アレイに追加します。たとえば、id が 2 の新しいボリュームを追加します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       storage:
         type: jbod
         volumes:
         - id: 0
           type: persistent-claim
           size: 100Gi
           deleteClaim: false
         - id: 1
           type: persistent-claim
           size: 100Gi
           deleteClaim: false
         - id: 2
           type: persistent-claim
           size: 100Gi
           deleteClaim: false
       # ...
     zookeeper:
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f KAFKA-CONFIG-FILE

3. 新しいトピックを作成するか、既存のパーティションを新しいディスクに再度割り当てします。

手順

1. 削除するディスクからすべてのパーティションを再度割り当てます。削除するディスクに割り当てられたままになっているパーティションのデータは削除される可能性があります。

2. Kafka リソースの spec.kafka.storage.volumes プロパティーを編集します。volumes アレイから 1 つまたは複数のボリュームを削除します。たとえば、ID が 1 と 2 のボリュームを削除します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       storage:
         type: jbod
         volumes:
         - id: 0
           type: persistent-claim
           size: 100Gi
           deleteClaim: false
       # ...
     zookeeper:
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. Kafka リソースの replicas プロパティーを編集します。以下は例になります。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       replicas: 3
       # ...
     zookeeper:
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. クラスターデプロイメントを指定する Kafka リソースが含まれる YAML 設定ファイルを開きます。

2. Kafka リソースの spec.kafka.config プロパティーで、Kafka 設定を 1 つまたは複数入力します。以下は例になります。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       config:
         default.replication.factor: 3
         offsets.topic.replication.factor: 3
         transaction.state.log.replication.factor: 3
         transaction.state.log.min.isr: 1
       # ...
     zookeeper:
       # ...

3. 新しい設定を適用してリソースを作成または更新します。

   次のように oc apply を使用します。

   oc apply -f kafka.yaml

   kafka.yaml は、設定するリソースの YAML 設定ファイルに置き換えます (例: kafka-persistent.yaml)。

手順

1. クラスターデプロイメントを指定する Kafka リソースが含まれる YAML 設定ファイルを開きます。

2. Kafka リソースの spec.zookeeper.replicas プロパティーで、複製された ZooKeeper サーバーの数を入力します。以下は例になります。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
       replicas: 3
       # ...

3. 新しい設定を適用してリソースを作成または更新します。

   次のように oc apply を使用します。

   oc apply -f kafka.yaml

   kafka.yaml は、設定するリソースの YAML 設定ファイルに置き換えます (例: kafka-persistent.yaml)。

手順

1. クラスターデプロイメントを指定する Kafka リソースが含まれる YAML 設定ファイルを開きます。

2. Kafka リソースの spec.zookeeper.config プロパティーで、1 つまたは複数の ZooKeeper 設定を指定します。以下は例になります。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
       config:
         autopurge.snapRetainCount: 3
         autopurge.purgeInterval: 1
       # ...

3. 新しい設定を適用してリソースを作成または更新します。

   次のように oc apply を使用します。

   oc apply -f kafka.yaml

   kafka.yaml は、設定するリソースの YAML 設定ファイルに置き換えます (例: kafka-persistent.yaml)。

手順

1. OpenShift コンソールを使用してターミナルを開くか、CLI から exec コマンドを実行します。

   以下は例になります。

   oc exec -it my-cluster-zookeeper-0 -- bin/kafka-topics.sh --list --zookeeper localhost:12181

   必ず localhost:12181 を使用してください。

   ZooKeeper に対して Kafka コマンドを実行できるようになりました。

手順

1. Kafka リソースの entityOperator プロパティーを編集します。以下は例になります。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
     entityOperator:
       topicOperator:
         watchedNamespace: my-topic-namespace
         reconciliationIntervalSeconds: 60
       userOperator:
         watchedNamespace: my-user-namespace
         reconciliationIntervalSeconds: 60

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. クラスターデプロイメントを指定するリソースの resources プロパティーを編集します。以下は例になります。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       resources:
         requests:
           cpu: "8"
           memory: 64Gi
         limits:
           cpu: "12"
           memory: 128Gi
       # ...
     zookeeper:
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. ノードがデプロイされたゾーンやラックを表すノードラベルについては、OpenShift 管理者に相談します。

2. ラベルをトポロジーキーとして使用し、Kafka リソースの rack プロパティーを編集します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       rack:
         topologyKey: topology.kubernetes.io/zone
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. Kafka リソースの livenessProbe または readinessProbe プロパティーを編集します。以下は例になります。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       readinessProbe:
         initialDelaySeconds: 15
         timeoutSeconds: 5
       livenessProbe:
         initialDelaySeconds: 15
         timeoutSeconds: 5
       # ...
     zookeeper:
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. Kafka リソースの metrics プロパティーを編集します。以下は例になります。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
       metrics:
         lowercaseOutputName: true
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. Kafka、KafkaConnect、KafkaConnectS2I、KafkaMirrorMaker、または KafkaBridge リソースの jvmOptions プロパティーを編集します。以下は例になります。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       jvmOptions:
         "-Xmx": "8g"
         "-Xms": "8g"
       # ...
     zookeeper:
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. Kafka リソースの image プロパティーを編集します。以下は例になります。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       image: my-org/my-image:latest
       # ...
     zookeeper:
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. Kafka リソースの tlsSidecar プロパティーを編集します。以下は例になります。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
     entityOperator:
       # ...
       tlsSidecar:
         resources:
           requests:
             cpu: 200m
             memory: 64Mi
           limits:
             cpu: 500m
             memory: 128Mi
       # ...
     cruiseControl:
       # ...
       tlsSidecar:
         resources:
           requests:
             cpu: 200m
             memory: 64Mi
           limits:
             cpu: 500m
             memory: 128Mi
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. 移動するトピックを一覧表示する topics.json という名前の JSON ファイルを準備します。これには、以下の構造が必要です。

   {
     "version": 1,
     "topics": [
       <TopicObjects>
     ]
   }

   ここで <TopicObjects> は、以下のようなコンマ区切りのオブジェクトリストになります。

   {
     "topic": <TopicName>
   }

   たとえば、topic-a と topic-b のすべてのパーティションを再割り当てするには、以下のような topics.json ファイルを準備する必要があります。

   {
     "version": 1,
     "topics": [
       { "topic": "topic-a"},
       { "topic": "topic-b"}
     ]
   }

2. topics.json ファイルをブローカー Pod の 1 つにコピーします。

   cat topics.json | oc exec -c kafka <BrokerPod> -i -- \
     /bin/bash -c \
     'cat > /tmp/topics.json'

3. kafka-reassign-partitions.sh コマンドを使用して、再割り当て JSON を生成します。

   oc exec <BrokerPod> -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --topics-to-move-json-file /tmp/topics.json \
     --broker-list <BrokerList> \
     --generate

   たとえば、 topic-a および topic-b のすべてのパーティションをブローカー 4 および 7 に移動する場合は、以下を実行しｍす。

   oc exec <BrokerPod> -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --topics-to-move-json-file /tmp/topics.json \
     --broker-list 4,7 \
     --generate

手順

1. kafka.spec.kafka.replicas 設定オプションを増やして、新しいブローカーを必要なだけ追加します。
2. 新しいブローカー Pod が起動したことを確認します。

3. 後でコマンドを実行するブローカー Pod に reassignment.json ファイルをコピーします。

   cat reassignment.json | \
     oc exec broker-pod -c kafka -i -- /bin/bash -c \
     'cat > /tmp/reassignment.json'

   以下は例になります。

   cat reassignment.json | \
     oc exec my-cluster-kafka-0 -c kafka -i -- /bin/bash -c \
     'cat > /tmp/reassignment.json'

4. 同じブローカー Pod から kafka-reassign-partitions.sh コマンドラインツールを使用して、パーティションの再割り当てを実行します。

   oc exec broker-pod -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --reassignment-json-file /tmp/reassignment.json \
     --execute

   レプリケーションをスロットルで調整する場合、--throttle とブローカー間のスロットル率 (バイト/秒単位) を渡すこともできます。以下は例になります。

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --reassignment-json-file /tmp/reassignment.json \
     --throttle 5000000 \
     --execute

   このコマンドは、2 つの再割り当て JSON オブジェクトを出力します。最初の JSON オブジェクトには、移動されたパーティションの現在の割り当てが記録されます。後で再割り当てを元に戻す必要がある場合に備え、この値をローカルファイル (Pod のファイル以外) に保存します。2 つ目の JSON オブジェクトは、再割り当て JSON ファイルに渡した目的の再割り当てです。

5. 再割り当ての最中にスロットルを変更する必要がある場合は、同じコマンドラインに別のスロットル率を指定して実行します。以下は例になります。

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --reassignment-json-file /tmp/reassignment.json \
     --throttle 10000000 \
     --execute

6. ブローカー Pod のいずれかから kafka-reassign-partitions.sh コマンドラインツールを使用して、再割り当てが完了したかどうかを定期的に確認します。これは先ほどの手順と同じコマンドですが、--execute オプションの代わりに --verify オプションを使用します。

   oc exec broker-pod -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --reassignment-json-file /tmp/reassignment.json \
     --verify

   以下に例を示します。

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --reassignment-json-file /tmp/reassignment.json \
     --verify

7. --verify コマンドによって、移動した各パーティションが正常に完了したことが報告されると、再割り当ては終了します。この最終的な --verify によって、結果的に再割り当てスロットルも削除されます。割り当てを元のブローカーに戻すために JSON ファイルを保存した場合は、ここでそのファイルを削除できます。

手順

1. 後でコマンドを実行するブローカー Pod に reassignment.json ファイルをコピーします。

   cat reassignment.json | \
     oc exec broker-pod -c kafka -i -- /bin/bash -c \
     'cat > /tmp/reassignment.json'

   以下は例になります。

   cat reassignment.json | \
     oc exec my-cluster-kafka-0 -c kafka -i -- /bin/bash -c \
     'cat > /tmp/reassignment.json'

2. 同じブローカー Pod から kafka-reassign-partitions.sh コマンドラインツールを使用して、パーティションの再割り当てを実行します。

   oc exec broker-pod -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --reassignment-json-file /tmp/reassignment.json \
     --execute

   レプリケーションをスロットルで調整する場合、--throttle とブローカー間のスロットル率 (バイト/秒単位) を渡すこともできます。以下は例になります。

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --reassignment-json-file /tmp/reassignment.json \
     --throttle 5000000 \
     --execute

   このコマンドは、2 つの再割り当て JSON オブジェクトを出力します。最初の JSON オブジェクトには、移動されたパーティションの現在の割り当てが記録されます。後で再割り当てを元に戻す必要がある場合に備え、この値をローカルファイル (Pod のファイル以外) に保存します。2 つ目の JSON オブジェクトは、再割り当て JSON ファイルに渡した目的の再割り当てです。

3. 再割り当ての最中にスロットルを変更する必要がある場合は、同じコマンドラインに別のスロットル率を指定して実行します。以下は例になります。

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --reassignment-json-file /tmp/reassignment.json \
     --throttle 10000000 \
     --execute

4. ブローカー Pod のいずれかから kafka-reassign-partitions.sh コマンドラインツールを使用して、再割り当てが完了したかどうかを定期的に確認します。これは先ほどの手順と同じコマンドですが、--execute オプションの代わりに --verify オプションを使用します。

   oc exec broker-pod -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --reassignment-json-file /tmp/reassignment.json \
     --verify

   以下に例を示します。

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --reassignment-json-file /tmp/reassignment.json \
     --verify

5. --verify コマンドによって、移動した各パーティションが正常に完了したことが報告されると、再割り当ては終了します。この最終的な --verify によって、結果的に再割り当てスロットルも削除されます。割り当てを元のブローカーに戻すために JSON ファイルを保存した場合は、ここでそのファイルを削除できます。

6. すべてのパーティションの再割り当てが終了すると、削除されるブローカーはクラスター内のいずれのパーティションにも対応しないはずです。これは、ブローカーのデータログディレクトリーにライブパーティションのログが含まれていないことを確認すると検証できます。ブローカーのログディレクトリーに、拡張正規表現 [a-zA-Z0-9.-]+\.[a-z0-9]+-delete$ と一致しないディレクトリーが含まれる場合、ブローカーにはライブパーティションがあるため、停止してはなりません。

   これを確認するには、以下のコマンドを実行します。

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     /bin/bash -c \
     "ls -l /var/lib/kafka/kafka-log_<N>_ | grep -E '^d' | grep -vE '[a-zA-Z0-9.-]+\.[a-z0-9]+-delete$'"

   N は削除された Pod(s) の数に置き換えます。

   上記のコマンドによって出力が生成される場合、ブローカーにはライブパーティションがあります。この場合、再割り当てが終了していないか、再割り当て JSON ファイルが適切ではありません。

7. ブローカーにライブパーティションがないことが確認できたら、Kafka リソースの Kafka.spec.kafka.replicas を編集できます。これにより、StatefulSet がスケールダウンされ、番号が最も大きいブローカー Pod(s) が削除されます。

手順

1. Kafka リソースの maintenanceTimeWindows プロパティー を追加または編集します。たとえば、0800 から 1059 までと、1400 から 1559 までのメンテナンスを可能にするには、以下のように maintenanceTimeWindows を設定します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
     maintenanceTimeWindows:
       - "* * 8-10 * * ?"
       - "* * 14-15 * * ?"

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file