Apache Kafka

最終更新: 2026-07-19 19:31:59 +0800 +0800
tags: Apache-Kafka | Event-Streaming | Kafka
原文は英語で書かれています。あなたはAI翻訳版を読んでいます。

Apache Kafka は、永続的なイベントストリーム向けの分散型追記ログシステムです。 プロデューサーはトピックパーティションにレコードを追記し、ブローカーはそれらの レコードを複製して保持し、コンシューマーはオフセットによってレコードを読み取ります。 したがって Kafka は、複数の独立したアプリケーションが、可用性や処理速度を結合せずに、 同じ順序付けられた履歴を監視する必要がある場合に適しています。

クラスターの役割とメタデータ #

Kafka クラスターには、次の 2 つのサーバー役割があります。

  • ブローカーはパーティションレプリカを保存し、produce リクエストと fetch リクエストを処理します。
  • KRaft コントローラーは、クラスターのメタデータ、ブローカーのメンバーシップ、トピック設定、レプリカの配置、パーティションのリーダーシップを管理します。

開発用クラスターでは、1 つのプロセスが一方または両方の役割を担えます。本番 クラスターでは、データプレーンの負荷がメタデータクォーラムの判断に干渉しないよう、 コントローラーとブローカーの役割を分離します。ZooKeeper は Kafka のレガシーな メタデータアーキテクチャに属し、Kafka 4.x には存在しません。

クライアントは完全なルーティングテーブルではなく、ブートストラップアドレスの一覧を 受け取ります。メタデータリクエストによって各パーティションを現在リードしている ブローカーが特定され、その後の produce リクエストと fetch リクエストは該当する リーダーへ直接送信されます。したがって、通知されたすべてのブローカーアドレスは クライアントのネットワークから名前解決でき、到達可能でなければなりません。外部 クライアントにコンテナー内だけで有効なホスト名を公開することは、よくあるデプロイ エラーです。

Kafka は TCP 上でバージョン管理されたバイナリプロトコルを使用します。 org.apache.kafka:kafka-clients アーティファクトは、Java のプロデューサー、 コンシューマー、管理 API を提供します。他の言語のクライアントも同じプロトコルを 実装します。

トピック、パーティション、レコード #

トピックは順序付けられたパーティションに分割されます。各レコードには次を含められます。

  1. キー
  2. 値
  3. タイムスタンプ
  4. 0 個以上のヘッダー

プロトコル境界では、キーと値はバイト配列です。シリアライザーはアプリケーション オブジェクトをこれらのバイトに変換し、デシリアライザーは逆の処理を行います。

ブローカーは、1 つのパーティションに追記されたレコードに単調増加するオフセットを 割り当てます。オフセットはそのパーティション内の位置を識別するものであり、グローバル に順序付けられたイベント ID ではありません。Kafka が保証する順序は 1 つの パーティション内に限られ、トピック全体では保証されません。

プロデューサーがパーティションを指定しない場合、パーティショナーはレコードキーと 現在のトピックメタデータを使用します。その結果、パーティション数が変わらない限り、 等しい非 null キーは同じパーティションへルーティングされます。この性質により、 アカウント ID やデバイス ID などのキーについてエンティティ単位の順序を維持できます。 パーティション数を増やすとキーとパーティションの対応が変わる可能性があるため、 パーティション拡張には明示的な順序付けと移行の分析が必要です。

各パーティションはブローカーのログディレクトリ下にセグメントファイルとして保存されます。 セグメントファイルにはバイナリのレコードバッチとオフセットおよびタイムスタンプの インデックスが含まれ、行指向のテキストログではありません。

保持とコンパクション #

コンシュームしてもレコードは削除されません。ブローカーはトピックのクリーンアップ ポリシーに従ってレコードを削除します。

  • cleanup.policy=delete は、設定された時間またはサイズの境界を超えた古いセグメントを削除します。
  • cleanup.policy=compact は、各キーの最新の値を保持します。
  • cleanup.policy=compact,delete は両方のポリシーを適用します。

コンパクションはオフセットを保持しますが、密な連続列は生成しません。したがって コンシューマーは、オフセット間にギャップを見つけることがあります。null 値を持つ キー付きレコードは トゥームストーンです。delete.retention.ms の経過後、 コンパクションによってトゥームストーンと、そのキーの古い値の両方が削除されることがあります。

管理ツールでコンパクション対象の状態トピックを設定します。

sh 
bin/kafka-topics.sh --bootstrap-server broker:9092 \
  --create --topic customer-state \
  --partitions 12 --replication-factor 3 \
  --config cleanup.policy=compact \
  --config min.cleanable.dirty.ratio=0.5

コンパクションは非同期です。置換値またはトゥームストーンが追記された直後に、 古い値が消えるとアプリケーションが想定してはなりません。

レプリケーションと耐久性 #

パーティションには 1 つのリーダーと 1 つ以上のレプリカがあります。プロデューサーと コンシューマーは通常、リーダーと通信します。フォロワーはリーダーのログを取得し、 リーダーシップが変更されるとコントローラーが後継を選出します。

確認済み書き込みの耐久性は、複数の独立した設定によって決まります。

  • replication.factor は、パーティションを保存するブローカー数を決定します。
  • acks=all は、リーダーが in-sync replica 要件を満たすまでプロデューサーを待機させます。
  • min.insync.replicas は、in-sync のレプリカが少なすぎる場合に acks=all の書き込みを拒否します。
  • 非クリーンなリーダー選出を許可すると、古いレプリカがリーダーになれるため、確認済みレコードが失われる可能性があります。

たとえば、レプリケーション係数 3 と min.insync.replicas=2 の構成では、2 コピーの 確認ルールを維持しながら、1 レプリカの利用不能を許容できます。ただし、書き込みに ついて 2 レプリカの利用不能は許容できません。この選択によって可用性と耐久性が結び付きます。

レプリカの配置は、ホスト、ラック、アベイラビリティゾーンなどの障害ドメインをまたぐ 必要があります。1 台の物理ホストに 3 レプリカを配置しても、ホスト障害への耐性は得られません。

トピックの作成と検査 #

トピック数とパーティション数は運用上の容量に関する判断であるため、重要なトピックは ブローカー側の自動作成に依存せず、明示的に作成します。

sh 
bin/kafka-topics.sh --bootstrap-server broker:9092 \
  --create --topic orders \
  --partitions 12 --replication-factor 3 \
  --config min.insync.replicas=2 \
  --config retention.ms=604800000

bin/kafka-topics.sh --bootstrap-server broker:9092 \
  --describe --topic orders

bin/kafka-configs.sh --bootstrap-server broker:9092 \
  --entity-type topics --entity-name orders --describe

Java Admin API でも同じコントロールプレーンを操作できます。

java 
import java.util.List;
import java.util.Map;
import java.util.Properties;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;

Properties config = new Properties();
config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");

try (Admin admin = Admin.create(config)) {
    NewTopic orders = new NewTopic("orders", 12, (short) 3)
        .configs(Map.of(
            "min.insync.replicas", "2",
            "retention.ms", "604800000"
        ));
    admin.createTopics(List.of(orders)).all().get();
}

createTopics は非同期です。返された future を待機すると、認可、検証、タイムアウトの 失敗が表面化します。

Java クライアントの依存関係 #

プロデューサー、コンシューマー、Admin クライアントは 1 つのアーティファクトに含まれます。

xml 
<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>${kafka.version}</version>
</dependency>

クライアントのバージョンは Kafka の互換性ドキュメントから選択します。プロトコル ネゴシエーションにより、多くの異なるクライアントとブローカーのバージョンを混在 させられますが、新しい API とレコード形式には依然として最低ブローカー要件があります。

レコードの生成 #

KafkaProducer はスレッドセーフです。パーティションごとのバッチにレコードを蓄積し、 各バッチを圧縮して、リクエストを非同期に送信します。したがって、長期間存続する プロデューサーは、レコードごとに 1 つのプロデューサーを構築するよりも高い性能を発揮します。

java 
import java.nio.charset.StandardCharsets;
import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

Properties config = new Properties();
config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
    StringSerializer.class.getName());
config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
    StringSerializer.class.getName());
config.put(ProducerConfig.ACKS_CONFIG, "all");
config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
config.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "zstd");
config.put(ProducerConfig.LINGER_MS_CONFIG, 5);

try (KafkaProducer<String, String> producer = new KafkaProducer<>(config)) {
    ProducerRecord<String, String> record =
        new ProducerRecord<>("orders", "customer-42", "{\"orderId\":123}");
    record.headers().add("schema-version",
        "1".getBytes(StandardCharsets.UTF_8));

    producer.send(record, (metadata, error) -> {
        if (error != null) {
            System.err.println("Produce failed: " + error);
            return;
        }
        System.out.printf("%s-%d@%d%n",
            metadata.topic(), metadata.partition(), metadata.offset());
    });
}

プロデューサーを閉じると、未処理の送信がフラッシュされます。アプリケーションは コールバックの失敗を伝播または記録する必要があります。クライアントのバッファに レコードを受け入れたことは、ブローカーが確認したことを意味しません。

スループットとレイテンシーに関係する重要な制御項目には次があります。

  • batch.size: 1 つのパーティションバッチに蓄積できる最大バイト数
  • linger.ms: 部分的に埋まったバッチに利用できる時間
  • compression.type: バッチ全体に適用する圧縮方式
  • buffer.memory: 未送信レコードに使用できるプロデューサーのメモリ
  • delivery.timeout.ms: 送信完了までの上限
  • max.request.size: クライアントが受け入れる最大リクエストサイズ

大きなバッチは圧縮率とリクエスト効率を向上させますが、メモリを消費し、レイテンシーを 増加させる場合があります。ブローカーとトピックのメッセージサイズ上限も、生成された バッチを受け入れられる値でなければなりません。

冪等性と再試行 #

ネットワーク障害によって、ブローカーがリクエストを追記したかどうかをプロデューサーが 判断できなくなることがあります。その状態で無条件に再試行すると、重複が生成される 可能性があります。冪等性のある生成では、プロデューサーの識別情報とシーケンス番号を 付加し、ブローカーがプロデューサーセッション内の重複した再試行を拒否できるようにします。

冪等性は、独立した 2 つのビジネスリクエスト、異なるプロデューサーインスタンスから 送信されたレコード、またはアプリケーションレベルのリプレイ後に再送されたレコードを 重複排除しません。安定したビジネス ID と下流側の冪等性は引き続き必要です。

レコードのコンシューム #

KafkaConsumer はスレッドセーフではありません。各コンシューマーは 1 つの アプリケーションスレッドが所有し、poll の呼び出し、割り当てコールバックの処理、 オフセットのコミットを行うべきです。並列処理には、完了した処理とオフセットコミット の間で慎重な調整が必要です。

java 
import java.time.Duration;
import java.util.List;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

Properties config = new Properties();
config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
config.put(ConsumerConfig.GROUP_ID_CONFIG, "order-indexer");
config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
    StringDeserializer.class.getName());
config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
    StringDeserializer.class.getName());
config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(config)) {
    consumer.subscribe(List.of("orders"));

    while (!Thread.currentThread().isInterrupted()) {
        ConsumerRecords<String, String> records =
            consumer.poll(Duration.ofSeconds(1));

        for (ConsumerRecord<String, String> record : records) {
            indexOrder(record.key(), record.value());
        }

        // Commits the next offsets after the complete batch succeeds.
        consumer.commitSync();
    }
}

この例は少なくとも 1 回の処理を提供します。indexOrder の後、commitSync の前に 障害が発生すると、再生が起こります。したがって indexOrder は冪等である必要が あります。または、送信先が安定したイベント ID によって重複排除しなければなりません。

auto.offset.reset は、グループに有効なコミット済みオフセットがない場合にのみ適用 されます。確立済みのグループにトピックを再読込させる設定ではありません。意図的な リプレイには seek、新しいグループ ID、または consumer-groups のリセット操作を使用します。

コンシューマーグループとリバランシング #

同じ group.id を持つコンシューマーはグループを形成します。各パーティションは、その グループ内で最大 1 つのアクティブなコンシューマーに割り当てられます。

  • コンシューマー数がパーティション数より少ない場合、一部のコンシューマーが複数のパーティションを所有します。
  • 数が等しい場合、コンシューマーごとに 1 パーティションを割り当てられます。
  • コンシューマー数がパーティション数より多い場合、余分なコンシューマーはアイドル状態になります。

異なるグループは、トピック履歴の独立したコピーを受け取ります。したがって、詐欺検知器 と検索インデクサーには異なるグループ ID を使用します。

メンバーシップまたはサブスクリプションが変更されると、割り当てが更新されます。 協調的アサイナーはパーティションの移動を減らし、静的メンバーシップは短時間の再起動 中に不要な再割り当てを避けられます。ただし、アプリケーションは取り消された パーティションの処理を停止し、完了したオフセットだけをコミットする必要があります。

割り当てとラグを次のコマンドで検査します。

sh 
bin/kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --describe --group order-indexer

ラグは log-end-offset - committed-offset です。これは未コミットの距離を測定する ものであり、処理レイテンシーではありません。低トラフィックのパーティションでは ラグが小さくても、未処理レコードの最古のものが古い場合があります。そのため、レコード ラグと時間ベースのサービス目標の両方を監視します。

max.poll.interval.ms は poll 呼び出し間の時間を制限します。大きなバッチを同期的に 処理するとこの間隔を超え、コンシューマーがグループから削除される可能性があります。 max.poll.records でバッチを制限する、正当な低速処理に合わせて間隔を増やす、または パーティションを一時停止し、連続した完了済みオフセットだけをコミットしながら処理を executor に移す方法があります。

配信セマンティクス #

Kafka アプリケーションは次の 3 つの結果を実装できます。

  • 最大 1 回: 処理前に進行またはコミットし、障害後の損失を受け入れます。
  • 少なくとも 1 回: コミット前に処理し、再生と重複を受け入れます。
  • Kafka の正確に 1 回の処理: Kafka トランザクションによって、生成されたレコードと消費したオフセットをアトミックにコミットします。

Kafka トランザクションは、データベース書き込み、メール、HTTP 呼び出しをアトミックには しません。外部作用には、冪等性キー、outbox/inbox 設計、重複排除、または別の調整 プロトコルが必要です。

トランザクションによる Consume-Transform-Produce #

各稼働中プロデューサーインスタンスには、安定した一意の transactional.id が必要です。 同じ ID を同時に再利用すると、古いプロデューサーがフェンスされ、2 つのインスタンスが 同じトランザクション所有者としてコミットすることを防ぎます。

java 
import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

Properties consumerConfig = new Properties();
consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, "order-normalizer");
consumerConfig.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
consumerConfig.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
    StringDeserializer.class.getName());
consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
    StringDeserializer.class.getName());

Properties producerConfig = new Properties();
producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
producerConfig.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,
    "order-normalizer-instance-1");
producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
    StringSerializer.class.getName());
producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
    StringSerializer.class.getName());

try (KafkaConsumer<String, String> consumer =
         new KafkaConsumer<>(consumerConfig);
     KafkaProducer<String, String> producer =
         new KafkaProducer<>(producerConfig)) {
    consumer.subscribe(List.of("orders"));
    producer.initTransactions();

    while (true) {
        var records = consumer.poll(Duration.ofSeconds(1));
        if (records.isEmpty()) {
            continue;
        }

        try {
            producer.beginTransaction();
            Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();

            for (var record : records) {
                producer.send(new ProducerRecord<>(
                    "normalized-orders", record.key(),
                    normalize(record.value())));
                offsets.put(
                    new TopicPartition(record.topic(), record.partition()),
                    new OffsetAndMetadata(record.offset() + 1)
                );
            }

            producer.sendOffsetsToTransaction(
                offsets, consumer.groupMetadata());
            producer.commitTransaction();
        } catch (ProducerFencedException fatal) {
            throw fatal;
        } catch (KafkaException retryableBatchFailure) {
            producer.abortTransaction();
        }
    }
}

中止されたレコードを隠す必要があるコンシューマーは、isolation.level=read_committed を使用します。トランザクションは調整コストとレイテンシーを増加させるため、冪等な 処理の代替として既定で使用するのではなく、Kafka の読み取り・処理・書き込みを アトミックにする必要がある場合に使用します。

Kafka Streams #

Kafka Streams は、別個のクラスターサービスではなく、組み込み可能な Java ライブラリです。 各アプリケーションインスタンスはコンシューマーグループに参加し、処理をタスクに分割し、 changelog トピックによってバックアップされたローカル状態を保存します。

java 
import java.util.Properties;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;

Properties config = new Properties();
config.put(StreamsConfig.APPLICATION_ID_CONFIG, "paid-order-counts");
config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
    Serdes.String().getClass());
config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
    Serdes.String().getClass());
config.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, "exactly_once_v2");

StreamsBuilder builder = new StreamsBuilder();
builder.<String, String>stream("orders")
    .filter((customerId, json) -> json.contains("\"status\":\"PAID\""))
    .groupByKey()
    .count()
    .toStream()
    .to("paid-orders-per-customer");

KafkaStreams streams = new KafkaStreams(builder.build(), config);
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
streams.start();

application.id を変更すると、異なるコンシューマーグループ、内部トピック、状態 ディレクトリを持つ新しい処理 ID が作成されます。表示名ではなく、永続的な アプリケーション状態として扱います。

Kafka Connect とレプリケーション #

Kafka Connect は再利用可能なソースコネクターとシンクコネクターを実行します。 ソースコネクターは外部システムからレコードを取り込み、シンクコネクターは Kafka レコードをエクスポートします。分散ワーカーはコネクター設定、オフセット、ステータスを Kafka トピックに保存し、ワーカー間でコネクタータスクをリバランスします。

Connect は標準的な統合を簡素化しますが、次を定義する必要性はなくなりません。

  • レコードキーとパーティショニング
  • スキーマ互換性
  • 削除とトゥームストーンのセマンティクス
  • 再試行とデッドレターの動作
  • 送信先の冪等性
  • 認証情報とネットワーク境界

MirrorMaker 2 は Connect を使用して Kafka クラスター間でトピックをレプリケートします。 レプリケーションは非同期であり、2 つのクラスターを 1 つの同期コンセンサスシステムに 変えるものではありません。災害復旧手順では、レプリケーションラグ、コンシューマー オフセットの変換、トピック設定、フェイルバックを考慮する必要があります。

スキーマと進化 #

Kafka はバイトを保存するだけで、アプリケーションスキーマを強制しません。Avro、 Protobuf、JSON Schema、または別のエンコーディングでレコード構造を定義できます。 スキーマレジストリはスキーマ ID とペイロードを関連付け、互換性のない変更を拒否できます。

互換性ルールは、データがどのようにリプレイされるかを反映する必要があります。

  • 新しいコンシューマーが、数か月保持された古いレコードに遭遇することがある。
  • プロデューサーの変更中も古いコンシューマーがデプロイされたままになることがある。
  • コンパクションされたトピックは古い値を無期限に保持することがある。
  • アーカイブからの復元によって、古いスキーマが再導入されることがある。

定義済みのデフォルト値を持つオプションフィールドの追加は、フィールドの意味を変更したり 名前を再利用したりするよりも進化させやすいものです。シリアル化形式が機械的な変更を 許容する場合でも、イベント名とフィールドのセマンティクスは長期間存続する契約を形成します。

セキュリティ #

Kafka はトランスポートセキュリティ、認証、認可を分離します。

  • TLS はトラフィックを暗号化し、クライアントに対してブローカーを認証します。
  • mTLS または SASL メカニズムはクライアントを認証します。
  • ACL はトピック、グループ、トランザクション ID、クラスターリソースに対する操作を認可します。

TLS クライアント設定はプロパティで指定できます。

properties 
security.protocol=SSL
ssl.truststore.location=/run/secrets/kafka.truststore.p12
ssl.truststore.password=${TRUSTSTORE_PASSWORD}
ssl.keystore.location=/run/secrets/client.keystore.p12
ssl.keystore.password=${KEYSTORE_PASSWORD}
ssl.key.password=${KEY_PASSWORD}

実際のパスワードをコミット済みのプロパティファイルに記載しないでください。 デプロイプラットフォームのシークレット機構から注入し、ファイルシステムへのアクセスを 制限します。

パーティションと容量計画 #

パーティション数はコンシューマーグループの並列性を制限しますが、パーティションを 増やしてもコストがかからないわけではありません。各パーティションはメタデータ、 オープンファイル、レプリカ取得処理、選出、復旧トラフィックを増加させます。開始時の パーティション数は、次から見積もります。

  1. ピーク時の ingress および egress スループット
  2. 有効なコンシューマー同時実行数の最大値
  3. ブローカーごとのネットワークおよびディスクスループット
  4. キーの偏り
  5. レプリケーションと復旧の目標
  6. 再パーティショニングを許容できるまでの予想成長量

保持容量はおおよそ次の式で表せます。

text 
ingress bytes/second × retention seconds × replication factor

セグメント粒度、インデックス、コンパクション、一時的な偏り、再割り当て、ブローカー 復旧のための余裕を追加します。定常トラフィックを処理できても、復旧目標内に 障害ブローカーを再構築できないクラスターは、容量不足です。

運用と障害診断 #

少なくとも次を監視します。

  • オフラインパーティションとレプリケーション不足のパーティション
  • コントローラーとブローカーの可用性
  • produce と fetch のレイテンシーおよびエラー率
  • リクエストハンドラーとネットワークスレッドの飽和
  • ディスク使用量とディスクレイテンシー
  • レプリカ取得のラグ
  • コンシューマーオフセットのラグとレコード経過時間
  • トランザクションと認証の失敗
  • パーティションの偏りと再割り当ての進行状況

有用な診断コマンドには次があります。

sh 
# Observe records without joining an application consumer group.
bin/kafka-console-consumer.sh --bootstrap-server broker:9092 \
  --topic orders --from-beginning \
  --property print.key=true --property print.offset=true

# Inspect one group's committed offsets and lag.
bin/kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --group order-indexer --describe

# Read broker API compatibility information.
bin/kafka-broker-api-versions.sh --bootstrap-server broker:9092

利用できないリーダー、in-sync レプリカ不足、到達不能な通知済みリスナー、認可失敗、 過負荷状態のブローカーは、いずれもアプリケーションからは送信タイムアウトに見える 可能性があります。再試行上限を変更する前に、クライアント例外をブローカーの メトリクス、パーティション状態、コントローラーイベントと相関させます。

参考資料 #

コメント

GitHub で閲覧・コメントする（投稿にはログインが必要です）