KRaft モードでの RHEL 上の Streams for Apache Kafka の使用

手順

1. RHEL を FIPS モードでインストールします。

   詳細は、RHEL ドキュメント のセキュリティー強化に関する情報を参照してください。

2. Streams for Apache Kafka のインストールに進みます。

3. FIPS 準拠のアルゴリズムとプロトコルを使用するように Streams for Apache Kafka を設定します。

   使用する場合は、次の設定が準拠していることを確認してください。

   • SSL 暗号スイートと TLS バージョンは、JDK フレームワークでサポートされている必要があります。
   • SCRAM-SHA-512 パスワードの長さは少なくとも 32 文字である必要があります。

1. 新しい kafka ユーザーとグループを追加します。

   groupadd kafka
   useradd -g kafka kafka
   passwd kafka

2. amq-streams-<version>-bin.zip ファイルの内容を抽出して /opt/kafka ディレクトリーに移動します。

   unzip amq-streams-<version>-bin.zip -d /opt
   mv /opt/kafka*redhat* /opt/kafka

3. /opt/kafka ディレクトリーの所有権を kafka ユーザーに変更します。

   chown -R kafka:kafka /opt/kafka

4. Kafka データを格納する /var/lib/kafka ディレクトリーを作成し、その所有権を kafka ユーザーに設定します。

   mkdir /var/lib/kafka
   chown -R kafka:kafka /var/lib/kafka

   Kafka のデフォルトの設定をシングルノードクラスターとして実行 できるようになりました。

   インストールを使用して、Kafka Connect などの他の Kafka コンポーネントを同じホストで実行することもできます。

   他のコンポーネントを実行するには、コンポーネント設定の bootstrap.servers プロパティーを使用して、Kafka ブローカーに接続するためのホスト名とポートを指定します。

   同じホスト上の単一の Kafka ブローカーを指すブートストラップサーバー設定の例

   bootstrap.servers=localhost:9092

   ただし、別のホストに Kafka コンポーネントをインストールして実行することを推奨します。

5. (オプション) Kafka コンポーネントを個別のホストにインストールします。

   1. インストールファイルを各ホストの /opt/kafka ディレクトリーに抽出します。
   2. /opt/kafka ディレクトリーの所有権を kafka ユーザーに変更します。

   3. Kafka ブローカーを実行しているホスト (またはマルチノード環境のホスト) にコンポーネントを接続する bootstrap.servers 設定を追加します。

      異なるホスト上の Kafka ブローカーを指すブートストラップサーバー設定の例

      bootstrap.servers=kafka0.<host_ip_address>:9092,kafka1.<host_ip_address>:9092,kafka2.<host_ip_address>:9092

      この設定は、Kafka Connect、MirrorMaker 2、および Kafka Bridge に使用できます。

手順

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

手順

1. Kafka ノードを停止します。

   su - kafka
   /opt/kafka/bin/kafka-server-stop.sh

2. Kafka ノードが停止していることを確認します。

   jcmd | grep kafka

1. Kafka ブローカーを停止します。

   /opt/kafka/bin/kafka-server-stop.sh

2. 完了後に再起動を必要とするブローカー設定に変更を加えます。

   詳細は、以下を参照してください。

   • Kafka の設定
   • Kafka ノードのアップグレード

3. Kafka ブローカーを再起動します。

   /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

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

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

5. ブローカで、レプリケーションが不十分なパーティションがゼロになるまで待ちます。コマンドラインから確認するか、メトリクスを使用できます。

   • --under-replicated-partitions パラメーターを指定して kafka-topics.sh コマンドを使用します。

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_host>:<port>  --describe --under-replicated-partitions

     以下に例を示します。

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --under-replicated-partitions

     このコマンドは、クラスターでレプリケーションが不十分なパーティションのトピックのリストを表示します。

     レプリケーションが不十分なパーティションのトピック

     Topic: topic3 Partition: 4 Leader: 2 Replicas: 2,3 Isr: 2
     Topic: topic3 Partition: 5 Leader: 3 Replicas: 1,2 Isr: 1
     Topic: topic1 Partition: 1 Leader: 3 Replicas: 1,3 Isr: 3
     # …

     In-Sync レプリカ (ISR) の数がレプリカの数より少ない場合、レプリケーションが不十分なパーティションが一覧表示されます。リストが返されない場合は、レプリケーションが不十分なパーティションはありません。

   • UnderReplicatedPartitions メトリックを使用します。

     kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions

     このメトリックからは、レプリカが不十分なパーティションの数がわかります。数がゼロになるまで待機します。

     ヒント

     トピックにレプリケーションが不十分なパーティションがあると、Kafka Exporter を使用してアラートを作成します。

手順

1. 環境変数設定プロバイダー JAR ファイルを Kafka libs ディレクトリーに追加します。

2. Kafka コンポーネントの設定プロパティーファイルで環境変数設定プロバイダーを初期化します。たとえば、Kafka のプロバイダーを初期化するには、設定を server.properties ファイルに追加します。

   環境変数の設定プロバイダーを有効にする設定

   config.providers.env.class=org.apache.kafka.common.config.provider.EnvVarConfigProvider

3. プロパティーファイルに設定を追加して、環境変数からデータをロードします。

   環境変数からデータをロードするための設定

   option=${env:<MY_ENV_VAR_NAME>}

   MY_ENV_VAR_NAME などの大文字または大文字の環境変数の命名規則を使用します。

4. 変更を保存します。

5. Kafka コンポーネントを再起動します。

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

トピックの作成

1. kafka-topics.sh ユーティリティーを使用し以下の項目を指定して、トピックを作成します。

   • --bootstrap-server における Kafka ブローカーのホストおよびポート。
   • --create オプション: 作成される新しいトピック。
   • --topic オプション: トピック名。
   • --partitions オプション: パーティション数。

   • --replication-factor オプション: トピックレプリケーション係数。

     また、--config オプションを使用して、デフォルトのトピック設定オプションの一部を上書きすることもできます。このオプションは複数回使用して、異なるオプションを上書きできます。

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_address> --create --topic <TopicName> --partitions <NumberOfPartitions> --replication-factor <ReplicationFactor> --config <Option1>=<Value1> --config <Option2>=<Value2>

     mytopic というトピックを作成するコマンドの例

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic mytopic --partitions 50 --replication-factor 3 --config cleanup.policy=compact --config min.insync.replicas=2

2. kafka-topics.sh を使用して、トピックが存在することを確認します。

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_address> --describe --topic <TopicName>

   mytopic というトピックを記述するコマンドの例

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic mytopic

トピックの記述

1. kafka-topics.sh ユーティリティーを使用し以下の項目を指定して、トピックを説明します。

   • --bootstrap-server における Kafka ブローカーのホストおよびポート。
   • --describe オプション: トピックを記述することを指定するために使用します。
   • --topic オプション: このオプションでトピック名を指定する必要があります。

   • --topic オプションを省略すると、利用可能なすべてのトピックを記述します。

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_host>:<port> --describe --topic <topic_name>

     mytopic というトピックを記述するコマンドの例

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic mytopic

     このコマンドは、このトピックに属するすべてのパーティションおよびレプリカをリスト表示します。また、すべてのトピック設定オプションも表示されます。

トピック設定の変更

1. kafka-configs.sh ツールを使用して、現在の設定を取得します。

   • --bootstrap-server オプションで Kafka ブローカーのホストおよびポートを指定します。
   • --entity-type を topic として、--entity-name をトピックの名前に設定します。

   • --describe オプション: 現在の設定を取得するために使用します。

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_host>:<port> --entity-type topics --entity-name <topic_name> --describe

     mytopic という名前のトピックの設定を取得するコマンドの例

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name mytopic --describe

2. kafka-configs.sh ツールを使用して、現在の設定を変更します。

   • --bootstrap-server オプションで Kafka ブローカーのホストおよびポートを指定します。
   • --entity-type を topic として、--entity-name をトピックの名前に設定します。
   • --alter オプション: 現在の設定を変更するために使用します。

   • --add-config オプション: 追加または変更するオプションを指定します。

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_host>:<port> --entity-type topics --entity-name <topic_name> --alter --add-config <option>=<value>

     mytopic という名前のトピックの設定を変更するコマンドの例

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name mytopic --alter --add-config min.insync.replicas=1

3. kafka-configs.sh ツールを使用して、既存の設定オプションを削除します。

   • --bootstrap-server オプションで Kafka ブローカーのホストおよびポートを指定します。
   • --entity-type を topic として、--entity-name をトピックの名前に設定します。
   • --delete-config オプション: 既存の設定オプションを削除するために使用します。

   • --remove-config オプション: 削除するオプションを指定します。

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_host>:<port> --entity-type topics --entity-name <topic_name> --alter --delete-config <option>

     mytopic という名前のトピックの設定を変更するコマンドの例

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name mytopic --alter --delete-config min.insync.replicas

トピックの削除

1. kafka-topics.sh ユーティリティーを使用してトピックを削除します。

   • --bootstrap-server における Kafka ブローカーのホストおよびポート。
   • --delete オプション: 既存のトピックを削除することを指定するために使用します。

   • --topic オプション: このオプションでトピック名を指定する必要があります。

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_host>:<port> --delete --topic <topic_name>

     mytopic というトピックを作成するコマンドの例

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic mytopic

2. kafka-topics.sh を使用して、トピックが削除されたことを確認します。

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_host>:<port> --list

   すべてのトピックをリスト表示するコマンドの例

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list

手順

1. テキストエディターでサンプルプロパティーファイルを開くか、新しいプロパティーファイルを作成し、ファイルを編集して接続情報と各 Kafka クラスターのレプリケーションフローを追加します。

   以下の例は、cluster-1 および cluster-2 の 2 つのクラスターを双方向に接続する設定を示しています。クラスター名は、clusters プロパティーで設定できます。

   MirrorMaker 2 の設定例

   clusters=cluster-1,cluster-2 1
   
   cluster-1.bootstrap.servers=<cluster_name>-kafka-bootstrap-<project_name_one>:443 2
   cluster-1.security.protocol=SSL 3
   cluster-1.ssl.truststore.password=<truststore_name>
   cluster-1.ssl.truststore.location=<path_to_truststore>/truststore.cluster-1.jks_
   cluster-1.ssl.keystore.password=<keystore_name>
   cluster-1.ssl.keystore.location=<path_to_keystore>/user.cluster-1.p12
   
   cluster-2.bootstrap.servers=<cluster_name>-kafka-bootstrap-<project_name_two>:443 4
   cluster-2.security.protocol=SSL 5
   cluster-2.ssl.truststore.password=<truststore_name>
   cluster-2.ssl.truststore.location=<path_to_truststore>/truststore.cluster-2.jks_
   cluster-2.ssl.keystore.password=<keystore_name>
   cluster-2.ssl.keystore.location=<path_to_keystore>/user.cluster-2.p12
   
   cluster-1->cluster-2.enabled=true 6
   cluster-2->cluster-1.enabled=true 7
   cluster-1->cluster-2.topics=.* 8
   cluster-2->cluster-1.topics=topic-1, topic-2 9
   cluster-1->cluster-2.groups=.* 10
   cluster-2->cluster-1.groups=group-1, group-2 11
   
   replication.policy.separator=- 12
   sync.topic.acls.enabled=false 13
   refresh.topics.interval.seconds=60 14
   refresh.groups.interval.seconds=60 15

   1

   各 Kafka クラスターは、そのエイリアスで識別されます。

   2

   ブートストラップアドレス およびポート 443 を使用した、cluster-1 の接続情報。両方のクラスターはポート 443 を使用し、OpenShift Routes を使用して Kafka に接続します。

   3

   ssl. プロパティーは、cluster-1 の TLS 設定を定義します。

   4

   cluster-2 の接続情報です。

   5

   ssl. プロパティーは、cluster-2 の TLS 設定を定義します。

   6

   cluster-1 から cluster-2 へのレプリケーションフローが有効になっています。

   7

   cluster-2 から cluster-1. へのレプリケーションフローが有効になっています。

   8

   cluster-1 から cluster-2 へのすべてのトピックのレプリケーションです。ソースコネクターは指定のトピックをレプリケーションします。チェックポイントコネクターは、指定されたトピックのオフセットを追跡します。

   9

   cluster-2 から cluster-1 への特定のトピックのレプリケーション。

   10

   cluster-1 から cluster-2 へのすべてのコンシューマーグループのレプリケーション。チェックポイントコネクターは、指定されたコンシューマーグループをレプリケーションします。

   11

   cluster-2 から cluster-1 への特定のコンシューマーグループのレプリケーション。

   12

   リモートトピック名の変更に使用する区切り文字を定義します。

   13

   有効にすると、同期されたトピックに ACL が適用されます。デフォルトは false です。

   14

   新しいトピックの同期をチェックする間隔です。

   15

   新しいコンシューマーグループの同期をチェックする間隔です。

2. オプション: 必要に応じて、リモートトピックの名前の自動変更をオーバーライドするポリシーを追加します。その名前の前にソースクラスターの名前を追加する代わりに、トピックが元の名前を保持します。

   このオプションの設定は、active/passive バックアップおよびデータ移行に使用されます。

   replication.policy.class=org.apache.kafka.connect.mirror.IdentityReplicationPolicy

3. オプション: コンシューマーグループのオフセットを同期する場合は、設定を追加して同期を有効にし、管理します。

   refresh.groups.interval.seconds=60
   sync.group.offsets.enabled=true 1
   sync.group.offsets.interval.seconds=60 2
   emit.checkpoints.interval.seconds=60 3

   1

   コンシューマーグループのオフセットを同期する任意設定。これは、active/passive 設定でのリカバリーに便利です。同期はデフォルトでは有効になっていません。

   2

   コンシューマーグループオフセットの同期が有効な場合は、同期の頻度を調整できます。

   3

   オフセット追跡のチェック頻度を調整します。オフセット同期の頻度を変更する場合は、これらのチェックの頻度も調整することを推奨します。

4. ターゲットクラスターで Kafka を起動します。

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

5. プロパティーファイルで定義したクラスター接続設定およびレプリケーションポリシーで MirrorMaker を起動します。

   /opt/kafka/bin/connect-mirror-maker.sh \
   /opt/kafka/config/connect-mirror-maker.properties

   MirrorMaker はクラスター間の接続を設定します。

6. ターゲットクラスターごとに、トピックがレプリケーションされていることを確認します。

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_host>:<port> --list

手順

1. MirrorMaker consumer.properties ファイルと producer.properties ファイルを編集して、MirrorMaker 2 機能をオフにします。

   以下に例を示します。

   replication.policy.class=org.apache.kafka.mirror.LegacyReplicationPolicy 1
   
   refresh.topics.enabled=false 2
   refresh.groups.enabled=false
   emit.checkpoints.enabled=false
   emit.heartbeats.enabled=false
   sync.topic.configs.enabled=false
   sync.topic.acls.enabled=false

   1

   MirrorMaker の以前のバージョンをエミュレートします。

   2

   MirrorMaker 2 の機能 (内部 チェックポイント および ハートビート トピックを含む) が無効になります

2. 変更を保存し、以前のバージョンの MirrorMaker で使用していたプロパティーファイルで MirrorMaker を再起動します。

   su - kafka /opt/kafka/bin/kafka-mirror-maker.sh \
   --consumer.config /opt/kafka/config/consumer.properties \
   --producer.config /opt/kafka/config/producer.properties \
   --num.streams=2

   consumer プロパティーはソースクラスターの設定を提供し、producer プロパティーはターゲットクラスターの設定を提供します。

   MirrorMaker はクラスター間の接続を設定します。

3. ターゲットクラスターで Kafka を起動します。

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

4. ターゲットクラスターの場合、トピックがレプリケーションされていることを確認します。

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_host>:<port> --list

手順

1. kafka ユーザーに切り替えます。

   su - kafka

2. kafka-configs.sh ツールを使用して、ブローカーのブローカーロガーのリストを表示します。

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_address> --describe --entity-type broker-loggers --entity-name BROKER-ID

   たとえば、ブローカー 0 の場合:

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type broker-loggers --entity-name 0

   これにより、TRACE、DEBUG、INFO、WARN、ERROR、または FATAL の各ロガーのログレベルが返されます。

   以下に例を示します。

   #...
   kafka.controller.ControllerChannelManager=INFO sensitive=false synonyms={}
   kafka.log.TimeIndex=INFO sensitive=false synonyms={}

3. 1 つ以上のブローカーロガーのログレベルを変更します。--alter および --add-config オプションを使用して、各ロガーとそのレベルを二重引用符のコンマ区切りリストとして指定します。

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_address> --alter --add-config "LOGGER-ONE=NEW-LEVEL,LOGGER-TWO=NEW-LEVEL" --entity-type broker-loggers --entity-name BROKER-ID

   たとえば、ブローカー 0 の場合:

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config "kafka.controller.ControllerChannelManager=WARN,kafka.log.TimeIndex=WARN" --entity-type broker-loggers --entity-name 0

   成功すると、以下が返されます。

   Completed updating config for broker: 0.

手順

1. kafka ユーザーに切り替えます。

   su - kafka

2. connect-log4j.properties ファイルで設定されているロガーの現在のログレベルを確認します。

   $ cat /opt/kafka/config/connect-log4j.properties
   
   # ...
   log4j.rootLogger=INFO, stdout, connectAppender
   # ...
   log4j.logger.org.reflections=ERROR

   curl コマンドを使用して、Kafka Connect API の admin/loggers エンドポイントからログレベルを確認します。

   curl -s http://localhost:8083/admin/loggers/ | jq
   
   {
     "org.reflections": {
       "level": "ERROR"
     },
     "root": {
       "level": "INFO"
     }
   }

   jq は出力を JSON 形式で出力します。このリストには、標準の org および root レベルのロガーに加えて、ログレベルが変更された特定のロガーが表示されます。

   Kafka Connect の admin.listeners 設定に TLS (トランスポート層セキュリティー) 認証を設定する場合、ロガーエンドポイントのアドレスは、https://localhost:8083 など、https としてプロトコルを使用して admin.listeners に指定された値になります。

   特定のロガーのログレベルを取得することもできます。

   curl -s http://localhost:8083/admin/loggers/org.apache.kafka.connect.mirror.MirrorCheckpointConnector | jq
   
   {
     "level": "INFO"
   }

3. PUT メソッドを使用して、ロガーのログレベルを変更します。

   curl -Ss -X PUT -H 'Content-Type: application/json' -d '{"level": "TRACE"}' http://localhost:8083/admin/loggers/root
   
   {
     # ...
   
     "org.reflections": {
       "level": "TRACE"
     },
     "org.reflections.Reflections": {
       "level": "TRACE"
     },
     "root": {
       "level": "TRACE"
     }
   }

   root ロガーを変更すると、デフォルトでルートログレベルを使用していたロガーのログレベルも変更されます。

手順

1. Red Hat カスタマーポータル から、最新バージョンの Red Hat Streams for Apache Kafka Cruise Control アーカイブをダウンロードします。

2. /opt/cruise-control ディレクトリーを作成します。

   sudo mkdir /opt/cruise-control

3. Cruise Control ZIP ファイルの中身を新しいディレクトリーに展開します。

   unzip amq-streams-<version>-cruise-control-bin.zip -d /opt/cruise-control

4. /opt/cruise-control ディレクトリーの所有権を kafka ユーザーに変更します。

   sudo chown -R kafka:kafka /opt/cruise-control

1. Kafka ブローカーを停止します。

   /opt/kafka/bin/kafka-server-stop.sh

2. Kafka 設定プロパティーファイルを編集して、Cruise Control Metrics Reporter を設定します。

   1. CruiseControlMetricsReporter クラスを metric.reporters 設定オプションに追加します。既存の Metrics Reporters を削除しないでください。

      metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter

   2. 以下の設定オプションおよび値を追加します。

      cruise.control.metrics.topic.auto.create=true
      cruise.control.metrics.topic.num.partitions=1
      cruise.control.metrics.topic.replication.factor=1

      これらのオプションにより、Cruise Control Metrics Reporter は、__CruiseControlMetrics トピックをログクリーンアップポリシー DELETE で作成します。詳細は、自動作成されたトピック および Cruise Control Metrics トピックのログクリーンアップポリシー を参照してください。

3. 必要に応じて SSL を設定します。

   1. Kafka 設定プロパティーファイルでは、関連するクライアント設定プロパティーを設定して、Cruise Control Metrics Reporter と Kafka ブローカー間の SSL を設定します。

      Metrics Reporter は、cruise.control.metrics.reporter という接頭辞を持つ、すべての標準のプロデューサー固有の設定プロパティーを受け入れます。たとえば、cruise.control.metrics.reporter.ssl.truststore.password です。

   2. Cruise Control 設定ファイル (/opt/cruise-control/config/cruisecontrol.properties) では、関連するクライアント設定プロパティーを設定して、Kafka ブローカーと Cruise Control サーバーとの間の SSL を設定します。

      Cruise Control は、Kafka から SSL クライアントプロパティーオプションを継承し、すべての Cruise Control サーバークライアントにこのプロパティーを使用します。

4. Kafka ブローカーを再起動します。

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

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

5. 残りのブローカーで手順 1-5 を繰り返します。

手順

1. Cruise Control プロパティーファイル (/opt/cruise-control/config/cruisecontrol.properties) を編集します。

2. 以下の設定例のように、プロパティーを設定します。

   # The Kafka cluster to control.
   bootstrap.servers=localhost:9092 1
   
   # The replication factor of Kafka metric sample store topic
   sample.store.topic.replication.factor=2 2
   
   # The configuration for the BrokerCapacityConfigFileResolver (supports JBOD, non-JBOD, and heterogeneous CPU core capacities)
   #capacity.config.file=config/capacity.json
   #capacity.config.file=config/capacityCores.json
   capacity.config.file=config/capacityJBOD.json 3
   
   # The list of goals to optimize the Kafka cluster for with pre-computed proposals
   default.goals={List of default optimization goals} 4
   
   # The list of supported goals
   goals={list of main optimization goals} 5
   
   # The list of supported hard goals
   hard.goals={List of hard goals} 6
   
   # How often should the cached proposal be expired and recalculated if necessary
   proposal.expiration.ms=60000 7
   
   #Set failure detection to true
   kafka.broker.failure.detection.enable=true 8

   1

   Kafka ブローカーのホストおよびポート番号 (常にポート 9092)。

   2

   Kafka メトリックサンプルストアトピックのレプリケーション係数。シングルノードの Kafka クラスターで Cruise Control を評価する場合は、このプロパティーを 1 に設定します。実稼働環境で使用する場合は、このプロパティーを 2 以上に設定します。

   3

   ブローカーリソースの最大容量制限を設定する設定ファイル。Kafka デプロイメント設定に適用されるファイルを使用します。詳細は、容量の設定 を参照してください。

   4

   完全修飾ドメイン名 (FQDN) を使用したデフォルトの最適化ゴールのコンマ区切りリスト。多くの主要な最適化ゴール (5 を参照) は、デフォルトの最適化ゴールとしてすでに設定されています。必要に応じて、目標を追加または削除できます。詳細は、「最適化ゴールの概要」 を参照してください。

   5

   FQDN を使用した、主な最適化ゴールのコンマ区切りリスト。最適化プロポーザルの生成にゴールが使用されないように完全に除外するには、それらをリストから削除します。詳細は、「最適化ゴールの概要」 を参照してください。

   6

   FQDN を使用したハードゴールのコンマ区切りリスト。主な最適化ゴールのうち 7 つは、すでに厳しい目標として設定されています。必要に応じて、目標を追加または削除できます。詳細は、「最適化ゴールの概要」 を参照してください。

   7

   デフォルトの最適化ゴールから生成される、キャッシュされた最適化プロポーザルを更新する間隔 (ミリ秒単位)。詳細は、「最適化プロポーザルの概要」 を参照してください。

   8

   Cruise Control が Kafka API を使用してブローカーの障害を検出できるようにします。

3. Cruise Control サーバーを起動します。デフォルトでは、サーバーはポート 9092 で起動します。オプションで別のポートを指定します。

   cd /opt/cruise-control/
   ./kafka-cruise-control-start.sh config/cruisecontrol.properties <port_number>

4. Cruise Control が実行していることを確認するには、Cruise Control サーバーの /state エンドポイントに GET リクエストを送信します。

   curl -X GET 'http://<cc_host>:<cc_port>/kafkacruisecontrol/state'

1. __CruiseControlMetrics トピックの現在の設定を取得します。

   opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_address> --entity-type topics --entity-name __CruiseControlMetrics --describe

2. トピック設定でログクリーンアップポリシーを変更します。

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_address> --entity-type topics --entity-name __CruiseControlMetrics --alter --add-config cleanup.policy=delete

手順

1. /rebalance、/add_broker、または /remove_broker エンドポイントへの POST 要求を使用して最適化プロポーザルを生成します。

   デフォルトのゴールを使用した /rebalance への要求の例

   curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/rebalance'

   キャッシュされた最適化プロポーザルは即座に返されます。

   注記

   NotEnoughValidWindows が返された場合、Cruise Control は最適化プロポーザルを生成するために十分なメトリクスデータを記録していません。数分待ってからリクエストを再送信します。

   指定されたゴールを使用した /rebalance への要求の例

   curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/rebalance?goals=RackAwareGoal,ReplicaCapacityGoal'

   要求が、指定のゴールを満たしていると、キャッシュされた最適化プロポーザルが即座に返されます。それ以外の場合は、提供されたゴールを使用して新しい最適化プロポーザルが生成されます。計算には時間がかかります。この動作を強制するには、リクエストに ignore_proposal_cache=true パラメーターを追加します。

   ハードゴールなしで指定されたゴールを使用した /rebalance への要求の例

   curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/rebalance?goals=RackAwareGoal,ReplicaCapacityGoal,ReplicaDistributionGoal&skip_hard_goal_check=true'

   指定のブローカーが含まれる /add_broker への要求の例

   curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/add_broker?brokerid=3,4'

   要求には、新規ブローカーの ID のみが含まれます。たとえば、この要求は ID 3 と 4 のブローカーを追加します。レプリカは、リバランス時に既存のブローカーから新しいブローカーに移動します。

   指定のブローカーを除外する /remove_broker への要求の例

   curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/remove_broker?brokerid=3,4'

   要求には、除外されるブローカーの ID が含まれます。たとえば、この要求は ID が 3 と 4 のブローカーを除外します。レプリカは、リバランス時に他の既存ブローカーから削除されるブローカーから移動されます。

   注記

   削除されるブローカーがトピックを除外した場合でも、レプリカは移動されます。

2. 応答に含まれる最適化プロポーザルを確認します。プロパティーは、保留中のクラスターリバランス操作を記述します。

   プロポーザルには、提案された最適化の概要、各デフォルトの最適化ゴールの概要、およびプロポーザルの実行後に予想されるクラスター状態が含まれます。

   以下の情報に特に注意してください。

   • Cluster load after rebalance の概要。要件を満たす場合には、概要を使用して、提案された変更の影響を評価する必要があります。
   • n inter-broker replica (y MB) moves はブローカー間でネットワークを介して移動されるデータ量を示します。値が高いほど、リバランス中の Kafka クラスターへの潜在的なパフォーマンスの影響が大きくなります。
   • n intra-broker replica (y MB) moves は、ブローカー内部で (ディスク間) でどれだけのデータを移動させるかを示します。値が大きいほど、個々のブローカーに対する潜在的なパフォーマンスの影響は大きくなります (ただし n inter-broker replica (y MB) moves の影響よりも小さい)。
   • リーダーシップ移動の数。これは、リバランス中のクラスターのパフォーマンスにほとんど影響しません。

手順

1. dryrun=false パラメーターを使用して、POST 要求を /rebalance、/add_broker、または /remove_broker エンドポイントに送信します。

   /add_broker または /remove_broker エンドポイントを使用して、ブローカーを含むまたは除外するプロポーザルを生成した場合は、同じエンドポイントを使用して、指定されたブローカーの有無にかかわらずリバランスを実行します。

   /rebalance への要求の例

   curl -X POST 'cruise-control-server:9090/kafkacruisecontrol/rebalance?dryrun=false'

   /add_broker への要求の例

   curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/add_broker?dryrun=false&brokerid=3,4'

   /remove_broker への要求の例

   curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/remove_broker?dryrun=false&brokerid=3,4'

   Cruise Control はクラスターリバランスを開始し、最適化プロポーザルを返します。

2. 最適化プロポーザルで要約された変更を確認します。変更が予想される内容ではない場合は、リバランスを停止 できます。

3. /user_tasks エンドポイントを使用して、クラスターリバランスの進捗を確認します。進行中のクラスターリバランスのステータスは "Active" です。

   Cruise Control サーバーで実行されるすべてのクラスターリバランスタスクを表示するには、以下を実行します。

   curl 'cruise-control-server:9090/kafkacruisecontrol/user_tasks'
   
   USER TASK ID      CLIENT ADDRESS  START TIME     STATUS  REQUEST URL
   c459316f-9eb5-482f-9d2d-97b5a4cd294d  0:0:0:0:0:0:0:1       2020-06-01_16:10:29 UTC  Active      POST /kafkacruisecontrol/rebalance?dryrun=false
   445e2fc3-6531-4243-b0a6-36ef7c5059b4  0:0:0:0:0:0:0:1       2020-06-01_14:21:26 UTC  Completed   GET /kafkacruisecontrol/state?json=true
   05c37737-16d1-4e33-8e2b-800dee9f1b01  0:0:0:0:0:0:0:1       2020-06-01_14:36:11 UTC  Completed   GET /kafkacruisecontrol/state?json=true
   aebae987-985d-4871-8cfb-6134ecd504ab  0:0:0:0:0:0:0:1       2020-06-01_16:10:04 UTC

4. 特定のクラスターリバランスタスクの状態を表示するには、user-task-ids パラメーターおよびタスク ID を指定します。

   curl 'cruise-control-server:9090/kafkacruisecontrol/user_tasks?user_task_ids=c459316f-9eb5-482f-9d2d-97b5a4cd294d'

1. 削除する各ブローカーに、ログ (log.dirs) にライブパーティションがないことを確認します。

   ls -l <LogDir> | grep -E '^d' | grep -vE '[a-zA-Z0-9.-]+\.[a-z0-9]+-delete$'

   ログディレクトリーが正規表現 (\.[a-z0-9]-delete$) と一致しない場合は、アクティブなパーティションが存在します。アクティブなパーティションが存在する場合は、リバランスが完了していること、または最適化プロポーザルの設定を確認します。プロポーザルを再度実行できます。次のステップに進む前に、アクティブなパーティションが存在しないことを確認します。

2. ブローカーを停止します。

   su - kafka
   /opt/kafka/bin/kafka-server-stop.sh

3. ブローカーが停止したことを確認します。

   jcmd | grep kafka

手順

1. クラスター内の他のブローカーと同じ設定を使用して新しいブローカーの設定ファイルを作成します。ただし、broker.id には他のブローカで使用されていない番号を指定してください。

2. 前のステップで作成した設定ファイルをkafka-server-start.sh スクリプトの引数に渡して、新しい Kafka ブローカーを起動します。

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

3. Kafka ブローカーが稼働していることを確認します。

   jcmd | grep Kafka

4. 新しいブローカーごとに上記の手順を繰り返します。

5. まだ行っていない場合は、kafka-reassign-partitions.sh ツールを使用して reassignment.json という名前の再割り当て JSON ファイルを生成します。

   再割り当て JSON ファイルを生成するコマンドの例

   /opt/kafka/bin/kafka-reassign-partitions.sh \
     --bootstrap-server localhost:9092 \
     --topics-to-move-json-file topics.json \ 1
     --broker-list 0,1,2,3,4 \ 2
     --generate

   1

   トピックを指定する JSON ファイル。

   2

   操作に含める Kafka クラスター内のブローカー ID。これは、ブローカー 4 が追加されたことを前提としています。

   現在のレプリカ割り当てと提案されたレプリカ割り当てを示す再割り当て JSON ファイルの例

   Current partition replica assignment
   {"version":1,"partitions":[{"topic":"my-topic","partition":0,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"my-topic","partition":1,"replicas":[1,2,3],"log_dirs":["any","any","any"]},{"topic":"my-topic","partition":2,"replicas":[2,3,0],"log_dirs":["any","any","any"]}]}
   
   Proposed partition reassignment configuration
   {"version":1,"partitions":[{"topic":"my-topic","partition":0,"replicas":[0,1,2,3],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":1,"replicas":[1,2,3,4],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":2,"replicas":[2,3,4,0],"log_dirs":["any","any","any","any"]}]}

   後で変更を元に戻す必要がある場合に備えて、このファイルのコピーをローカルに保存します。

6. --execute オプションを使用してパーティションの再割り当てを実行します。

   /opt/kafka/bin/kafka-reassign-partitions.sh \
     --bootstrap-server localhost:9092 \
     --reassignment-json-file reassignment.json \
     --execute

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

   /opt/kafka/bin/kafka-reassign-partitions.sh \
     --bootstrap-server localhost:9092 \
     --reassignment-json-file reassignment.json \
     --throttle 5000000 \
     --execute

7. --verify オプションを使用して、再割り当てが完了したことを確認します。

   /opt/kafka/bin/kafka-reassign-partitions.sh \
     --bootstrap-server localhost:9092 \
     --reassignment-json-file reassignment.json \
     --verify

   --verify コマンドによって、移動した各パーティションが正常に完了したことが報告されると、再割り当ては終了します。この最終的な --verify によって、結果的に再割り当てスロットルも削除されます。

手順

1. まだ行っていない場合は、kafka-reassign-partitions.sh ツールを使用して reassignment.json という名前の再割り当て JSON ファイルを生成します。

   再割り当て JSON ファイルを生成するコマンドの例

   /opt/kafka/bin/kafka-reassign-partitions.sh \
     --bootstrap-server localhost:9092 \
     --topics-to-move-json-file topics.json \ 1
     --broker-list 0,1,2,3 \ 2
     --generate

   1

   トピックを指定する JSON ファイル。

   2

   操作に含める Kafka クラスター内のブローカー ID。これは、ブローカー 4 が削除されたことを前提としています。

   現在のレプリカ割り当てと提案されたレプリカ割り当てを示す再割り当て JSON ファイルの例

   Current partition replica assignment
   {"version":1,"partitions":[{"topic":"my-topic","partition":0,"replicas":[3,4,2,0],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":1,"replicas":[0,2,3,1],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":2,"replicas":[1,3,0,4],"log_dirs":["any","any","any","any"]}]}
   
   Proposed partition reassignment configuration
   {"version":1,"partitions":[{"topic":"my-topic","partition":0,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"my-topic","partition":1,"replicas":[1,2,3],"log_dirs":["any","any","any"]},{"topic":"my-topic","partition":2,"replicas":[2,3,0],"log_dirs":["any","any","any"]}]}

   後で変更を元に戻す必要がある場合に備えて、このファイルのコピーをローカルに保存します。

2. --execute オプションを使用してパーティションの再割り当てを実行します。

   /opt/kafka/bin/kafka-reassign-partitions.sh \
     --bootstrap-server localhost:9092 \
     --reassignment-json-file reassignment.json \
     --execute

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

   /opt/kafka/bin/kafka-reassign-partitions.sh \
     --bootstrap-server localhost:9092 \
     --reassignment-json-file reassignment.json \
     --throttle 5000000 \
     --execute

3. --verify オプションを使用して、再割り当てが完了したことを確認します。

   /opt/kafka/bin/kafka-reassign-partitions.sh \
     --bootstrap-server localhost:9092 \
     --reassignment-json-file reassignment.json \
     --verify

   --verify コマンドによって、移動した各パーティションが正常に完了したことが報告されると、再割り当ては終了します。この最終的な --verify によって、結果的に再割り当てスロットルも削除されます。

4. 削除する各ブローカーに、ログ (log.dirs) にライブパーティションがないことを確認します。

   ls -l <LogDir> | grep -E '^d' | grep -vE '[a-zA-Z0-9.-]+\.[a-z0-9]+-delete$'

   ログディレクトリーが正規表現 (\.[a-z0-9]-delete$) と一致しない場合は、アクティブなパーティションが存在します。アクティブなパーティションがある場合は、再割り当てが完了したどうか、あるいは、再割り当て JSON ファイルの設定を確認します。再割り当てはもう一度実行できます。次のステップに進む前に、アクティブなパーティションが存在しないことを確認します。

5. ブローカーを停止します。

   su - kafka
   /opt/kafka/bin/kafka-server-stop.sh

6. Kafka ブローカーが停止していることを確認します。

   jcmd | grep kafka