Kafkaをプログラムで管理する方法

紹介

Apache KafkaはJavaで書かれたオープンソースの分散イベントおよびストリーム処理プラットフォームです。これは、要求の厳しいリアルタイムデータフィードを処理するために構築されており、基本的にスケーラブルで、高いスループットと可用性を備えています。このプロジェクトは、Kafkaクラスタとのメッセージの生成と消費、およびトピックやパーティションの管理などの管理タスクを行うためのシェルスクリプトを提供しています。これらは探求や実験に役立つ一方で、実際のアプリケーションはプログラムでKafkaにアクセスします。そのために、Kafkaは広く使用されているプログラミング言語や環境向けの多くのクライアントライブラリを提供しています。

このチュートリアルでは、KafkaAdminClient APIを使用してKafkaクラスタのリソースを管理する方法を学びます。また、プログラム的にクラスタに関する情報を取得し、トピックの作成、一覧表示、削除を行う方法も学びます。さらに、Javaに依存せずにKafkaクラスタにアクセスできるkcat CLIユーティリティについても学ぶことができます。

最後に、Kafkaクルーズコントロールを設定する方法を学びます。これは、Kafkaクラスターの内部プロセスを自動的に最適化するデーモンであり、より高い効率と信頼性を実現します。

前提条件

このチュートリアルを完了するためには、次のものが必要です:

• 少なくとも4GBのRAMと2つのCPUを持つマシン。Ubuntuサーバーの場合、セットアップ手順については初期サーバー設定に従ってください。

• ドロップレットまたはローカルマシンにJava Development Kit (JDK) 11をインストールします。UbuntuにJavaをインストールする手順については、「Aptを使ってUbuntuにJavaをインストールする方法」チュートリアルを参照してください。

• 3つのブローカーを持つApache Kafkaクラスター。セットアップ手順については、KRaftを使用したマルチノードKafkaクラスターのセットアップ方法に関するチュートリアルに従うことができます。

• CLIチュートリアルに従ってデータをソースするために設定されたKafkaプロデューサーを持つJavaプロジェクト。

• Javaプロジェクトの標準ディレクトリレイアウトに関する理解。詳細については、公式Mavenドキュメントの「標準ディレクトリレイアウトの紹介」トピックを参照してください。

• あなたのDropletまたはローカルマシンにPython 3がインストールされており、新しい仮想環境が設定されています。Ubuntuサーバーの場合、UbuntuサーバーにPython 3をインストールし、プログラミング環境をセットアップする方法に関するチュートリアルを訪れてください。ステップ1とステップ2のみを完了する必要があります。このチュートリアルでは、仮想環境は~/venvと見なされます。

ステップ1 - Kafka AdminClientの利用

前提条件の一部として、Kafkaにプログラムでアクセスし、java_demoトピックにメッセージを生成するために必要な依存関係を含むJavaプロジェクトを作成しました。このステップでは、KafkaのAdminClientクラスを利用して、クラスタをプログラムで管理するクラスを作成します。

dokafkaプロジェクトが保存されているディレクトリにナビゲートします。プロジェクトの構造は、ソースコードがの下に保存されていることを示しています。

クラスを AdminClientDemo.java という名前のファイルに保存します。以下のコマンドを実行して作成し、編集のために開きます:

nano src/main/java/com/dokafka/AdminClientDemo.java

次の行を追加してください:

package com.dokafka;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.Node;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

public class AdminClientDemo {
  private static final Logger log = LoggerFactory.getLogger(AdminClientDemo.class);

  public static void main(String[] args) {
    String bootstrapServers = "kafka1.your_domain:9092";

    Properties properties = new Properties();
    properties.put("bootstrap.servers", bootstrapServers);
    final KafkaAdminClient client = AdminClient.create(properties);

    try {
      Collection<Node> nodes = client.describeCluster().nodes().get();
      if (nodes == null)
        log.info("There seem to be no nodes in the cluster!");
      else
        log.info(String.format("Count of nodes: %s\n", nodes.size()));
    } catch (Exception e) {
      log.error("An error occurred", e);
    }
  }
}

まず、AdminClientDemoクラスを定義し、使用するクラスをインポートします。また、クラスのメンバーとしてLoggerのインスタンスを作成します。ProducerDemoと同様に、main()メソッドではまずKafkaクラスタのアドレス（bootstrapServers）を宣言します。kafka1.your_domainを実際のドメイン名に置き換えることを忘れないでください。

次に、Kafkaコンシューマーを操作するための設定を表すキーと値のペアを保持するPropertiesオブジェクトをインスタンス化します。bootstrap.serversプロパティをKafkaクラスターのアドレスに設定します。

その後、AdminClientを宣言してインスタンス化します。

final KafkaAdminClient client = AdminClient.create(properties);

クライアントは、トピック、パーティション、およびオフセットのリスト作成、作成、削除など、クラスタ内で管理アクションを実行できます。ここでは、クラスタ内に存在するノードの数を取得するためにそれを使用します:

...
    try {
      Collection<Node> nodes = client.describeCluster().nodes().get();
      if (nodes == null)
        log.info("There seem to be no nodes in the cluster!");
      else
        log.info(String.format("Count of nodes: %s\n", nodes.size()));
    } catch (Exception e) {
      log.error("An error occurred", e);
    }
...

もし返されるNodeオブジェクトがない場合は、適切なメッセージをログに記録します。そうでなければ、受け取ったKafkaノードの数を出力します。完了したら、ファイルを保存して閉じます。

次に、AdminClientDemoをコンパイルして実行するためのスクリプトを作成します。それをrun-adminclient.shというファイルに保存します。ファイルを作成し、編集のために開いてください:

nano run-adminclient.sh

次の行を追加してください:

#!/bin/bash
mvn clean
mvn package
java -cp target/dokafka-1.0-SNAPSHOT.jar:target/lib/* com.dokafka.AdminClientDemo

ファイルを保存して閉じてから、それを実行可能としてマークしてください:

chmod +x run-adminclient.sh

最後に、それを実行してみてください。

./run-adminclient.sh

出力は長くなるので、その終わりはこのように見えるはずです:

...
[main] INFO com.dokafka.AdminClientDemo - Count of nodes: 3

AdminClientDemoは、あなたのKafkaインストールに正常に接続し、ノードの数を取得しました。

トピックの作成とリスト化

KafkaAdminClient は、トピックを作成し、後でクラスター内に存在するすべてのトピックを一覧表示するために使用する createTopics() および listTopics() メソッドを提供します。

まず、次のコマンドを実行してAdminClientDemoを編集のために開きます:

nano src/main/java/com/dokafka/AdminClientDemo.java

コードを次のように修正してください:

...
    try {
      NewTopic newTopic = new NewTopic("newTopic", 1, (short) 1);
      CreateTopicsResult result = client.createTopics(
        Collections.singleton(newTopic)
      );
      result.all().get();

      ListTopicsOptions options = new ListTopicsOptions();
      options.listInternal(true);
      Collection<TopicListing> topics = client.listTopics(options).listings().get();
      for (TopicListing topic: topics) {
        log.info("Topic: " + topic.name());
      }
    } catch (Exception e) {
      log.error("An error occurred", e);
    }
...

最初に、クラスター内で作成されるトピックを説明する NewTopic のインスタンスを作成します。その名前（"newTopic"）とパーティション数およびレプリカ数を渡しますが、これらを short にキャストする必要があります。 createTopics() メソッドは、NewTopic インスタンスの Collection を受け取るので、newTopic だけを含む単一の要素を持つリストを構築します。

このメソッドは非同期であるため、呼び出されるとすぐに実行フローはmain()に戻りますが、バックグラウンドで実行を続けます。次にトピックをリストする必要があるので、result.all().get()で完了するのを待つ必要があります。resultには作成されたトピックのリストが含まれており、それらのall()を取得するには操作が完了する必要があります。

次に、トピックを取得するプロセスの設定を保持するListTopicsOptionsインスタンスを構築します。それをlistInteral()メソッドに渡し、Kafkaが内部使用のために考慮するトピックを取得することを許可するためにtrueを渡します。次に、client.listTopics()に渡し、listings()を取得します。その後、ループを通じてトピック名をログに記録します。

作業が終わったらファイルを保存して閉じてから、スクリプトを実行してください:

./run-adminclient.sh

クラスタ内のすべてのトピックは、出力の最後にリストされます。

...
[main] INFO com.dokafka.AdminClientDemo - Topic: newTopic
[main] INFO com.dokafka.AdminClientDemo - Topic: java_demo

トピックを削除する

トピックを削除するには（複数のトピックも含む）、KafkaAdminClientのdeleteTopics()メソッドを使用します。これを使って、あなたが作成したばかりのnewTopicを削除します。AdminClientDemoを編集のために開いてください:

nano src/main/java/com/dokafka/AdminClientDemo.java

トピック作成のコードをハイライトされた行で置き換えてください:

...
      DeleteTopicsResult deleted = client.deleteTopics(Collections.singleton("newTopic"));
      deleted.all().get();
      log.info("Topic newTopic deleted!");

      ListTopicsOptions options = new ListTopicsOptions();
      options.listInternal(true);
      Collection<TopicListing> topics = client.listTopics(options).listings().get();
      for (TopicListing topic: topics) {
        log.info("Topic: " + topic.name());
      }
...

ここでは、トピック名を表す文字列のCollectionをclient.deleteTopics()に渡します。次に、すべてのトピックが処理されるまで、deleted.all()を呼び出して処理が完了するのを待ちます。作業が完了したらファイルを保存して閉じ、次にrun-adminclient.shスクリプトを実行します:

./run-adminclient.sh

新しいトピックはトピックのリストに存在しないことがわかります。

[main] INFO com.dokafka.AdminClientDemo - Topic newTopic deleted!
[main] INFO com.dokafka.AdminClientDemo - Topic: java_demo

このステップでは、KafkaAdminClientを使用してクラスターにアクセスし、プログラムによってその情報を取得しました。また、クラスター内のトピックをリスト、作成、および削除する方法も見てきました。これから、kcat CLIツールを使用してクラスターにアクセスする方法を学びます。

ステップ2 - クラスターを管理するためのkcatの使用

このステップでは、Javaに依存せずにKafkaクラスターにアクセスし構成するためのコマンドラインツールであるkcatをダウンロードしてインストールする方法を学びます。

まず、あなたのOSに適したパッケージをインストールする必要があります。MacOSを使用している場合は、Brewでkcatをダウンロードできます:

brew install kcat

DebianおよびUbuntuシステムでは、通常通りaptを使用してインストールできます:

sudo apt install kafkacat

kafkacatはkcatの古い名前ですが、互換性の理由からパッケージマネージャーに残されています。他のディストリビューションのインストール手順については、公式ドキュメントを訪れてください。

メッセージの生成と消費

Kafkaに付属するスクリプトの基本的な操作の一つは、トピックからデータをストリーミングすることです。kcatを使用すると、複数のトピックから同時にストリーミングできます:

kcat -b your_broker_address:9092 first_topic second_topic ...

java_demoトピックからメッセージをコンソールにストリーミングするには、次のコマンドを実行します:

kcat -b kafka1.your_domain:9092 java_demo

Kafka-console-consumer.sh スクリプトは、トピックの最初からすべてのメッセージを読み取ることもできます。同様のことを kcat で達成するには、-t を渡してください:

kcat -b kafka1.your_domain:9092 -t java_demo

出力には、前提条件の一部として作成したメッセージが含まれます。

% Auto-selecting Consumer mode (use -P or -C to override)
% Reached end of topic java_demo [1] at offset 0
% Reached end of topic java_demo [2] at offset 0
Hello World!
% Reached end of topic java_demo [0] at offset 0
% Reached end of topic java_demo [3] at offset 0
% Reached end of topic java_demo [4] at offset 1
% Reached end of topic java_demo [5] at offset 0

消費されたメッセージをJSON形式で出力することもでき、-Jを渡すことで実行できます。

kcat -b kafka1.your_domain:9092 -t java_demo -J

出力はこのようになります:

% Auto-selecting Consumer mode (use -P or -C to override)
% Reached end of topic java_demo [2] at offset 0
% Reached end of topic java_demo [0] at offset 0
% Reached end of topic java_demo [1] at offset 0
{"topic":"java_demo","partition":4,"offset":0,"tstype":"create","ts":1714922509999,"broker":1,"key":null,"payload":"Hello World!"}
% Reached end of topic java_demo [3] at offset 0
% Reached end of topic java_demo [5] at offset 0
% Reached end of topic java_demo [4] at offset 1

トピックにメッセージを生成するには、-Pを渡してプロデューサーモードに切り替えます:

kcat -b kafka1.your_domain:9092 -t java_demo -P

同様に、kafka-console-producer.shと同じく、kcatはENTERで区切られたメッセージを受け入れます。プロンプトを終了するには、CTRL + Cを押し、続けてENTERを押して下さい。

kcatのテンプレーティング機能を使用して、メッセージに関する追加情報を出力することができます。お好きなフォーマットを指定するには、次のコマンドのように-fを使用してください:

kcat -b kafka1.your_domain:9092 -t java_demo -f 'Topic %t[%p], offset: %o, key: %k, payload: %S bytes: %s\n'

ここでは、java_demoトピックの最初からすべてのメッセージを読み取るように指示し、トピック名(%t)、パーティション番号(%p)、オフセット(%o)、メッセージキー(%k)、および値の長さ(%S)、メッセージ自体(%s)を出力します。

出力はこのようになります:

% Auto-selecting Consumer mode (use -P or -C to override)
% Reached end of topic java_demo [2] at offset 0
% Reached end of topic java_demo [1] at offset 0
Topic java_demo[4], offset: 0, key: , payload: 12 bytes: Hello World!
% Reached end of topic java_demo [0] at offset 0
% Reached end of topic java_demo [3] at offset 0
% Reached end of topic java_demo [4] at offset 1
% Reached end of topic java_demo [5] at offset 0

kcatは文字列テンプレートを使用し、メッセージレコードに関する追加情報を出力しました。

クラスタメタデータのリスト

クラスタのメタデータを整理された形式で表示するには、-Lを渡します:

kcat -b kafka1.your_domain:9092 -L

出力はクラスタ内のブローカー、トピック、およびそのパーティションを一覧表示します:

 3 brokers:
  broker 1 at kafka1.your_domain:9092
  broker 2 at kafka2.your_domain:9092 (controller)
  broker 3 at kafka3.your_domain:9092
 1 topics:
  topic "java_demo" with 6 partitions:
    partition 0, leader 3, replicas: 3,1 isrs: 3,1
    partition 1, leader 1, replicas: 1,2 isrs: 1,2
    partition 2, leader 2, replicas: 2,3 isrs: 2,3
    partition 3, leader 2, replicas: 2,1 isrs: 2,1
    partition 4, leader 1, replicas: 1,3 isrs: 1,3
    partition 5, leader 3, replicas: 3,2 isrs: 3,2

Kafka-topics.shスクリプトと同様に、kcatはパーティションリーダー、レプリカ、およびインシンクレプリカセットをリストします。

出力をJSON形式で得るために、-Jを渡すこともできます:

kcat -b kafka1.your_domain:9092 -L -J

出力はこのようになります:

{
    "originating_broker": {
        "id": 2,
        "name": "kafka2.your_domain:9092/2"
    },
    "query": {
        "topic": "*"
    },
    "controllerid": 3,
    "brokers": [
        {
            "id": 1,
            "name": "kafka1.your_domain:9092"
        },
        {
            "id": 2,
            "name": "kafka2.your_domain:9092"
        },
        {
            "id": 3,
            "name": "kafka3.your_domain:9092"
        }
    ],
    "topics": [
        {
            "topic": "java_demo",
            "partitions": [
                {
                    "partition": 0,
                    "leader": 3,
                    "replicas": [
                        {
                            "id": 3
                        }
                    ],
                    "isrs": [
                        {
                            "id": 3
                        }
                    ]
                },
                ...
            ]
        }
    ]
}

このステップでは、Javaに依存せずにKafkaクラスターに簡単にアクセスし、管理するためのツールであるkcatをインストールしました。クラスターやそのトピックに関する情報を取得する方法、メッセージを生成し消費する方法を見てきました。次に、Cruise Controlを使用して、クラスター内の適切な再バランスを自動化する方法を学びます。

ステップ3 - Kafka Cruise Controlによるリバランスの自動化

クルーズコントロールは、LinkedInが開発したオープンソースプロジェクトで、クラスタ内のKafkaブローカーの活動を継続的に監視し、リソースの使用量とスループットを最適化するために作業負荷を再バランスします。このステップでは、クラスターに対してそれを有効にし、その操作を監視する方法を学びます。

デフォルトでは、Cruise Controlにはクラスタが最適に動作するための適切な目標（ゴールと呼ばれる）が設定されています。主な目標には、各トピックが適切な数のレプリカを持つこと、CPU、ネットワーク、ディスクの使用量をバランスよくし、指定された制限内に保つこと、トピックパーティションのレプリカの数が十分であることが含まれます。カスタムゴールの実装もサポートされています。

ソースからCruise Controlをコンパイルする必要があります。まず、次のコマンドを実行して公式のGitリポジトリをクローンします:

git clone https://github.com/linkedin/cruise-control.git

それに移動してください:

cd cruise-control

次に、Gradleを使用してプロジェクトをコンパイルします:

./gradlew jar

ビルドプロセスには時間がかかります。出力の最後はこれに似た形になります:

...
BUILD SUCCESSFUL in 2m 41s
17 actionable tasks: 17 executed

クルーズコントロールは、メトリクスレポータとともにコンパイルされるようになりました。このレポータは、cruise-control-metrics-reporter/build/libs/のJARファイルにあります。レポータは、ブローカーに関するメトリクスをクルーズコントロールが監視できるクラスタのトピックに送信します。

次に、以下のコマンドを実行してすべての依存関係をターゲットディレクトリにコピーします:

./gradlew jar copyDependantLibs

出力は次のようになります:

BUILD SUCCESSFUL in 15s
17 actionable tasks: 1 executed, 16 up-to-date

次に、クラスター内のすべてのKafkaブローカーを特別なメトリクスレポーターを使用するように設定する必要があります。これを行うには、次のコマンドを実行してKafkaがインストールされているlibs/ディレクトリにコピーしてください:

cp cruise-control-metrics-reporter/build/libs/* /home/kafka/kafka/libs/

次に、新しいレポータを利用するためにKafkaブローカーの設定を変更する必要があります。ブローカーのserver.propertiesファイルを編集のために開いてください。

nano /home/kafka/kafka/config/kraft/server.properties

ファイルの末尾に次の行を追加してください:

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

作業が完了したら、ファイルを保存して閉じてください。ブローカーを再起動するには、次のコマンドを実行します:

sudo systemctl restart kafka

クラスタ内のすべてのブローカーはクルーズコントロールメトリックレポータを有効にする必要があるため、他のノードでもメトリックレポータJARを取得するために上記の手順を繰り返してください。数分後、kcatを使用してクラスタ内のトピックをリストすることで、メトリックレポータが動作を開始したかどうかを確認できます:

...
  topic "__CruiseControlMetrics" with 6 partitions:
    partition 0, leader 3, replicas: 3,2 isrs: 3,2
    partition 1, leader 2, replicas: 2,3 isrs: 2,3
    partition 2, leader 3, replicas: 3,2 isrs: 3,2
    partition 3, leader 2, replicas: 2,3 isrs: 2,3
    partition 4, leader 2, replicas: 2,3 isrs: 2,3
    partition 5, leader 3, replicas: 3,2 isrs: 3,2

ブローカーを正しく最適化するために、Cruise Controlは各ブローカーのハードウェア仕様を把握している必要があります。この設定は、configの下にあるcapacity.jsonというファイルに保存されています。編集のために開いてください:

nano config/capacity.json

デフォルトの状態では、ファイルは次のようになります:

{
  "brokerCapacities":[
    {
      "brokerId": "-1",
      "capacity": {
        "DISK": "100000",
        "CPU": "100",
        "NW_IN": "10000",
        "NW_OUT": "10000"
      },
      "doc": "This is the default capacity. Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
    },
    {
      "brokerId": "0",
      "capacity": {
        "DISK": "500000",
        "CPU": "100",
        "NW_IN": "50000",
        "NW_OUT": "50000"
      },
      "doc": "This overrides the capacity for broker 0."
    }
  ]
}

brokerCapacities 配列はブローカーIDに関連付けられた要素を含んでいます。各要素には、ディスクスペース、CPU許容量、およびブローカーへのネットワークスループットとブローカーからのネットワークスループットのフィールドが含まれています。

要件の一部として、3つのKafkaブローカーをデプロイしました。各ブローカーの正しい仕様を追加することでファイルを修正してください:

{
  "brokerCapacities":[
    {
      "brokerId": "-1",
      "capacity": {
        "DISK": "100000",
        "CPU": "100",
        "NW_IN": "10000",
        "NW_OUT": "10000"
      },
      "doc": "This is the default capacity. Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
    },
    {
      "brokerId": "1",
      "capacity": {
        "DISK": "100000",
        "CPU": "100",
        "NW_IN": "10000",
        "NW_OUT": "10000"
      },
      "doc": ""
    },
    {
      "brokerId": "2",
      "capacity": {
        "DISK": "100000",
        "CPU": "100",
        "NW_IN": "10000",
        "NW_OUT": "10000"
      },
      "doc": ""
    },
    {
      "brokerId": "3",
      "capacity": {
        "DISK": "100000",
        "CPU": "100",
        "NW_IN": "10000",
        "NW_OUT": "10000"
      },
      "doc": ""
    }
  ]
}

Dropletのサイズに応じてディスク容量を設定し、ファイルを保存して閉じてください。

次に、クルーズコントロールを構成して、KRaftモードでクラスタに接続します。configの下にあるcruisecontrol.propertiesファイルを修正します。次のコマンドを実行して編集のために開きます:

nano config/cruisecontrol.properties

次の行を見つけてください:

...
The Kafka cluster to control.
bootstrap.servers=localhost:9092
...

bootstrap.serversパラメータは、接続するブローカーおよびそのクラスターの範囲を指定します。これを次のように変更してください:

...
The Kafka cluster to control.
bootstrap.servers=kafka1.your_domain:9092

Switch to KRaft mode
kafka.broker.failure.detection.enable=true
...

パラメーター kafka.broker.failure.detection.enable は、Cruise Control に ZooKeeper 設定を無視し、KRaft モードをネイティブに使用するよう指示します。kafka1.your_domain をあなたの Kafka ブローカードメイン名に置き換えてください。

次に、次の行を見つけてください。

...
The configuration for the BrokerCapacityConfigFileResolver (supports JBOD, non-JBOD, and heterogeneous CPU core capacities)
#capacity.config.file=config/capacity.json
capacity.config.file=config/capacityJBOD.json
...

最初のパラメータ設定をアクティブにする:

...
The configuration for the BrokerCapacityConfigFileResolver (supports JBOD, non-JBOD, and heterogeneous CPU core capacities)
capacity.config.file=config/capacity.json
#capacity.config.file=config/capacityJBOD.json
...

ここでは、config/の下にあるcapacity.jsonファイルをブローカ容量の仕様として設定します。

作業が完了したらファイルを保存して閉じてください。Cruise Controlが設定されたので、次のコマンドを使用して別のターミナルで実行してください:

./kafka-cruise-control-start.sh config/cruisecontrol.properties

クルーズコントロールがクラスタを継続的に監視して最適化するため、多くのストリーミング出力が発生します。

クルーズコントロールCLIの使用

クルーズコントロールは、設定および管理タスクのためにポート9090でREST APIを公開しています。しかし、プロジェクトはまた、APIを直感的なコマンドラインインターフェースでラップするPythonツールcccliも維持しています。このセクションでは、それをインストールし、クラスターに関する情報を取得する方法を学びます。

事前条件として作成したPython 3の仮想環境に移動します。

cd ~/venv

次のコマンドを実行してアクティブにします:

./bin/activate

次に、pipを使用してツールをインストールします:

pip install cruise-control-client

出力の終わりはこのような感じになります:

...
Installing collected packages: urllib3, idna, charset-normalizer, certifi, requests, cruise-control-client
Successfully installed certifi-2024.2.2 charset-normalizer-3.3.2 cruise-control-client-1.1.3 idna-3.7 requests-2.31.0 urllib3-2.2.1

インストーラーはcccliコマンドを適切なPythonパッケージにバインドしました。

現在のクラスターの負荷に関する統計を取得するには、次のコマンドを実行し、あなたのクルーズコントロールインスタンスのアドレスを-aに渡します:

cccli -a localhost:9090 load

出力は、メトリックレポータによって提供された、監視しているパラメータの詳細を示します:

Starting long-running poll of http://localhost:9090/kafkacruisecontrol/load?allow_capacity_estimation=False


              HOST         BROKER              RACK         DISK_CAP(MB)            DISK(MB)/_(%)_            CORE_NUM         CPU(%)          NW_IN_CAP(KB/s)       LEADER_NW_IN(KB/s)     FOLLOWER_NW_IN(KB/s)         NW_OUT_CAP(KB/s)        NW_OUT(KB/s)       PNW_OUT(KB/s)    LEADERS/REPLICAS
kafka1.your_domain,             1,kafka1.your_domain,          100000.000,              0.023/00.00,                  1,         0.999,               10000.000,                   0.012,                   0.002,               10000.000,              0.028,              0.080,            23/45
kafka2.your_domain,             2,kafka2.your_domain,          100000.000,              0.238/00.00,                  1,         0.819,               10000.000,                   0.028,                   0.009,               10000.000,              0.066,              0.105,            27/48
kafka3.your_domain,             3,kafka3.your_domain,          100000.000,              0.383/00.00,                  1,         1.292,               10000.000,                   0.041,                   0.006,               10000.000,              0.081,              0.108,            28/49

クルーズコントロールは、ブローカーの障害、目標違反、またはメトリックの異常が発生した場合にクラスタを自動的に修復するように設定できます。ブローカーの障害が発生した場合の修復プロセスを有効にするには、次のコマンドを実行してください:

cccli -a localhost:9090 admin --enable-self-healing-for broker_failure

タッチされた設定の古い状態と新しい状態が表示されます:

{
selfHealingEnabledBefore: {BROKER_FAILURE=false}, selfHealingEnabledAfter: {BROKER_FAILURE=true}
}

このステップでは、クラスタ内の各KafkaブローカーノードにCruise Controlメトリクスレポータをダウンロードしてセットアップしました。その後、Cruise Controlを構成して開始し、クラスタの運用を継続的に監視および最適化します。また、コマンドラインからCruise Controlを即座に監視および構成するためのツールであるcccliのインストールと使用方法も見ました。提供される内容についての詳細は、公式ドキュメントを訪れてください。

結論

これで、KafkaAdminClient APIを使ってコードからKafkaクラスタを管理できるようになりました。また、kcatの使い方や、Kafkaクラスタの効率と信頼性を向上させるためにCruise Controlを設定する方法も学びました。

著者は「Write for DOnations」プログラムの一環として、寄付を受けるためにApache Software Foundationを選びました。