1. Home
2. Platform
3. 管理
4. Kafka の運用
5. Self-Balancing Clusters

Self-Balancing チュートリアル

この例では、ローカルのマルチブローカークラスターを、レプリケーション係数の適切な構成でセットアップする方法や、ブローカーを削除してバランス調整をモニタリングすることで Self-Balancing をテストする方法について説明します。

ブローカーの削除に続いて、クラスターにブローカーを追加し、空のブローカーにデータが再配分される状況をモニタリングします。

多くのタスクはコマンドラインで実行しますが、Confluent Control Center をバランス調整の進捗を詳しく確認するためのビューとして利用して、コントローラーが実行されている場所を確認することができます。

ちなみに

この同じワークフローを使用して、お使いのクラウドプロバイダーでクラスターをセットアップできます。適切にクラウドノードを表すように host:port の名前を設定するだけです。

前提条件

この例を進める前に、ローカルマシンに次のコンポーネントがインストールされていることを検証します。

• Confluent Platform 6.0.0 以降
• Confluent CLI
• Confluent Platform 実行用の Java 1.8 または 1.11

Kafka ブローカーの構成

この例では 5 つのブローカーがあるクラスターのデモを示します。次の手順では、ブローカーのプロパティファイルの作成を 2 段階で進めます。まず、すべてのブローカーに適用するデフォルト server.properties ファイルの基本部分を構成し、その後、元のファイルをベースに 4 つの追加プロパティファイルを作成して、マルチブローカークラスターをセットアップします。セットアップが完了したら、ブローカーごとに 1 つ、合計 5 つのサーバープロパティファイルが作成されることになります。

ちなみに

ブローカーの数に関係なく、クラスターにこの同じモデルを使用できます。3 つを超えるブローカーを使用すると、ブローカーを削除するシナリオ向けの構成が簡単になります。多くの Confluent Platform 内部トピックで、レプリケーション係数が 3 にデフォルト設定されているからです。

$CONFLUENT_HOME/etc/kafka/server.properties で、次の変更を行ってファイルを保存します。

Control Center 用の Metrics Reporter の有効化

Confluent Metrics Reporter が有効になると、Control Center の Brokers Overview ページに、すべてのブローカーのメトリクス、個別のブローカーをドリルダウンして統計情報の詳細を確認できるクリックボタンの付いたリストと、ブローカーを削除するオプションが表示されます。この設定が有効ではない場合、Control Center にブローカーのメトリクスや管理オプションは表示されません。

次の 2 行のコメントを解除すると、ブローカーがメトリクスを Control Center に送信できます。

metric.reporters=io.confluent.metrics.reporter.ConfluentMetricsReporter
confluent.metrics.reporter.bootstrap.servers=localhost:9092

Copy

ちなみに

この構成は、クラスター内のすべてのブローカーに適用できます。confluent.metrics.reporter.bootstrap.servers のポート番号を変更して、各ブローカーのリスナーポートに一致させる必要はありません（一致させた構成でも問題ありません）。

Self-Balancing のレプリケーション係数の構成

ブローカーを削除する場合に Self-Balancing が機能するように、レプリケーション係数をクラスターのブローカー数より少なく、ただしレプリケーションを実行するのに十分な数にする必要があります。たとえば、レプリケーション係数 1 のトピックは設定できません。削除されるブローカーにトピックがある場合、バランス調整に使用されるレプリカがないからです。ブローカーの削除を試みると次のエラーが表示されます。

my-laptop$ ./bin/kafka-remove-brokers --bootstrap-server localhost:9092,localhost:9093,localhost:9094 --broker-id 2 --delete 2>&1 | grep -v SLF4J
Initiating remove broker call...
Error while issuing request
Error while executing broker removal: org.apache.kafka.common.errors.InvalidBrokerRemovalException: The given broker removal operation was invalid for some reason, e.g invalid broker ids or partitions that would become unavailable as a result the removal. See the broker logs for more details.

Copy

次の手順では、レプリケーション係数とレプリカを 2 に再設定し、変更が有効になるように、必要に応じてプロパティのコメントを解除する方法を示します。これを記述する時点で、以下に示す一連のプロパティにおいては広範囲にわたってアップデートが必要になります。ただし、サーバーファイルにレプリケーション係数またはレプリカ数が 1 あるいは 2 を超える他のプロパティがある場合は、このチュートリアルのために変更します。これらの値を変更している "理由" をよく考えてください。たとえば、これは #confluent.balancer.max.replicas=10000 のような設定には当てはまりません。これは単に、ブローカーごとに許可されるレプリカの最大数を指定しているためです。

1. $CONFLUENT_HOME/etc/kafka/server.properties ファイルで replication.factor をすべて検索し、その値を、1 より大きくブローカーの数より小さい数に設定します。このクラスターの場合は、すべての replication.factor を「2」に設定します。 server.properties を検索すると、以下のプロパティが見つかるので、コメントアウトされている場合は、コメントを解除します。

   offsets.topic.replication.factor=2
   transaction.state.log.replication.factor=2
   confluent.license.topic.replication.factor=2
   confluent.metadata.topic.replication.factor=2
   confluent.balancer.topic.replication.factor=2
   

   Copy

2. Connect を実行する場合は、このプロパティファイル内のレプリケーション係数も変更します。$CONFLUENT_HOME/etc/kafka/connect-distributed.properties ファイルで replication.factor をすべて検索し、その値を、1 より大きくブローカーの数より小さい数に設定します。このクラスターの場合は、すべての replication.factor を「2」に設定します。connect-distributed.properties を検索すると、以下のプロパティが見つかります。コメントアウトされている場合は、コメントを解除します。

   offset.storage.replication.factor=2
   config.storage.replication.factor=2
   status.storage.replication.factor=2
   

   Copy

ちなみに

• ユーザーが独自のトピックを作成する場合は、ブローカー数に応じて、必要となるレプリケーション係数を設定するようにします。これについては次のセクションで説明します。
• レプリカ数とレプリケーション係数を 2 に制限すると、クラスターをわずか 2 つのブローカーに縮小して再度 5 つまで拡張するオプションを利用できます。

Self-Balancing が有効であることの検証

server.properties の "Confluent Data Balancer Settings" で、confluent.balancer.enable=true、つまり Self-Balancing がオンであることを検証します。このプロパティの値が true に指定され、その行がコメントアウトされていないことが必要です。Confluent Platform に付属するこのサンプルファイルで Self-Balancing が有効に設定されています。したがってこの部分を編集する必要はありません。

ちなみに

confluent.balancer.enable が server.properties で明確に指定されていない場合、あるいはこの行がコメントアウトされている場合、デフォルトは false、つまりオフです。

ファイルの保存

アップデートされた server.properties を保存します。

この server.properties ファイルは、5 つのブローカーの最初のものである "broker 0" 用です。このファイルを、次のセクションで説明する、broker 1、2、3 に対応するファイルのベースとして使用できます。必要に応じてレプリケーション係数をアップデート済みであるため、他の 2 ファイルでは最小の変更を行います。

5 つのブローカーがあるクラスターの基本構成の作成

まず、前のステップで Metrics Reporter、レプリケーション係数、Self-Balancing をアップデートした server.properties ファイルをコピーし、構成を次のように変更して、新しいファイルを他の 4 つのブローカーを表す名前に変更します。

ファイル	構成
server.properties	前の手順でレプリケーション係数が正しく設定されているので、このファイルで他に変更は必要ありません。 以下の基本プロパティではデフォルト値を使用します（値がコメントアウトされている場合は、そのままにしておきます）。 broker.id=0 #listeners=PLAINTEXT:// :9092 log.dirs=/tmp/kafka-logs 次のリスナー構成を追加して、このブローカーの REST エンドポイントを指定します。 confluent.http.server.listeners=http://localhost:8090 Copy
server-1.properties	以下の基本プロパティの値が一意になるように変更します（listeners の行のコメントアウトを解除します）。 broker.id=1 listeners=PLAINTEXT:// :9093 log.dirs=/tmp/kafka-logs-1 リスナー構成で、このブローカーに一意の REST エンドポイントを指定します。 confluent.http.server.listeners=http://localhost:8091 Copy
server-2.properties	以下の基本プロパティの値が一意になるように変更します（listeners の行のコメントアウトを解除します）。 broker.id=2 listeners=PLAINTEXT:// :9094 log.dirs=/tmp/kafka-logs-2 リスナー構成で、このブローカーに一意の REST エンドポイントを指定します。 confluent.http.server.listeners=http://localhost:8092 Copy
server-3.properties	以下の基本プロパティの値が一意になるように変更します（listeners の行のコメントアウトを解除します）。 broker.id=3 listeners=PLAINTEXT:// :9095 log.dirs=/tmp/kafka-logs-3 リスナー構成で、このブローカーに一意の REST エンドポイントを指定します。 confluent.http.server.listeners=http://localhost:8093 Copy
server-4.properties	以下の基本プロパティの値が一意になるように変更します（listeners の行のコメントアウトを解除します）。 broker.id=4 listeners=PLAINTEXT:// :9096 log.dirs=/tmp/kafka-logs-4 リスナー構成で、このブローカーに一意の REST エンドポイントを指定します。 confluent.http.server.listeners=http://localhost:8094 Copy

ちなみに

• Confluent Platform は Self-Balancing が有効な状態で出荷されます。この設定（confluent.balancer.enable=true）はそのままにします。
• このチュートリアルでは、セキュリティプロトコルを有効にした状態で Self-Balancing Clusters を実行する方法を説明していません。セキュアモードで Self-Balancing を実行する方法については、「セキュリティに関する考慮事項」を参照してください。

この手順を完了すると、$CONFLUENT_HOME/etc/kafka/ 内に 5 つのサーバープロパティファイルが作成されます。

• server.properties （broker 0 に対応）
• server-1.properties （broker 1 に対応）
• server-2.properties （broker 2 に対応）
• server-3.properties （broker 3 に対応）
• server-4.properties （broker 4 に対応）

REST エンドポイントとアドバタイズされたリスナーを使用する Control Center の構成

コマンドライン出力に加えて、Control Center では構成の開始を検証し、バランス調整の進捗状況をモニタリングするときに活用できます。

クラスター内のすべてのブローカーの REST エンドポイントに加え、実行する他のコンポーネント用にアドバタイズされたリスナーを Control Center に指定する必要があります。その構成を行っていない場合、ブローカーおよびコンポーネントは Control Center に表示されません。

$CONFLUENT_HOME/etc/confluent-control-center/control-center.properties で次の変更を行って、ファイルを保存します。

1. ブローカーの REST エンドポイントを構成します。

   該当する Control Center プロパティファイルで confluent.controlcenter.streams.cprest.url を使用して、controlcenter.cluster の REST エンドポイントを定義します。

   $CONFLUENT_HOME/etc/confluent-control-center/control-center.properties ファイルで、Kafka の REST エンドポイントの URL のデフォルト値のコメントを解除し、マルチブローカー構成に合わせて次のように変更します。

   # Kafka REST endpoint URL
   confluent.controlcenter.streams.cprest.url=http://localhost:8090,http://localhost:8091,http://localhost:8092,http://localhost:8093,http://localhost:8094
   

   Copy

2. Kafka Connect、ksqlDB、および Schema Registry の構成のコメントを解除して、コンポーネントクラスターのデフォルトのアドバタイズされた URL を Control Center に提供します。

   # A comma separated list of Connect host names
   confluent.controlcenter.connect.cluster=http://localhost:8083
   
   # KSQL cluster URL
   confluent.controlcenter.ksql.ksqlDB.url=http://localhost:8088
   
   # Schema Registry cluster URL
   confluent.controlcenter.schema.registry.url=http://localhost:8081
   

   Copy

ちなみに

• Self-Balancing には組み込みの REST Proxy は必要ありませんが、Control Center で Self-Balancing を管理するには必要です。Self-Balancing の構成オプションの「Control Center の必須構成」と、「Control Center の構成リファレンス」の「confluent.controlcenter.streams.cprest.url」も参照してください。
• role-based access control (RBAC) またはその他のセキュリティ プロトコルを使用して Self-Balancing を実行している場合は、Kafka ブローカーで、このチュートリアルに示されている内容以外にも、追加の構成が必要になります。詳細については、「セキュリティに関する考慮事項」を参照してください。

Confluent Platform の起動、トピックの作成とテストデータの生成

次の手順で、個別のコマンドウィンドウでサーバーを起動し、トピックを作成して、そのトピックにデータを生成します。

1. ZooKeeper を個別のコマンドウィンドウで起動します。

   ./bin/zookeeper-server-start etc/kafka/zookeeper.properties
   

   Copy

2. 各ブローカーを個別のコマンドウィンドウで起動します。

   ./bin/kafka-server-start etc/kafka/server.properties
   

   Copy

   ./bin/kafka-server-start etc/kafka/server-1.properties
   

   Copy

   ./bin/kafka-server-start etc/kafka/server-2.properties
   

   Copy

   ./bin/kafka-server-start etc/kafka/server-3.properties
   

   Copy

   ./bin/kafka-server-start etc/kafka/server-4.properties
   

   Copy

3. 他の Confluent Platform コンポーネントを個別のウィンドウで起動します。

   • Kafka REST

     ./bin/kafka-rest-start etc/kafka-rest/kafka-rest.properties
     

     Copy

   • Control Center

     ./bin/control-center-start etc/confluent-control-center/control-center.properties
     

     Copy

4. （省略可）各コンポーネントを構成し起動します。

   次のコンポーネントを実行する場合は、まずプロパティファイルを編集します。 replication.factor の値をすべて検索し、（5 つのブローカーのクラスターと連動するために） 2 または 3に置き換えます。replication.factor の値が 2 未満または 4 を超えている場合、次のセクションで示すように、Self-Balancing によるブローカーの適切な削除を妨げるレプリケーション係数が設定したシステムトピックとなります。

   • （省略可） Kafka Connect

     ./bin/connect-distributed etc/kafka/connect-distributed.properties
     

     Copy

   • （省略可） ksqlDB

     ./bin/ksql-server-start etc/ksqldb/ksql-server.properties
     

     Copy

   • （省略可） Schema Registry

     ./bin/schema-registry-start etc/schema-registry/schema-registry.properties
     

     Copy

5. 3 パーティションとレプリケーション係数 2 で 1 つまたは複数のトピックを作成します。

   kafka-topics --create --topic my-sbc-test --partitions 3 --replication-factor 2 --bootstrap-server localhost:9092
   

   Copy

   トピックが正常に作成されたことを確認する必要があります。次のとおり、既存のトピックのリストも取得できます。

   kafka-topics --list --bootstrap-server localhost:9092
   

   Copy

   --describe オプションで特定トピックの詳細情報が取得できます。

   kafka-topics --describe --topic my-sbc-test --bootstrap-server localhost:9092
   

   Copy

   または、--describe オプションですべてのトピックの詳細情報が得られます。

   kafka-topics --describe --bootstrap-server localhost:9092
   

   Copy

   ちなみに

   • Self-Balancing が機能するためには、クラスターのブローカー数に適切にマッピングされたレプリケーション係数が、トピックに設定される必要があります。レプリケーション係数には 1 より大きく、ブローカーの総数より少ない数が必要です。5 ブローカーのクラスターではレプリケーション係数を 2 ～ 4 にする必要があります。
   • 複数パーティションのトピックを作成し、5 つすべてのブローカーにデータを配分します。

6. データをトピックに生成します。

   個別のコマンドウィンドウで、kafka-producer-perf-test コマンドを使用し、データをトピック my-sbc-test に生成します。

   kafka-producer-perf-test \
      --producer-props bootstrap.servers=localhost:9092 \
      --topic my-sbc-test \
      --record-size 1000 \
      --throughput 1000 \
      --num-records 3600000
   

   Copy

   追加のトピックを作成した場合、このコマンドでデータをそれらのトピックにも送信できます。

この時点で、Self-Balancing をモニタリングし、テストする準備ができました。ブローカーを削除し、再び追加して、処理中のバランス調整をモニタリングします。次のセクションでは、これを実行する 2 種類の方法（コマンドラインまたは Control Center）を示します。両方の方法を試す場合は、これらの手順を順に進めるか、あるいは Confluent Control Center とコマンドラインの両方を区別なく使用できます。ただし、Self-Balancing で主に使用する方法としてコマンドラインまたは Control Center のいずれかを使用することを想定しています。

コマンドラインの使用

次のセクションでは、コマンド kafka-remove-brokers でブローカーを削除し、バランス調整の進捗状況をモニタリングする方法について説明します。ブローカーを再起動（実質的には、ブローカーを "追加"）すると、Self-Balancing によりすべてのノードにデータが再配分されます。

ブローカーのステータスとトピックデータの検証

コマンドラインで、トピック、トピックのデータ配分、ブローカー数など、デプロイの現在のステータスを検証します。

1. kafka-broker-api-versions および grep を id に使用して、オンラインのブローカーを表示します。

   kafka-broker-api-versions --bootstrap-server localhost:9092 | grep 'id: '
   

   Copy

   出力は以下のようになります。

   localhost:9095 (id: 3 rack: null) -> (
   localhost:9093 (id: 1 rack: null) -> (
   localhost:9096 (id: 4 rack: null) -> (
   localhost:9094 (id: 2 rack: null) -> (
   localhost:9092 (id: 0 rack: null) -> (
   

   Copy

2. kafka-topics --describe を使用して、作成したテストトピックに関する情報を表示します。

   kafka-topics --bootstrap-server localhost:9092 --topic my-sbc-test --describe
   

   Copy

   出力は以下のようになります。

   Topic: my-sbc-test        PartitionCount: 3       ReplicationFactor: 2    Configs: segment.bytes=1073741824
         Topic: my-sbc-test  Partition: 0    Leader: 0       Replicas: 0,4   Isr: 0,4        Offline:
         Topic: my-sbc-test  Partition: 1    Leader: 4       Replicas: 4,1   Isr: 4,1        Offline:
         Topic: my-sbc-test  Partition: 2    Leader: 1       Replicas: 1,2   Isr: 1,2        Offline:
   

   Copy

ブローカーの削除

Self-Balancing が有効であり、Confluent Platform が動作している状態で、 kafka-remove-brokers によってブローカーを削除し、バランス調整をモニタリングします。

以下の手順を開始する前に、Confluent Platform と Self-Balancing を最低 20 分実行して、Self-Balancing を初期化する時間を設けてください（詳細については、トラブルシューティングの「Self-Balancing の初期化中にブローカーの削除の試行に失敗する」を参照してください）。

また、この例では、ブローカー ID が 0 のコントローラーを削除しないでください。

ちなみに

• 実際には、リードブローカーを削除できます。この場合、クラスターバランシングに短い遅延が発生する可能性があるため、この例ではリードブローカーを削除しないことをお勧めしています。詳細については、「リードブローカー（コントローラー）が削除される、または見つからない場合の動作」を参照してください。
• 削除するブローカーに格納されているレプリカがいずれかのトピックにとって唯一のレプリカである場合、このブローカーの削除は失敗します。詳細については、「制限」を参照してください。

1. ブローカーを削除します。

   たとえば、次のコマンドでブローカー 1 を削除し、そのデータをクラスターの残りのブローカーに移動します。

   ./bin/kafka-remove-brokers --bootstrap-server \
   localhost:9092,localhost:9093,localhost:9094,localhost:9095,localhost:9096 \
   --broker-id 1 --delete 1>&2 | grep -v SLF4J
   

   Copy

   Self-Balancing はコマンドの応答を確認し、次のようなフィードバックを返します。

   Initiating remove broker call...
   Started remove broker task for broker 1.
   You can check its status by calling this command again with the `--describe` option.
   

   Copy

2. コマンドラインからバランス調整の進捗状況をモニタリングします。

   シャットダウンおよびバランス調整操作を追跡できます。 --describe オプションを前のコマンドで、--delete の代わりに指定します。

   ./bin/kafka-remove-brokers --bootstrap-server \
   localhost:9092,localhost:9093,localhost:9094,localhost:9095,localhost:9096 \
   --broker-id 1 --describe 1>&2 | grep -v SLF4J
   

   Copy

   進行中のバランス調整では、次のようなフィードバックが得られます。

   Broker 1 removal status:
     Partition Reassignment: IN_PROGRESS
     Broker Shutdown: COMPLETE
   

   Copy

   ブローカーの削除が完了したら、--describe コマンドで次の結果が表示されます。

   Broker 1 removal status:
     Partition Reassignment: COMPLETE
     Broker Shutdown: COMPLETE
   

   Copy

   注釈

   次の エラー が表示された場合は、Self-Balancing はまだ 初期化中 の可能性があります。初期化には最大で 30 分かかります。この場合は、数分後にブローカーの削除を再試行すると成功するはずです。

   Broker 1 removal status:
     Partition Reassignment: FAILED
     Broker Shutdown: CANCELED
   

   Copy

3. kafka-broker-api-versions を再実行して、オンラインのブローカーを表示します。

   kafka-broker-api-versions --bootstrap-server localhost:9092 | grep 'id: '
   

   Copy

   出力は以下のようになります。

   localhost:9092 (id: 0 rack: null) -> (
   localhost:9096 (id: 4 rack: null) -> (
   localhost:9094 (id: 2 rack: null) -> (
   localhost:9095 (id: 3 rack: null) -> (
   

   Copy

   ブローカー 1 がオフラインであることがわかります。

   ちなみに

   kakfa-topics --describe をすべてのトピックまたは特定のトピックで、次のコマンドとともに再実行することもできます。バランス調整に関連する変更が表示されることもありますが、トピックとトピックデータが変わらず利用できることを検証します。

   kafka-topics --describe --bootstrap-server localhost:9092
   kafka-topics --bootstrap-server localhost:9092 --topic my-sbc-test --describe
   

   Copy

ブローカーの追加（再起動）

ブローカーの削除処理（前のセクション）が完了した後に、ブローカーを再起動できます。（これは、説明を簡単にするために、削除したものと同じブローカーを使用して "ブローカーを追加する" 例です）。

1. ブローカー（たとえばブローカー 1）を再起動し、バランス調整を監視します。

   これを実行する一番簡単な方法は、ブローカー 1 を開始したコマンドウィンドウに戻ることです。 ブローカーが停止しているのがわかります。キーボードの上方向キーを押してから、Enter キーを押して、開始したときと同じコマンドをそのまま再実行します。

   ./bin/kafka-server-start etc/kafka/server-1.properties
   

   Copy

   Self-Balancing はコマンドの応答を確認し、ブローカーの再起動時に、次のようなフィードバックを返します。

   [2020-06-26 17:45:44,986] INFO BROKER Aggregator rolled out 1 new windows, reset 1 windows, current window range [1593213000000, 1593219000000], abandon 0 samples. (com.linkedin.cruisecontrol.monitor.sampling.aggregator.MetricSampleAggregator)
   [2020-06-26 17:46:06,314] INFO DataBalancer: Scheduling DataBalanceEngine broker addition: [1] (io.confluent.databalancer.ConfluentDataBalanceEngine)
   [2020-06-26 17:46:06,314] INFO DataBalancer: Starting addBrokers call (io.confluent.databalancer.ConfluentDataBalanceEngine)
   

   Copy

2. ブローカーが立ち上がったら、kafka-broker-api-versions を再実行して、オンラインのすべてのブローカーを表示します。

   kafka-broker-api-versions --bootstrap-server localhost:9092 | grep 'id: '
   
   Your output should resemble:
   

   Copy

   localhost:9095 (id: 3 rack: null) -> (
   localhost:9093 (id: 1 rack: null) -> (
   localhost:9096 (id: 4 rack: null) -> (
   localhost:9094 (id: 2 rack: null) -> (
   localhost:9092 (id: 0 rack: null) -> (
   

   Copy

   ブローカー 1 がオンラインに戻っていることがわかります。

Control Center の使用

次のセクションでは、Control Center でブローカーを削除し、バランス調整の進捗状況をモニタリングする方法について説明します。ブローカーを再起動すると、Self-Balancing によりすべてのノードにデータが再配分されます。

ブローカーのステータスとトピックデータの検証

Control Center で、デプロイの現在のステータス（Self-Balancing の設定、リードブローカー（コントローラー）、トピックデータ、ブローカー数）を検証します。ローカルデプロイの場合、Control Center を使用するには、ウェブブラウザーで http://localhost:9021/ にアクセスします。

• コントローラーが動作している場所を検証するには、Control Center に移動し、クラスターを選択して、Brokers をクリックします。この例では、コントローラーはブローカー 0 で動作しています。

  

  ちなみに

  ブローカーのメトリクスやブローカーのリストが Brokers overview ページに表示されない場合、Metrics Reporter が有効であることを確認します（「Control Center 用の Metrics Reporter の有効化」を参照）。有効ではない場合はブローカーを停止し、各ファイルを編集して再起動します。

• Self-Balancing ブローカータスクのステータスを表示するには、Brokers をクリック後、Self-Balancing カードをクリックします。

  

  Self-Balancing により、クラスターの各ブローカーのタスクステータスが表示されます（次のセクションで、ブローカーを削除したり再度追加したりする操作の際に、これが表示されます）。

  

• オンラインのすべてのブローカーを確認するには、Brokers ページの下までスクロールすればブローカーリストが表示されます。または、Cluster settings の Broker defaults タブをクリックします。

  

• Self-Balancing が有効であることを検証するには、Cluster settings の Self-balancing タブをクリックします。

  

• トピックに生成されたメッセージを表示するには、Topics、my-sbc-test、Messages タブの順にクリックします。

  

ブローカーの削除

Self-Balancing が有効で Confluent Platform が動作している状態で、ブローカーを削除し、バランス調整をモニタリングします。この例では、ブローカー ID が 0 のコントローラーを削除しないように注意します。

ちなみに

• 実際には、リードブローカーを削除できます。この場合、クラスターバランシングに短い遅延が発生する可能性があるため、この例ではリードブローカーを削除しないことをお勧めしています。詳細については、「リードブローカー（コントローラー）が削除される、または見つからない場合の動作」を参照してください。
• 削除するブローカーに格納されているレプリカがいずれかのトピックにとって唯一のレプリカである場合、このブローカーの削除は失敗します。詳細については、「制限」を参照してください。

1. Brokers overview ページの Control Center オプションを使用してブローカーを削除します。

   • Brokers を選択し、Overview ページの下までスクロールして現在オンラインのブローカーのリストを表示します。

   • 削除するブローカーをクリックします（ブローカーをクリックすると、ブローカーの詳細が表示され、また削除オプションも表示されます）。

     

   • ブローカーの詳細ページの下にある Remove broker をクリックし、入力フィールドに「REMOVE」と入力してこのアクションを行うことを確認します。

     

   • Continue をクリックして、ブローカーの削除タスクを開始します。

   注釈

   メトリクス不足が原因でブローカーの削除に失敗したという エラーメッセージ が表示された場合は、Self-Balancing がまだ 初期化中 です。初期化には最大で 30 分かかります。この場合は、数分後にブローカーの削除を再試行すると成功するはずです。

2. Control Center を使用してバランス調整をモニタリングします。

   • Control Center で、Brokers の Self-balancing をクリックして、進捗状況を追跡します。

     Self-Balancing により、クラスターの各ブローカーの詳細ステータスが表示されます。この例では、ブローカー 1 は Remove broker tasks で in-progress ステータスが表示されています。

     

   • 削除操作の実行中、削除対象のブローカーには、Brokers overview ページの下のブローカーリストで赤の「失敗」インジケーターが表示されています。

     

   • バランス調整が完了すると、Brokers overview ページと Cluster settings の Broker defaults のどちらにも、4 つのブローカー（0、2、3、4）のみがリストに表示されます。

     
     

ブローカーの追加（再起動）

1. ブローカー（たとえばブローカー 1）を再起動し、バランス調整を監視します。（これは、説明を簡単にするために、削除したものと同じブローカーを使用して "ブローカーを追加する" 例です）。

   ブローカーを再起動するには、コマンドラインを使用する必要があります。ブローカー 1 を起動したコマンドウィンドウに戻ります。ブローカーが停止していることがわかります。キーボードの上方向キーを押してから、Enter キーを押して、開始したときと同じコマンドをそのまま再実行します。

   ./bin/kafka-server-start etc/kafka/server-1.properties
   

   Copy

   Self-Balancing はコマンドの応答を確認し、次のようなフィードバックを返します。

   [2020-06-26 17:45:44,986] INFO BROKER Aggregator rolled out 1 new windows, reset 1 windows, current window range [1593213000000, 1593219000000], abandon 0 samples. (com.linkedin.cruisecontrol.monitor.sampling.aggregator.MetricSampleAggregator)
   [2020-06-26 17:46:06,314] INFO DataBalancer: Scheduling DataBalanceEngine broker addition: [1] (io.confluent.databalancer.ConfluentDataBalanceEngine)
   [2020-06-26 17:46:06,314] INFO DataBalancer: Starting addBrokers call (io.confluent.databalancer.ConfluentDataBalanceEngine)
   

   Copy

2. Control Center を使用して、Self-Balancing の進捗をモニタリングします（Brokers の Self-balancing）。

   たとえば、ブローカーが再度追加されると、Add broker tasks に実行中のインジケーターが表示されます。

   

   バランス調整が完了したら、Brokers ページの下のブローカーリストに移動して、ブローカー 1 がオンラインに戻ったこと、合計で 5 ブローカーになったことを確認します。

シャットダウンとクリーンアップのタスク

以下のシャットダウンとクリーンアップのタスクを実行します。

1. kafka-producer-perf-test を、個別のコマンドウィンドウで Ctrl キーを押しながら C キーを押して停止します。
2. 他のすべてのコンポーネントを、可能な場合は起動したときと逆の順序で、それぞれのコマンドウィンドウで Ctrl キーを押しながら C キーを押して停止します。たとえば、まず Control Center を停止し、次に他のコンポーネントを停止してから Kafka のブローカーを停止します。最後に ZooKeeper を停止します。

おすすめの関連情報

• 構成オプションとコマンド
• クイックスタートデモ（Docker）
• Self-Balancing の活用 （「Confluent REST Proxy API リファレンス」で詳しく説明）