第14章 AMQ Streams および Kafka のアップグレード

手順

1. 既存の Cluster Operator リソースのバックアップを作成します。

   oc get all -l app=strimzi -o yaml > strimzi-backup.yaml

2. Cluster Operator を更新します。

   Cluster Operator が稼働している namespace に従い、インストールファイルを編集します。

   Linux の場合は、以下を使用します。

   sed -i 's/namespace: .*/namespace: my-namespace/' install/cluster-operator/*RoleBinding*.yaml

   MacOS の場合は、以下を使用します。

   sed -i '' 's/namespace: .*/namespace: my-namespace/' install/cluster-operator/*RoleBinding*.yaml

   既存の Cluster Operator Deploymentで 1 つ以上の環境変数を編集した場合、install/cluster-operator/050-Deployment-cluster-operator.yaml ファイルを編集し、Cluster Operator の新規バージョンに加えた変更を反映させます。

3. 設定を更新したら、残りのインストールリソースとともにデプロイします。

   oc apply -f install/cluster-operator

   ローリングアップデートが完了するのを待ちます。

4. Kafka Pod のイメージを取得して、アップグレードが正常に完了したことを確認します。

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

   イメージタグには、新しい AMQ Streams バージョンと Kafka バージョンが順に示されます。例: <New AMQ Streams version>-kafka-<Current Kafka version>

5. 既存のリソースを更新して、非推奨になったカスタムリソースプロパティーを処理します。

   • AMQ Streams リソースのアップグレード

手順

1. 必要に応じてエディターで Kafka クラスター設定を更新します。

   oc edit kafka my-cluster

   1. 現行のバージョンの Kafka の log.message.format.version が新しい Kafka バージョンでも同じ場合は、次の手順に進みます。

      それ以外の場合、Kafka.spec.kafka.config の log.message.format.version が 現行 バージョンのデフォルトに設定されていることを確認してください。

      たとえば、Kafka 2.3.0 からのアップグレードでは以下のようになります。

      kind: Kafka
      spec:
        # ...
        kafka:
          version: 2.3.0
          config:
            log.message.format.version: "2.3"
            # ...

      log.message.format.version が設定されていない場合は、現行バージョンに設定します。

      注記

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

   2. Kafka.spec.kafka.version を変更し、新バージョンを指定します (log.message.format.version は現行バージョンのままにします)。

      たとえば、Kafka 2.3.0 から 2.4.0 へのアップグレードは以下のようになります。

      apiVersion: v1alpha1
      kind: Kafka
      spec:
        # ...
        kafka:
          version: 2.4.0 1
          config:
            log.message.format.version: "2.3" 2
            # ...

      1

      これは新バージョンに変更されます。

      2

      これは現行バージョンのままです。

   3. Kafka バージョンのイメージが Cluster Operator の STRIMZI_KAFKA_IMAGES に定義されているイメージとは異なる場合は、Kafka.spec.kafka.image を更新します。

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

2. エディターを保存して終了し、ローリングアップデートの完了を待ちます。

   注記

   新バージョンの Kafka に新しい ZooKeeper バージョンがある場合、追加のローリングアップデートが発生します。

   更新をログで確認するか、または Pod 状態の遷移を監視して確認します。

   oc logs -f <cluster-operator-pod-name> | grep -E "Kafka version upgrade from [0-9.]+ to [0-9.]+, phase ([0-9]+) of \1 completed"

   oc get po -w

   現行バージョンと新バージョンの Kafka で、Interbroker プロトコルのバージョンが異なる場合は、Cluster Operator ログで INFO レベルのメッセージを確認します。

   Reconciliation #<num>(watch) Kafka(<namespace>/<name>): Kafka version upgrade from <from-version> to <to-version>, phase 2 of 2 completed

   または、現行バージョンと新バージョンの Kafka で、Interbroker プロトコルのバージョンが同じ場合は、以下を確認します。

   Reconciliation #<num>(watch) Kafka(<namespace>/<name>): Kafka version upgrade from <from-version> to <to-version>, phase 1 of 1 completed

   ローリングアップデートでは以下を行います。

   • 各 Pod が新バージョンの Kafka のブローカーバイナリーを使用していることを確認します。

   • 新バージョンの Kafka の Interbroker プロトコルを使用してメッセージを送信するように、ブローカーを設定します。

     注記

     クライアントは引き続き旧バージョンを使用するため、ブローカーはメッセージを旧バージョンに変換してからクライアントに送信します。この余分な負荷を最小化するには、できるだけ速やかにクライアントを更新します。

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

   「クライアントをアップグレードするストラテジー」を参照してください。

   警告

   この手順を完了すると、ダウングレードできません。この時点で更新を元に戻す必要がある場合は、「Kafka ブローカーおよびクライアントアプリケーションのダウングレード」の手順に従います。

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

   1. Kafka Connect では、KafkaConnect.spec.version を更新します。
   2. MirrorMaker では、KafkaMirrorMaker.spec.version を更新します。

4. 1. で特定された log.message.format.version が、新しいバージョンと同じ場合は、次の手順に進みます。

   それ以外の場合は、Kafka.spec.kafka.config の log.message.format.version を、現在使用している新バージョンの Kafka のデフォルトバージョンに変更します。

   たとえば、2.4.0 へのアップグレードでは以下のようになります。

   apiVersion: v1alpha1
   kind: Kafka
   spec:
     # ...
     kafka:
       version: 2.4.0
       config:
         log.message.format.version: "2.4"
         # ...

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

   これで、Kafka クラスターおよびクライアントが新バージョンの Kafka を使用するようになります。

1. Kafka クライアント .jar ファイルを新バージョンに置き換えます。
2. コンシューマー設定で、partition.assignment.strategy に cooperative-sticky を追加します。たとえば、range ストラテジーが設定されている場合は、設定を range, cooperative-sticky に変更します。
3. グループ内の各コンシューマーを順次再起動し、再起動後に各コンシューマーがグループに再度参加するまで待ちます。
4. コンシューマー設定から前述の partition.assignment.strategy を削除して、グループの各コンシューマーを再設定し、cooperative-sticky ストラテジーのみを残します。
5. グループ内の各コンシューマーを順次再起動し、再起動後に各コンシューマーがグループに再度参加するまで待ちます。

1. Kafka Streams の .jar ファイルを新バージョンに置き換えます。
2. Kafka Streams の設定で、upgrade.from 設定パラメーターをアップグレード前の Kafka バージョンに設定します (例: 2.3)。
3. 各ストリームプロセッサー (ノード) を順次再起動します。
4. upgrade.from 設定パラメーターを Kafka Streams 設定から削除します。
5. グループ内の各コンシューマーを順次再起動します。

手順

1. 必要に応じてエディターで Kafka クラスター設定を更新します。

   oc edit を使用します。

   oc edit kafka my-cluster

   1. Kafka.spec.kafka.version を変更して、以前のバージョンを指定します。

      たとえば、Kafka 2.4.0 から 2.3.0 へのダウングレードは以下のようになります。

      apiVersion: v1alpha1
      kind: Kafka
      spec:
        # ...
        kafka:
          version: 2.3.0 1
          config:
            log.message.format.version: "2.3" 2
            # ...

      1

      これは以前のバージョンに変更されます。

      2

      これは変更されません。

      注記

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

   2. Kafka バージョンのイメージが Cluster Operator の STRIMZI_KAFKA_IMAGES に定義されているイメージとは異なる場合は、Kafka.spec.kafka.image を更新します。

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

2. エディターを保存して終了し、ローリングアップデートの完了を待ちます。

   更新をログで確認するか、または Pod 状態の遷移を監視して確認します。

   oc logs -f <cluster-operator-pod-name> | grep -E "Kafka version downgrade from [0-9.]+ to [0-9.]+, phase ([0-9]+) of \1 completed"

   oc get po -w

   Kafka の前バージョンと現行バージョンで Interbroker プロトコルのバージョンが異なる場合、Cluster Operator ログで INFO レベルのメッセージを確認します。

   Reconciliation #<num>(watch) Kafka(<namespace>/<name>): Kafka version downgrade from <from-version> to <to-version>, phase 2 of 2 completed

   または、Kafka の前バージョンと現行バージョンで Interbroker プロトコルのバージョンが同じ場合は、以下を確認します。

   Reconciliation #<num>(watch) Kafka(<namespace>/<name>): Kafka version downgrade from <from-version> to <to-version>, phase 1 of 1 completed

3. すべてのクライアントアプリケーション (コンシューマー) をダウングレードして、以前のバージョンのクライアントバイナリーを使用します。

   これで、Kafka クラスターおよびクライアントは以前の Kafka バージョンを使用するようになります。