JA
Tiếng ViệtEnglish한국어Deutsch日本語PortuguêsEspañolFrançais简体中文
ホーム / アーキテクチャ・設計 / Kafka

Kafkaの監視: consumer lagと監視対象のメトリクス

Cris The Dev · 2026/10/12 · 9 分で読めます · 418

1. 概要

Kafkaシステムは一見「健全」に見えるかもしれません。Brokerは実行中、Producerはメッセージを送れています。しかし、Consumerは数百万のメッセージの後ろに遅れていて、誰も気付かないかもしれません。この状態を反映するメトリクスがconsumer lagです。

この記事では、consumer lagの計算方法と、CLIやJavaのAdminClient、Spring BootのMicrometerを使った3つの測定方法を学びます。その後、brokerやproducer、consumerの最も重要なメトリクスと、それらに対するアラート設定方法を見ていきます。

2. Consumer lagとは何か

Kafkaの各partitionは、追記専用のログです。注目すべき2つのoffsetがあります。

  • Log-end offset (LEO): partitionに次に書き込まれるメッセージのoffset。
  • Committed offset: consumer groupがcommitしたoffset。つまり、restartされた場合にgroupが読み続ける位置。

1つのpartitionのlagは次のように計算されます。

lag = log-end offset - committed offset

Consumer group全体のlagは、そのgroupが読むすべてのpartitionのlagの合計です。

Lagが0は、consumerが追い付いたことを意味します。メッセージが継続的に到着するため、lagが軽く変動することは正常です。Lagが時間経過に伴い一貫して増加することは懸念すべき兆候です。その場合、consumerの処理速度がproducerの書き込み速度より遅いということです。

注意: トランザクションやcompactionを使うtopicでは、offsetに「隙間」（制御レコード、削除されたメッセージ）がある場合があります。Lagはoffsetベースで計算されるため、実際に読む必要があるメッセージ数より若干多くなる可能性があります。これはエラーではありません。

3. CLIでlagを確認する

最速の方法は、Kafkaに付属するスクリプトを使うことです。

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group order-service

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

GROUP          TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG   CONSUMER-ID       HOST
order-service  orders  0          15230           15232           2     consumer-1-a1b2   /10.0.0.5
order-service  orders  1          14980           18410           3430  consumer-1-c3d4   /10.0.0.6
order-service  orders  2          15102           15102           0     consumer-1-e5f6   /10.0.0.7

Partition 1は3430メッセージ遅れていますが、他の2つのpartitionはほぼ追い付いています。このようなlagのずれには通常2つの原因があります。1つは「ホット」なkeyが1つのpartitionに集中すること。もう1つはhost 10.0.0.6 上のconsumerに問題があることです。

CLIはトラブルシューティング時に非常に有用です。ただし、毎分このコマンドを手動で実行することはできません。次のセクションでは、lag測定を自動化します。

4. AdminClientでlagを計算する

kafka-clients依存関係を追加します（Spring Bootがversionを管理します）。

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
</dependency>

まず、計算ロジックを純粋な関数に抽出してテストしやすくします。

public final class LagCalculator {

    private LagCalculator() {}

    public static Map<TopicPartition, Long> compute(Map<TopicPartition, Long> committed,
                                                    Map<TopicPartition, Long> endOffsets) {
        Map<TopicPartition, Long> lag = new HashMap<>();
        committed.forEach((tp, committedOffset) -> {
            Long end = endOffsets.get(tp);
            if (end != null) {
                lag.put(tp, Math.max(0L, end - committedOffset));
            }
        });
        return lag;
    }
}

この関数は、commitされたoffsetとend offsetの両方を持つpartitionのみのlagを計算します。Math.max(0L, …) には理由があります。Committedされたoffsetとendのoffsetは別々のリクエストで取得されるため、commitされたoffsetが取得したばかりのend offsetより「新しい」ことがあります。その場合、負の結果は意味がないので、0に設定します。

次はAdminClientを呼び出すクラスです。

public class ConsumerLagChecker {

    private final Admin admin;

    public ConsumerLagChecker(Admin admin) {
        this.admin = admin;
    }

    public Map<TopicPartition, Long> lagOf(String groupId) throws Exception {
        Map<TopicPartition, Long> committed = admin.listConsumerGroupOffsets(groupId)
                .partitionsToOffsetAndMetadata()
                .get(10, TimeUnit.SECONDS)
                .entrySet().stream()
                .filter(e -> e.getValue() != null)
                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().offset()));

        Map<TopicPartition, OffsetSpec> request = committed.keySet().stream()
                .collect(Collectors.toMap(tp -> tp, tp -> OffsetSpec.latest()));

        Map<TopicPartition, Long> endOffsets = admin.listOffsets(request)
                .all()
                .get(10, TimeUnit.SECONDS)
                .entrySet().stream()
                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().offset()));

        return LagCalculator.compute(committed, endOffsets);
    }
}

処理は次のようになります。

  • listConsumerGroupOffsetsはgroupのcommittedされたoffsetを返します。partitionが一度もcommitされていない場合、値はnullの可能性があるため、フィルタリングしています。
  • listOffsetsをOffsetSpec.latest()と一緒に使用して、各partitionのlog-end offsetを取得します。
  • 各呼び出しにはタイムアウトが10秒あります。これにより、clusterに問題が発生した場合にthreadが無限に待機することはありません。

使用方法:

try (Admin admin = Admin.create(Map.of(
        AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"))) {
    new ConsumerLagChecker(admin).lagOf("order-service")
            .forEach((tp, lag) -> System.out.println(tp + " -> " + lag));
}
orders-0 -> 2
orders-1 -> 3430
orders-2 -> 0

4.1. LagCalculatorのテスト

JUnit 5とAssertJを使用します（spring-boot-starter-testに含まれます）。

class LagCalculatorTest {

    private final TopicPartition p0 = new TopicPartition("orders", 0);
    private final TopicPartition p1 = new TopicPartition("orders", 1);

    @Test
    void shouldComputeLagPerPartition() {
        Map<TopicPartition, Long> lag = LagCalculator.compute(
                Map.of(p0, 100L, p1, 50L),
                Map.of(p0, 120L, p1, 50L));

        assertThat(lag).containsEntry(p0, 20L).containsEntry(p1, 0L);
    }

    @Test
    void shouldClampNegativeLagToZero() {
        Map<TopicPartition, Long> lag = LagCalculator.compute(
                Map.of(p0, 130L),
                Map.of(p0, 120L));

        assertThat(lag).containsEntry(p0, 0L);
    }

    @Test
    void shouldSkipPartitionWithoutEndOffset() {
        Map<TopicPartition, Long> lag = LagCalculator.compute(
                Map.of(p0, 10L), Map.of());

        assertThat(lag).isEmpty();
    }
}

5. Spring BootのMicrometerにlagを送信する

ActuatorとPrometheusレジストリを追加します。

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-registry-prometheus</artifactId>
</dependency>
management.endpoints.web.exposure.include=health,prometheus

5.1. Client側の組み込みメトリクス

MeterRegistryがある場合、Spring Bootは自動的にMicrometerConsumerListenerをConsumerFactoryに接続します。Kafkaコンシューマーのすべての内部メトリクスは/actuator/prometheusに表示されます。例として：

kafka_consumer_fetch_manager_records_lag_max{client_id="consumer-order-service-1",...} 3430.0
kafka_consumer_fetch_manager_records_lag{topic="orders",partition="1",...} 3430.0

records-lag-maxはこのconsumerが観測した最大lagです。Kafkaクライアントはcommittedされたoffsetではなく、fetch位置から自動的に計算します。

5.2. 外部から独立して測定するlagのGauge

クライアント側のメトリクスには大きな弱点があります。Consumerが停止すると、メトリクスも消えます。 Lagが最も急速に増加しているまさにその時に、データがなくなります。そのため、AdminClientを使用した独立したプロセスからlagを測定する必要があります。

@Component
public class ConsumerLagMetrics {

    private final ConsumerLagChecker checker;
    private final MeterRegistry registry;
    private final Map<TopicPartition, AtomicLong> gauges = new ConcurrentHashMap<>();

    public ConsumerLagMetrics(KafkaAdmin kafkaAdmin, MeterRegistry registry) {
        this.checker = new ConsumerLagChecker(Admin.create(kafkaAdmin.getConfigurationProperties()));
        this.registry = registry;
    }

    @Scheduled(fixedDelay = 30_000)
    public void refresh() throws Exception {
        checker.lagOf("order-service").forEach((tp, lag) ->
                gauges.computeIfAbsent(tp, this::register).set(lag));
    }

    private AtomicLong register(TopicPartition tp) {
        AtomicLong holder = new AtomicLong();
        Gauge.builder("kafka.consumer.group.lag", holder, AtomicLong::get)
                .tag("group", "order-service")
                .tag("topic", tp.topic())
                .tag("partition", String.valueOf(tp.partition()))
                .register(registry);
        return holder;
    }
}

@Configurationクラスで@EnableSchedulingを有効にしてください。30秒ごとに、gauge kafka_consumer_group_lagが更新されます。この値はcommittedされたoffsetに基づいているため、groupのすべてのconsumerが停止した場合でも正確です。

Production環境では、多くのチームは上記のコードではなく、kafka-exporterやBurrowなどの専門ツールを使用しています。それらの原理は同じです。commitされたoffsetをlog-end offsetと比較します。

注意: Service consumerが複数のinstanceを持つ場合、このgaugeをその中に配置しないでください。各instanceが同じlagの値を出力し、dashboardの合計が間違ったものになります。別の監視serviceで実行するか、1つのinstanceでのみ実行してください。

6. 監視すべきメトリクス

Lagは問題があることを示しています。以下のメトリクスは、問題がどこにあるのかを見つけるのに役立ちます。

6.1. Broker

メトリクス (JMX) 説明 望ましい値
UnderReplicatedPartitions レプリケーションが同期していないpartition 0
OfflinePartitionsCount Leaderを持たないpartition 0
ActiveControllerCount アクティブなcontrollerの数（cluster全体） 1
IsrShrinksPerSec / IsrExpandsPerSec ISRの縮小・拡大 ほぼ0、連続的な変動なし
RequestHandlerAvgIdlePercent Request handler threadの遊休率 > 0.3
NetworkProcessorAvgIdlePercent Network threadの遊休率 > 0.3

OfflinePartitionsCount > 0は深刻な問題です。そのpartitionは読み取りと書き込みができません。UnderReplicatedPartitionsが長時間続く場合、brokerの過負荷、ディスクの低速化、またはネットワークの問題を示していることが多いです。

6.2. Producer

  • record-error-rate: 再試行を使い果たした後の送信失敗率。0以外の値はすべて調査する価値があります。
  • request-latency-avg: brokerへのリクエストレイテンシー。増加するとthroughputが低下します。
  • record-retry-rate: 多くの再試行はbrokerまたはネットワークが不安定であることを意味します。
  • buffer-available-bytes: bufferが枯渇するとsend()がブロックされます。

6.3. Consumer

  • records-lag-max: 上記で説明しました。
  • records-consumed-rate: 消費速度。topicへのproducerの書き込み速度と比較してください。
  • rebalance-rate-per-hourとlast-rebalance-seconds-ago: 連続したrebalanceはconsumerを「固まらせ」ます。これはlagが増加する一般的な原因です。
  • commit-latency-avg: commitが遅いとthroughputが低下します。特にcommitが同期的な場合です。
  • last-poll-seconds-ago: max.poll.interval.msを超えると、consumerはgroupから削除されます。

7. 合理的なアラート設定

最も一般的なエラーは、「lag > 1000の場合はアラート」のような絶対的な閾値を設定することです。50,000メッセージ/秒のtopicの場合、1000は無意味な数字です。1分あたり数メッセージのtopicの場合、1000は既に数時間の遅延を意味する可能性があります。

より良い2つのアプローチがあります。1つ目は、lagが継続的に増加するときにアラートすることです。

groups:
  - name: kafka-consumer
    rules:
      - alert: KafkaConsumerLagGrowing
        expr: sum by (group, topic) (deriv(kafka_consumer_group_lag[10m])) > 0
          and sum by (group, topic) (kafka_consumer_group_lag) > 500
        for: 15m
        labels:
          severity: warning
        annotations:
          summary: "{{ $labels.group }}の{{ $labels.topic }}上のlagが15分間継続的に増加しています"

このルールは、2つの条件が両方とも真の場合にのみトリガーされます。lagに増加傾向がある（deriv > 0）と、最小レベルを超えています。この条件は15分間続く必要があります。これにより、デプロイ後の短期的な増加を無視できます。

2つ目のアプローチは、lagを時間に変換することです。lagを消費速度で除して、consumerがどの程度遅れているかを秒単位で推定します。この数字はSLAと直接関連しています。例えば、「注文は2分以内に処理される必要があります」は「lag下 5000」よりもはるかに理解しやすいです。

さらに、absent(kafka_consumer_group_lag)などのメトリクスが消えた場合に対する別のアラートを用意する必要があります。データがないことは、すべてが問題ないことを意味しません。

8. 一般的なエラー

  • records-lag-maxのみに依存する。 Consumerが停止するとこのメトリクスは消えます。常に外部からのlag測定ソースを持つことが必要です。
  • アクティブなconsumerがないgroupのlagを読む。 CLIはcommittedされたoffsetをまだ表示しますが、CONSUMER-ID列は空です。ここでのlag増加は、誰も処理していないことを意味します。
  • Partitionの数を超えてconsumerを増やす。 余分なconsumerはアイドル状態になり、lagは減少しません。並列処理を増やすには、partitionの数を増やす必要があります。
  • Rebalanceを見落とす。 バッチ処理が長すぎてmax.poll.interval.msを超えます。Consumerが削除され、groupがrebalanceし、その後繰り返されます。Lagはのこぎり歯のように増加します。max.poll.recordsを減らすか、重い処理を別のthreadに移してください。
  • AdminClientを頻繁に呼び出す。 数百のgroupに対して毎秒offsetをクエリすると、brokerに不要な負荷が生じます。ほとんどのシステムでは、15〜60秒のサイクルで十分です。

9. 結論

Consumer lagはKafkaシステムがデータに追い付いているかどうかを示す最も直接的なメトリクスです。Lagはlog-end offsetからcommittedされたoffsetを引くことで計算されます。kafka-consumer-groups.shでlagを表示したり、AdminClientで計算したり、Micrometerを使ってPrometheusに出力したりできます。

効果的な監視のために、クライアント側のメトリクスと独立した測定ソースを組み合わせます。Lagとともに、UnderReplicatedPartitionsやOfflinePartitionsCountなどのbrokerメトリクス、およびconsumer側のrebalance頻度を監視します。最後に、固定されたlag数ではなく、傾向と遅延時間に基づいてアラートを設定します。こうすることで、誤報が少なくなり、実際の問題を見落とさなくなります。

kafkamonitoringspring bootmicrometerprometheus
著者のアプリGlowmi · Lantern NightA cozy Mid-Autumn match-3 puzzleiOS · Android

同じトピックの記事

Kafka · 2026/10/12 · 8 分で読めます · 312

Change Data CaptureとDebeziumとKafka

Debeziumを使ってPostgreSQLのWALを読み込み、すべてのデータ変更をKafkaにプッシュし、Spring Bootで消費。データ書き込みコードの変更は不要です。