《Java 消息中间件·可靠消息与消费治理》5.1 lag 的三种口径与误读

三个数都不是 bug

先看一组在同一时刻、同一个分区上取到的数：

如果你在值班时同时打开监控面板、客户端日志和一段临时脚本，看到的就是这种场面。三个数都来自 Kafka，都不是脏数据，却指向三个不同的问题。把它们混着用，轻则误判「还能追上」，重则按错误的基数扩容。

本篇只做一件事：把 lag 的三种口径定义清楚，用一次真实取数把差异摆出来，再说清每种口径该用在哪。

实验：制造一个「处理了但没提交」的中间态

环境：Temurin 25.0.5+8、kafka-clients 4.3.0、apache/kafka:4.3.0 单节点 KRaft，2026-10-29。以下数字均为单次运行，随调度波动，不作基准。

步骤：

1. 建单分区主题 library.loan.created.lag，写入 1000 条记录，LEO=1000。
2. 消费者组 libraryd-lag-probe，enable.auto.commit=false，max.poll.records=100，从最早开始读。
3. 处理 400 条，但只在处理到第 200 条时 commitSync() 一次。
4. 停下来，不提交、不轮询，保持 70 秒，期间用 CLI 和客户端指标各取一次数。

关键代码就三行：

ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
for (ConsumerRecord<String, String> ignored : records) {
    processed++;
}
if (processed >= 200 && committed < 200) {
    consumer.commitSync();
}

处理到 400 条时循环退出，此刻的状态是：position=400、committed=200、LEO=1000。三个口径的分歧全部由这三个数决定。

口径一：LEO 减去 committed

这是监控面板和 --describe 用的口径。CLI 原始输出：

GROUP              TOPIC                    PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID       HOST         CLIENT-ID
libraryd-lag-probe library.loan.created.lag 0          200             1000            800  lag-probe-7fbfaf0f /172.17.0.1 lag-probe

同一时刻用 kafka-get-offsets.sh 取分区末尾，确认 LEO：

library.loan.created.lag:0:1000

公式：

lag_committed = LEO - committed = 1000 - 200 = 800

它的语义是「这个组还没有承诺处理完多少条」。committed 是消费组对外的承诺：一旦提交，这条位移之前的记录在 Kafka 的语义里就算交付了。哪怕进程下一秒崩溃，重启后也从 200 之后开始读，前 200 条不会再投递。

这个口径有三个特点：

• 它滞后于实际处理进度。处理了 400 条只提交 200 条，口径一看到的是 800，而真实未处理的是 600。提交越稀疏，偏差越大。
• 它是锯齿状的。批量提交时，每次提交前 lag 冲高，提交后断崖下跌。用瞬时值做告警会误报，通常要看一段时间的均值或「连续 N 个采样点都高于阈值」。
• 它是跨实例可加的。每个分区的 LEO 减去该分区被提交的位移，求和就是整个组的积压。多实例、多分区都能直接相加，这也是监控系统偏爱它的原因。

还有一个小细节：当组里没有提交记录时，--describe 的 CURRENT-OFFSET 和 LAG 显示 -，不是 0。实验中一个刚启动、还没提交过的组输出如下：

GROUP                TOPIC                    PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
libraryd-backlog-d   library.loan.created.backlog 0      -               30784           -

把 - 当 0 处理，会得出「没有积压」的相反结论。解析脚本必须先判断字段是否为数字。

口径二：LEO 减去 position

position() 是消费者「下一条要 fetch 的位移」。手动取数：

TopicPartition tp = new TopicPartition(topic, 0);
long position = consumer.position(tp);
long leo = consumer.endOffsets(List.of(tp)).get(tp);
System.out.printf("leo=%d position=%d manual=%d%n", leo, position, leo - position);

同一时刻输出：

[rdv] leo=1000 position=400 committed=200 processed=400 manual(leo-position)=600 cli(leo-committed)=800

lag_position = 1000 - 400 = 600。

它回答的是「客户端还要从 broker 拉多少条」。注意用词是「拉」，不是「处理」。因为 position 前进的时机不是业务处理完，而是 poll() 把记录交出去的时候。4.3.0 的 FetchCollector 在返回记录之前就推进了 position：

if (nextInLineFetch.nextFetchOffset() > position.offset) {
    SubscriptionState.FetchPosition nextPosition = new SubscriptionState.FetchPosition(
            nextInLineFetch.nextFetchOffset(),
            nextInLineFetch.lastEpoch(),
            position.currentLeader);
    log.trace("Updating fetch position from {} to {} for partition {} and returning {} records from `poll()`",
            position, nextPosition, tp, partRecords.size());
    subscriptions.position(tp, nextPosition);
    positionAdvanced = true;
}

Long partitionLag = subscriptions.partitionLag(tp, fetchConfig.isolationLevel);
if (partitionLag != null)
    metricsManager.recordPartitionLag(tp, partitionLag);

这段代码说明两件事：position 在 poll() 返回前就跳到了这批记录之后；客户端指标是在 position 已经前进之后才计算的。所以手上正在处理的那一批，不计入客户端的 lag。

用一个更直接的实验验证。新组、max.poll.records=500，只 poll() 一次拿到 500 条，然后一条都不处理，静置 5 秒：

afterPoll batch=500 processed=0 position=500 leo=1000 leoMinusPosition=500
metric name=records-lag value=500.0
sleeping 5000ms holding the batch, no processing, no poll
sleepDone batchStillHeld=500 processed=0 position=500

processed=0，但 position 已经是 500，lag 只剩 500。如果下游卡住、消费者拿着 500 条不动，这段时间里口径二会稳定显示 500，看不出「这 500 条一条都没处理」。要判断处理进度，必须用 committed，或者自己统计已处理计数。

口径三：客户端指标

同一个消费者暴露的指标里，名字带 records-lag 的有三个：records-lag、records-lag-avg、records-lag-max。它们的计算入口是 SubscriptionState.partitionLag：

public synchronized Long partitionLag(TopicPartition tp, IsolationLevel isolationLevel) {
    TopicPartitionState topicPartitionState = assignedState(tp);
    if (topicPartitionState.position == null) {
        return null;
    } else if (isolationLevel == IsolationLevel.READ_COMMITTED) {
        return topicPartitionState.lastStableOffset == null ? null : topicPartitionState.lastStableOffset - topicPartitionState.position.offset;
    } else {
        return topicPartitionState.highWatermark == null ? null : topicPartitionState.highWatermark - topicPartitionState.position.offset;
    }
}

注意减的是 highWatermark（高水位），不是 LEO。单 broker 时两者相等；多副本且 follower 落后时，HW 小于 LEO，客户端指标会比 LEO - position 更小。本篇是单节点实验，没有测到 HW 与 LEO 分离的场景，这一点只作源码层面的说明，不算实测。

records-lag-max 的官方描述只有一句，但信息量很大：

The maximum lag in terms of number of records for any partition in this window. NOTE: This is based on current offset and not committed offset

「in this window」是关键。它是一个窗口内最大值，不是当前值。实验结果：

[rdv] metric group=consumer-fetch-manager-metrics name=records-lag value=600.0
[rdv] metric group=consumer-fetch-manager-metrics name=records-lag-avg value=750.0
[rdv] metric group=consumer-fetch-manager-metrics name=records-lag-max value=900.0

records-lag 是当前值 600；records-lag-max 是 900，因为消费者在 position=100 时记录过一次 900，这个最大值还留在采样窗口里。900 既不是当前 lag，也不是 committed 口径的 800，它是「过去几十秒里最坏的那个瞬间」。

窗口过期后更麻烦。上面那个消费者静置 70 秒不轮询，再读一次：

[settled] leo=1000 position=400 committed=200 processed=400 manual(leo-position)=600 cli(leo-committed)=800
[settled] metric name=records-lag value=600.0
[settled] metric name=records-lag-avg value=NaN(窗口内无样本)
[settled] metric name=records-lag-max value=NaN(窗口内无样本)

records-lag 保留最后值 600；avg 和 max 因为窗口里没有新样本，变成 NaN。很多监控 SDK 会把 NaN 丢弃或填 0，面板上就出现「lag 归零」的假象，而 committed 口径明明还有 800 条没处理。这是「面板和代码对不上」最常见的一种成因：不是数据错，是指标停止采样后过期了。

三种口径放在一张表里

三行数据对应同一个事实：处理了 400 条，只承诺了 200 条，还有 600 条没拉。同一个事实的三种写法，对应三种完全不同的运维动作。

为什么监控面板和代码里不一样

把差异逐条归因，一共七种，前三种在本篇都实测到了：

1. 提交频率。 committed 落后于 position，口径一比口径二大。批量提交时这个差值最大可以接近一个提交批次。
2. position 在 poll 时前进。 客户端指标和口径二都不算手上那批。下游越慢、max.poll.records 越大，少算的越多。
3. 窗口统计会过期。 records-lag-max 是窗口最大值，空转后变 NaN；records-lag 是最后值，会冻结。两者都不适合当「当前积压」。
4. HW 与 LEO 分离。 多副本时客户端用 HW，CLI 用 LEO。follower 落后期间客户端指标偏小，本篇未实测，只作源码说明。
5. read_committed 用 LSO。 有长事务打开时，endOffsets 和客户端指标都停在 LSO，而 --describe 的 LOG-END-OFFSET 仍是 LEO。事务场景下这三个数会差出一整段被 LSO 挡住的消息。
6. 多分区聚合方式不同。 面板通常对分区求和，records-lag-max 只取最大值。10 个分区各积压 100 条，max 显示 100，sum 是 1000。
7. 组状态不同。 组不存在或在再平衡时，CLI 的 LAG 可能是 -；客户端指标可能还没开始记录。把「没数」当「没积压」是事故的常见起点。

代码里取数要注意的版本细节

4.3.0 的 KafkaConsumer 移除了单分区重载 committed(TopicPartition)，只剩集合版本。照着旧书写会直接编译失败：

错误: 不兼容的类型: TopicPartition 无法转换为 Set<TopicPartition>

正确写法：

Map<TopicPartition, OffsetAndMetadata> committed =
        consumer.committed(Set.of(tp));
OffsetAndMetadata om = committed.get(tp);
long committedOffset = om == null ? -1 : om.offset();

endOffsets 返回的是 LEO，不是 HW，也不是 LSO。要拿 LSO 只能靠 read_committed 消费者内部的 lastStableOffset，公开 API 没有直接入口。这也是为什么「自己算 lag」在事务场景下容易算错。

读指标时，consumer.metrics() 的 key 是 MetricName，要按 name() 过滤，不能按 toString() 匹配：

consumer.metrics().entrySet().stream()
        .filter(e -> e.getKey().name().equals("records-lag-max"))
        .forEach(e -> System.out.println(e.getValue().metricValue()));

同一个 records-lag 会出现两次：一次是组级汇总，一次是带 topic/partition 标签的分区级。按分区诊断时要看后者，别把汇总值当成分区值。

取数脚本的四个坑

如果你要自己写一个 lag 采集脚本，下面四条都是本机踩过的：

• --describe 很慢。 每次调用要新起一个 JVM 加 AdminClient，本机实测单次 3.6 秒。用 shell 循环每秒采样是不现实的，实际分辨率会掉到 4 到 6 秒。要秒级曲线，就在一个常驻 JVM 里用 AdminClient 循环取数，别反复起进程。
• - 不是 0。 组没有提交记录时 LAG 是 -。脚本里 awk '$6 ~ /^[0-9]+$/' 过滤掉非数字，比 int($6) 安全。
• NaN 不是 0。 客户端指标空转后会变 NaN。采集端要么丢弃、要么显式标记，绝不能填 0。
• 求和与取最大要分清。 告警要跨分区求和；records-lag-max 只代表最坏的那个分区。用 max 做容量估算会低估总量，用 sum 做单分区诊断会掩盖热点。

这四条不影响 Kafka 的正确性，但足够让面板和代码对不上。先把采集链路修对，再谈阈值。

每种口径该用在哪

一个实用的组合是：告警用 committed 口径，追积压的执行决策用 position 口径，客户端指标只用于排查单个实例「为什么拉不动」。三种口径各司其职，不要试图让一个数回答所有问题。

回到 libraryd

libraryd 的 library.loan.created 主题按 bookId 分区，消费组负责更新借阅投影。值班时看到面板 lag 上涨，先分清是哪一种：

• 面板涨、客户端 records-lag 不涨：消费者在拉但提交慢，先看提交频率和下游事务耗时；
• 面板和客户端都涨：真的拉不动，看消费速率和分区数，进入 5.2 的处置流程；
• 面板涨、客户端指标是 NaN：消费者可能已经停止轮询，先确认进程活着、max.poll.interval.ms 没超；
• 只有 records-lag-max 高：大概率是历史峰值还在窗口里，等一个窗口再看。

下一节把「数字大」翻译成「能不能追上」，用三条处置路径做一次完整演练。

阅读导航：上一节 4.3 扩缩容与优雅停机 ｜ 下一节 5.2 积压处置：扩容、跳过与降级