第4章 KRaft モードへの移行

手順

1. Kafka クラスターのクラスター ID を取得します。

   zookeeper-shell ツールを使用します。

   /opt/kafka/bin/zookeeper-shell.sh localhost:2181 get /cluster/id

   このコマンドはクラスター ID を返します。

2. KRaft コントローラークォーラムをクラスターにインストールします。

   1. controller.properties ファイルを使用して、各ホスト上でコントローラーノードを設定します。

      各コントローラーには少なくとも次の設定が必要です。

      • 一意のノード ID
      • 移行有効フラグの true への設定
      • ZooKeeper の接続の詳細
      • コントローラークォーラムが使用するリスナー名
      • コントローラー投票者のクォーラム

      • ブローカー間通信のリスナー名

        コントローラー設定の例

        process.roles=controller
        node.id=1
        
        zookeeper.metadata.migration.enable=true
        zookeeper.connect=zoo1.my-domain.com:2181,zoo2.my-domain.com:2181,zoo3.my-domain.com:2181
        
        listeners=CONTROLLER://0.0.0.0:9090
        controller.listener.names=CONTROLLER
        listener.security.protocol.map=CONTROLLER:PLAINTEXT
        controller.quorum.voters=1@localhost:9090
        inter.broker.listener.name=PLAINTEXT

        コントローラークォーラムの形式は、コンマ区切りリストの <node_id>@<hostname>:<port> です。KRaft コントローラーが移行を開始するには、ブローカー間リスナー名が必要です。

   2. 各コントローラーノードのログディレクトリーをセットアップします。

      /opt/kafka/bin/kafka-storage.sh format -t <uuid> -c /opt/kafka/config/kraft/controller.properties

      戻り値:

      Formatting /tmp/kraft-controller-logs

      <uuid> は、取得したクラスター ID に置き換えます。クラスター内の各コントローラーノードに同じクラスター ID を使用します。

      デフォルトでは、controller.properties 設定ファイルで指定されたログディレクトリー (log.dirs) は /tmp/kraft-controller-logs に設定されます。/tmp ディレクトリーは通常、システムが再起動するたびにクリアされるため、開発環境のみに適しています。必要に応じて、コンマ区切りリストを使用して複数のログディレクトリーを設定します。

   3. 各コントローラーを起動します。

      /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/kraft/controller.properties

   4. Kafka が稼働していることを確認します。

      jcmd | grep kafka

      戻り値:

      process ID kafka.Kafka /opt/kafka/config/kraft/controller.properties

      各コントローラーのログをチェックして、KRaft クラスターに正常に参加していることを確認します。

      tail -f /opt/kafka/logs/controller.log

3. 各ブローカーで移行を有効にします。

   1. 実行中の場合は、ホスト上で実行している Kafka ブローカーを停止します。

      /opt/kafka/bin/kafka-server-stop.sh
      jcmd | grep kafka

      マルチノードクラスターを使用する場合は、「Kafka ブローカーの正常なローリング再起動の実行」 を参照してください。

   2. server.properties ファイルを使用して移行を有効にします。

      各ブローカーには少なくとも次の追加設定が必要です。

      • ブローカー間プロトコルバージョンをバージョン 3.8 に設定する
      • 移行有効フラグ
      • コントローラーノードと一致するコントローラー設定
      • コントローラー投票者のクォーラム

      ブローカー設定の例

      broker.id=0
      inter.broker.protocol.version=3.8
      
      zookeeper.metadata.migration.enable=true
      zookeeper.connect=zoo1.my-domain.com:2181,zoo2.my-domain.com:2181,zoo3.my-domain.com:2181
      
      listeners=CONTROLLER://0.0.0.0:9090
      controller.listener.names=CONTROLLER
      listener.security.protocol.map=CONTROLLER:PLAINTEXT
      controller.quorum.voters=1@localhost:9090

      ZooKeeper の接続の詳細はすでに存在するはずです。

   3. 更新したブローカーを再起動します。

      /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties

      移行は自動的に開始しますが、クラスター内のトピックとパーティションの数によっては時間がかかる場合があります。

   4. Kafka が稼働していることを確認します。

      jcmd | grep kafka

      戻り値:

      process ID kafka.Kafka /opt/kafka/config/kraft/server.properties

4. アクティブなコントローラーのログをチェックして、移行が完了したことを確認します。

   /opt/kafka/bin/zookeeper-shell.sh localhost:2181 get /controller

   Completed migration of metadata from ZooKeeper to KRaft. という INFO ログエントリーを探します。

5. 各ブローカーを KRaft モードに切り替えます。

   1. 前述の手順と同様にブローカーを停止します。

   2. server.properties ファイル内のブローカー設定を更新します。

      • broker.id を、同じ ID を使用する node.id に置き換えます。
      • ブローカーの broker KRaft ロールを追加します。
      • ブローカー間プロトコルバージョン (inter.broker.protocol.version) を削除します。
      • 移行有効フラグ (zookeeper.metadata.migration.enable) を削除します。
      • ZooKeeper 設定を削除します。
      • コントローラーおよびブローカー通信用のリスナー (control.plane.listener.name) を削除します。

      KRaft のブローカー設定の例

      node.id=0
      process.roles=broker
      
      listeners=CONTROLLER://0.0.0.0:9090
      controller.listener.names=CONTROLLER
      listener.security.protocol.map=CONTROLLER:PLAINTEXT
      controller.quorum.voters=1@localhost:9090

   3. ブローカー設定で ACLS を使用している場合は、authorizer.class.name プロパティーを使用して、オーソライザーを KRaft ベースの標準オーソライザーに更新します。

      ZooKeeper ベースのブローカーは authorizer.class.name=kafka.security.authorizer.AclAuthorizer を使用します。

      KRaft ベースのブローカーに移行する場合は、authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer を指定します。

   4. 前述の手順と同様にブローカーを再起動します。

6. 各コントローラーを移行モードから切り替えます。

   1. ブローカーと同じ前述の方法でコントローラーを停止します。

   2. controller.properties ファイル内のコントローラー設定を更新します。

      • ZooKeeper 接続の詳細を削除します。
      • zookeeper.metadata.migration.enable プロパティーを削除します。
      • inter.broker.listener.name を削除します。

      移行後のコントローラー設定の例

      process.roles=controller
      node.id=1
      
      listeners=CONTROLLER://0.0.0.0:9090
      controller.listener.names=CONTROLLER
      listener.security.protocol.map=CONTROLLER:PLAINTEXT
      controller.quorum.voters=1@localhost:9090

   3. ブローカーと同じ前述の方法でコントローラーを再起動します。