10.5. Kafka のアップグレード

ブローカーレベルのコンシューマーの最初のストラテジー

1. コンシューマーとして機能するアプリケーションをすべてアップグレードします。
2. ブローカーレベル log.message.format.version を新バージョンに変更します。
3. プロデューサーとして機能するアプリケーションをアップグレードします。

1. コンシューマーとして機能するアプリケーションをすべてアップグレードします。
2. トピックレベルの message.format.version を新バージョンに変更します。
3. プロデューサーとして機能するアプリケーションをアップグレードします。

1. トピックレベルの message.format.version を、旧バージョンに変更します (または、デフォルトがブローカーレベルの log.message.format.version のトピックを利用します)。
2. コンシューマーおよびプロデューサーとして機能するアプリケーションをすべてアップグレードします。
3. アップグレードしたアプリケーションが正しく機能することを確認します。
4. トピックレベルの message.format.version を新バージョンに変更します。

手順

1. Kafka クラスター設定を更新します。

   oc edit kafka <my_cluster>

2. 設定されている場合は、inter.broker.protocol.version および log.message.format.version プロパティーが 現在の バージョンに設定されていることを確認してください。

   たとえば、Kafka バージョン 3.2.3 から 3.3.1 にアップグレードする場合、現在のバージョンは 3.2 です。

   kind: Kafka
   spec:
     # ...
     kafka:
       version: 3.2.3
       config:
         log.message.format.version: "3.2"
         inter.broker.protocol.version: "3.2"
         # ...

   log.message.format.version および inter.broker.protocol.version が設定されていない場合、AMQ Streams では、次のステップの Kafka バージョンの更新後、これらのバージョンを現在のデフォルトに自動的に更新します。

   注記

   log.message.format.version および inter.broker.protocol.version の値は、浮動小数点数として解釈されないように文字列である必要があります。

3. Kafka.spec.kafka.version を変更して、新しい Kafka バージョンを指定します。現在の Kafka バージョンのデフォルトで log.message.format.version および inter.broker.protocol.version のままにします。

   注記

   kafka.version を変更すると、クラスターのすべてのブローカーがアップグレードされ、新しいブローカーバイナリーの使用が開始されます。このプロセスでは、一部のブローカーは古いバイナリーを使用し、他のブローカーはすでに新しいバイナリーにアップグレードされています。inter.broker.protocol.version を現在の設定のままにしておくと、ブローカーはアップグレード中に相互に通信し続けることができます。

   たとえば、Kafka 3.2.3 から 3.3.1 にアップグレードする場合:

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     # ...
     kafka:
       version: 3.3.1 

   1

   
       config:
         log.message.format.version: "3.2" 

   2

   
         inter.broker.protocol.version: "3.2" 

   3

   
         # ...

   1

   Kafka のバージョンが新しいバージョンに変更されます。

   2

   メッセージ形式のバージョンは変更されません。

   3

   ブローカー間のプロトコルバージョンは変更されません。

   警告

   新しい Kafka バージョンの inter.broker.protocol.version が変更された場合は、Kafka をダウングレードできません。ブローカー間プロトコルのバージョンは、__consumer_offsets に書き込まれたメッセージなど、ブローカーによって保存される永続メタデータに使用されるスキーマを判断します。ダウングレードされたクラスターはメッセージを理解しません。

4. Kafka クラスターのイメージが Kafka.spec.kafka.image の Kafka カスタムリソースで定義されている場合、image を更新して、新しい Kafka バージョンでコンテナーイメージを示すようにします。

   Kafka バージョンおよびイメージマッピング を参照してください。

5. エディターを保存して終了し、ローリング更新の完了を待ちます。

   Pod の状態の遷移を監視して、ローリング更新の進捗を確認します。

   oc get pods my-cluster-kafka-0 -o jsonpath='{.spec.containers[0].image}'

   ローリング更新により、各 Pod が新バージョンの Kafka のブローカーバイナリーを使用するようになります。

6. クライアントのアップグレードに選択したストラテジー に応じて、新バージョンのクライアントバイナリーを使用するようにすべてのクライアントアプリケーションをアップグレードします。

   必要に応じて、Kafka Connect および MirrorMaker の version プロパティーを新バージョンの Kafka として設定します。

   1. Kafka Connect では、KafkaConnect.spec.version を更新します。
   2. MirrorMaker では、KafkaMirrorMaker.spec.version を更新します。
   3. MirrorMaker 2.0 の場合は、KafkaMirrorMaker2.spec.version を更新します。

7. 設定されている場合、新しい inter.broker.protocol.version バージョンを使用するように Kafka リソースを更新します。それ以外の場合は、ステップ 9 に進みます。

   たとえば、Kafka 3.3.1 にアップグレードする場合:

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     # ...
     kafka:
       version: 3.3.1
       config:
         log.message.format.version: "3.2"
         inter.broker.protocol.version: "3.3"
         # ...

8. Cluster Operator によってクラスターが更新されるまで待ちます。

9. 設定されている場合、新しい log.message.format.version バージョンを使用するように Kafka リソースを更新します。それ以外の場合は、ステップ 10 に進みます。

   たとえば、Kafka 3.3.1 にアップグレードする場合:

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     # ...
     kafka:
       version: 3.3.1
       config:
         log.message.format.version: "3.3"
         inter.broker.protocol.version: "3.3"
         # ...

   重要

   Kafka 3.0.0 以降、inter.broker.protocol.version が 3.0 以上に設定されていると、log.message.format.version オプションは無視されるため、設定する必要はありません。

10. Cluster Operator によってクラスターが更新されるまで待ちます。

    • これで、Kafka クラスターおよびクライアントが新バージョンの Kafka を使用するようになります。
    • ブローカーは、ブローカー間プロトコルバージョンと新バージョンの Kafka のメッセージ形式バージョンを使用して、メッセージを送信するように設定されます。