3.4. KRaft モードでの Kafka クラスターの実行

手順

1. Kafka クラスターの一意の ID を生成します。

   そのために kafka-storage ツールを使用できます。

   /opt/kafka/bin/kafka-storage.sh random-uuid

   このコマンドは ID を返します。KRaft モードでクラスター ID が必要です。

2. クラスター内の各ノードの設定プロパティーファイルを作成します。

   ファイルは、Kafka で提供される例に基づくことができます。

   1. デュアルロールを broker、controller、または broker, controller として指定します。

      たとえば、組み合わせたロールに process.roles=broker, controller を指定します。

   2. クラスター内の各ノードに 0 から始まる一意の node.id を指定します。

      たとえば、node.id=1 です。

   3. <node_id>@<hostname:port> の形式で controller.quorum.voters のリストを指定します。

      たとえば、controller.quorum.voters=1@localhost:9093 です。

   4. リスナーを指定します。

      • 各リスナーの名前、ホスト名、ポートを設定します。

        たとえば、listeners=PLAINTEXT:localhost:9092,CONTROLLER:localhost:9093 です。

      • ブローカー間通信に使用するリスナー名を設定します。

        たとえば、inter.broker.listener.name=PLAINTEXT です。

      • コントローラークォーラムで使用するリスナー名を設定します。

        たとえば、controller.listener.names=CONTROLLER です。

      • Kafka への接続のためにクライアントにアドバタイズされる各リスナーの名前、ホスト名、およびポートを設定します。

        たとえば、advertised.listeners=PLAINTEXT:localhost:9092 です。

3. Kafka クラスターの各ノードにログディレクトリーを設定します。

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

   戻り値:

   Formatting /tmp/kraft-combined-logs

   <uuid> は、生成したクラスター ID に置き換えます。クラスター内の各ノードで、同じ ID を使用します。

   ブローカー用に作成したプロパティーファイルを使用してブローカー設定を適用します。

   デフォルトでは、server.properties 設定ファイルで指定されたログディレクトリー (log.dirs) は /tmp/kraft-combined-logs に設定されます。/tmp ディレクトリーは通常、システムが再起動するたびにクリアされるため、開発環境のみに適しています。

   複数のログディレクトリーを設定するには、コンマ区切りリストを追加できます。

4. 各 Kafka ノードを起動します。

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

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

   jcmd | grep kafka

   戻り値:

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

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

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