Apache Pulsar PIP-187 深度解析：订阅 Backlog 精确分析 API（analyzeBacklog）的设计、实现与实战-平芜编程栈

消息队列
流处理
后端
微服务
消息路由

【免费下载链接】pulsar

Apache Pulsar - distributed pub-sub messaging system

项目地址：https://gitcode.com/gh_mirrors/pu/pulsar

导读

Apache Pulsar 传统的 backlog 统计只给出 "entries"（存储条目）数量，而消费者实际感受到的工作量与消息（messages）数量成正比——尤其在批处理（batching）与服务端过滤器（PIP-105 Subscription Filters）共同作用的场景下，entries 与 messages 严重不成比例，导致积压监控失真。PIP-187 为 Pulsar 引入了一套全新的 "订阅 backlog 精确分析" 能力：通过新增 ManagedCursor 扫描 API、Broker REST 接口、Java PulsarAdmin 方法与 pulsar-admin CLI 命令，从指定位置（默认最近未确认位置 lastMarkDeletePosition）开始逐条读取存储中的消息并施加订阅过滤器，最终返回 entries、messages、过滤器裁决结果等精确统计。本文将以 pip/pip-187.md 为骨架，结合仓库源码与测试用例，完整讲解该特性的 API 形态、配置参数、底层实现原理、客户端循环扫描机制与最佳实践。

问题背景：为什么传统 backlog 统计不够"精确"

PIP-187 提案明确指出，在 PIP-187 落地之前，Pulsar 无法为一个订阅提供精确的 backlog 数值，原因集中在两点：

只有 entries 数量，没有 messages 数量。Pulsar 底层存储（ManagedLedger）以 entry 为单位组织数据，而单个 entry 可以承载一条或多条消息。当生产者开启批处理（batching）时，一个 entry 内可能打包多条消息，消费者侧的实际处理工作量与消息条数成正比，与 entry 条数不成正比——仅靠 entry 数无法估算真实的消费压力。
服务端过滤器（PIP-105）可能过滤掉部分消息。PIP-105 引入了订阅级别的服务端过滤器（Subscription Filters），经过过滤器 REJECT/RESCHEDULE 的消息实际上不会投递给消费者，但它们依然占用存储和 entry 计数，导致传统统计口径下的 backlog 虚高。

此外，由于订阅具备极强的动态性（订阅可以从 earliest 位置创建、过滤器可以随时间变化），这些信息无法通过简单的计数器在写入路径上预先维护，唯一可靠的精确值只能通过"事后扫描"计算得出。这正是 analyzeBacklog 系列 API 存在的根本原因。

API 变更总览

PIP-187 的实现横跨了存储层、Broker 管理接口与客户端工具链四个层次：

层次	新增内容	说明
存储层（ManagedCursor）	`scan(Optional<Position>, Predicate<Entry>, ...)`	从指定位置扫描游标到尾部，带时间与条目上限
Broker REST	`POST .../subscription/{subName}/analyzeBacklog`	异步接口，返回精确分析结果
Java PulsarAdmin	`analyzeSubscriptionBacklog(...)`系列重载	同步/异步、支持客户端循环扫描
pulsar-admin CLI	`pulsar-admin topics analyze-backlog`	命令行入口，支持进度输出与 NDJSON

下面逐层展开。

存储层：ManagedCursor.scan API

PIP-187 在内部 ManagedCursor 接口中新增了扫描能力，原始提案给出的签名是：

CompletableFuture<ScanOutcome> scan(Optional<Position> startingPosition, Predicate<Entry> condition, long maxEntries, long timeOutMs);

在仓库中的最终实现位于 managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java，签名在此基础上增加了batchSize参数，用于控制每次从存储读取的条目数：

default CompletableFuture<ScanOutcome> scan(Optional<Position> startingPosition, Predicate<Entry> condition, int batchSize, long maxEntries, long timeOutMs) { return CompletableFuture.failedFuture(new UnsupportedOperationException()); }

关键语义如下：

起点：startingPosition未提供时，从lastMarkDeletePosition（最近一次 mark-delete 位置，即已确认消息的边界）开始；提供时从该位置开始。
终止条件：扫描受maxEntries（最大条目数）与timeOutMs（最大耗时）双重限制，目的是防止在积压巨大时产生"巨大且无用"的全量扫描，起到保护 Broker 的作用。
条件回调：Predicate<Entry>逐条检查 entry，返回false可提前终止扫描；返回的ScanOutcome枚举则向调用方传达本次扫描的最终状态（COMPLETED表示正常扫完，否则视为被中断/中止）。
成本警告：接口 Javadoc 明确标注 "this is an expensive operation"，因为它需要从存储（BookKeeper）读取每条消息并执行过滤逻辑。

Broker REST 接口

端点与请求形态

提案中的 REST 原型为 GET + QueryParam 传 position，但在仓库中的最终实现（pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java）演进为POST + RequestBody 传 position：

POST /admin/v2/persistent/{tenant}/{namespace}/{topic}/subscription/{subName}/analyzeBacklog

参数说明：

参数	位置	必填	说明
`tenant`/`namespace`/`topic`	Path	是	主题的完整定位信息，`topic`需 URL 编码
`subName`	Path	是	被分析的订阅名，需 URL 编码
`position`	RequestBody（`ResetCursorData`，含`ledgerId`、`entryId`）	否	扫描起点；不传则从`lastMarkDeletePosition`开始
`authoritative`	Query	否（默认 false）	是否为 leader broker 重定向到本 broker 的调用（内部使用）

服务端会先校验主题所有权（非 authoritative 场景下若本 Broker 不服务该 namespace，会返回 307 重定向到正确的 Broker），再校验订阅是否存在（不存在返回 404）以及调用方是否具备该主题的CONSUME权限（无权限返回 401/403）。

响应模型：AnalyzeSubscriptionBacklogResult

返回的 JSON 对象对应 pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/stats/AnalyzeSubscriptionBacklogResult.java，字段如下：

字段	类型	含义
`entries`	long	扫描到的存储条目总数
`messages`	long	扫描到的逻辑消息总数（含批处理展开）
`markerMessages`	long	扫描到的内部 marker 消息数（如事务标记，不计入待投递消息）
`filterAcceptedEntries`	long	被过滤器 ACCEPT 的条目数
`filterRejectedEntries`	long	被过滤器 REJECT 的条目数
`filterRescheduledEntries`	long	被过滤器 RESCHEDULE 的条目数
`filterAcceptedMessages`	long	被过滤器 ACCEPT 的消息数
`filterRejectedMessages`	long	被过滤器 REJECT 的消息数
`filterRescheduledMessages`	long	被过滤器 RESCHEDULE 的消息数
`aborted`	boolean	是否因内部限制（超时、条目上限）被中止
`firstMessageId`	String	扫描到的第一条消息 ID（`ledgerId:entryId`）
`lastMessageId`	String	扫描到的最后一条消息 ID（`ledgerId:entryId`）

其中aborted=true表示请求被某些内部限制（如超时或达到最大条目数）中止。PIP-187 明确说明：API 不会提供中止原因的更多细节，避免接口过于繁琐、难以维护；如需排查，Broker 日志中会有详细记录。firstMessageId/lastMessageId是最终实现相对提案的增强，它们为客户端循环扫描（见下文）提供了续扫锚点。

Java PulsarAdmin API

同步与异步入口

接口定义位于 pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java，实现位于 pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java。核心方法族如下：

// 同步：从指定起点（empty 表示最后已处理位置）开始分析 AnalyzeSubscriptionBacklogResult analyzeSubscriptionBacklog(String topic, String subscriptionName, Optional<MessageId> startPosition) throws PulsarAdminException; // 同步：额外指定客户端侧最大扫描条目数（用于客户端循环终止） AnalyzeSubscriptionBacklogResult analyzeSubscriptionBacklog(String topic, String subscriptionName, Optional<MessageId> startPosition, long backlogScanMaxEntries) throws PulsarAdminException; // 同步：自定义终止谓词 AnalyzeSubscriptionBacklogResult analyzeSubscriptionBacklog(String topic, String subscriptionName, Optional<MessageId> startPosition, Predicate<AnalyzeSubscriptionBacklogResult> terminatePredicate) throws PulsarAdminException; // 异步版本（三个重载一一对应） CompletableFuture<AnalyzeSubscriptionBacklogResult> analyzeSubscriptionBacklogAsync(String topic, String subscriptionName, Optional<MessageId> startPosition); CompletableFuture<AnalyzeSubscriptionBacklogResult> analyzeSubscriptionBacklogAsync(String topic, String subscriptionName, Optional<MessageId> startPosition, long backlogScanMaxEntries); CompletableFuture<AnalyzeSubscriptionBacklogResult> analyzeSubscriptionBacklogAsync(String topic, String subscriptionName, Optional<MessageId> startPosition, Predicate<AnalyzeSubscriptionBacklogResult> terminatePredicate);

topic必须是持久主题（persistent://...），startPosition为Optional.empty()时表示从订阅最后已处理位置（mark-delete 位置）开始扫描。所有方法均为"潜在昂贵操作"，Javadoc 反复强调其会从存储读取消息、并考虑批处理消息与订阅过滤器。

客户端循环扫描（Client-side Loop）机制

Broker 侧有subscriptionBacklogScanMaxEntries（默认 10000）与subscriptionBacklogScanMaxTimeMs（默认 2 分钟）两个硬上限，积压超过上限时单次请求必然aborted=true。为解决"想拿到完整精确积压"与"Broker 单次扫描有上限"的矛盾，客户端实现了自动循环续扫：

客户端先发起一次analyze-backlog请求；
若服务端返回aborted=true且未满足客户端终止条件，客户端取出响应中的lastMessageId，将其entryId + 1作为新的起点（TopicsImpl.java），再次发起请求；
循环中每次返回的统计通过mergeBacklogResults累加合并；
直到服务端返回aborted=false（扫描完整），或累计entries达到客户端设定的backlogScanMaxEntries，或满足自定义terminatePredicate；
防御性兜底：若某次返回entries <= 0或lastMessageId为空，视为扫描完成，终止循环（TopicsImpl.java）。

需要特别强调两点语义（源码 Javadoc 原话）：

若服务端subscriptionBacklogScanMaxEntries >= 客户端 backlogScanMaxEntries，则客户端参数不生效；
backlogScanMaxEntries并不能精确控制服务端扫描的条目数，它只决定客户端循环何时停止——实际扫描的条目数会是服务端单次上限的整数倍。

这一机制让运维人员可以在不调大 Broker 全局限制（避免单次请求长期占用线程与 IO）的前提下，通过客户端循环拿到任意深的精确积压。

pulsar-admin CLI 命令

对应命令定义在 pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java：

pulsar-admin topics analyze-backlog <persistent://tenant/namespace/topic> \ -s <subscription-name> \ [-p <ledgerId:entryId>] \ [-b <backlogScanMaxEntries>] \ [-q] \ [--plain]

选项	必填	默认	说明
`-s, --subscription`	是	—	待分析的订阅名
`-p, --position`	否	最近未确认位置	扫描起点，格式`ledgerId:entryId`，如`-p 5:100`；也支持`-1:-1`（从 earliest 起扫）
`-b, --backlog-scan-max-entries`	否	不限制	客户端循环终止阈值；必须大于 0，否则报参数错误
`-q, --quiet`	否	false	关闭循环扫描时的进度输出
`--plain`	否	false	以 NDJSON（每行一条 JSON）而非美化 JSON 输出结果

不指定-b时命令调用单次analyzeSubscriptionBacklog(topic, sub, startPosition)；指定-b时以result.getEntries() >= backlogScanMaxEntries为终止谓词调用循环版本，并且每次循环都会打印一次中间结果（除非--quiet）。最终结果统一通过print输出，--plain便于脚本用 jq 等工具解析。

Broker 配置项

两个新增配置定义在 pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java，并同步出现在 conf/broker.conf：

# Maximum time in ms for a Analise backlog operation to complete subscriptionBacklogScanMaxTimeMs=120000 # Maximum number of entries to be read within a Analise backlog operation subscriptionBacklogScanMaxEntries=10000

配置项	默认值	说明
`subscriptionBacklogScanMaxTimeMs`	`120000`（2 分钟）	单次服务端扫描允许的最大耗时，超时则返回`aborted=true`
`subscriptionBacklogScanMaxEntries`	`10000`	单次服务端扫描允许处理的最大条目数，达到即返回`aborted=true`

注意：调大这两个值可能让单次 HTTP 请求长时间挂起，甚至先于命令完成前触发 HTTP 请求超时（以及 NAT/防火墙的 idle timeout），因此源码 Javadoc 给出的建议是尽量保持 Broker 侧限制收敛，需要深扫时依靠客户端循环（backlogScanMaxEntries）来获取更多条目。此外扫描时每次读取的批大小复用dispatcherMaxReadBatchSize配置（见下文实现细节）。

实现原理：扫描过程在 Broker 内部如何运作

核心执行路径

PersistentSubscription.analyzeBacklog是整条链路的枢纽，位于 pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java，调用链如下：

PersistentTopics.analyzeSubscriptionBacklog (REST) → PersistentTopicsBase.internalAnalyzeSubscriptionBacklogForNonPartitionedTopic → Subscription.analyzeBacklog(Optional<Position>) → PersistentSubscription.analyzeBacklog → cursor.duplicateNonDurableCursor(...) // 创建临时非持久游标 → newNonDurableCursor.scan(position, condition, batchSize, maxEntries, timeOutMs) → managedLedger.asyncDeleteCursor(...) // 扫描结束清理临时游标

实现要点：

临时非持久游标：扫描基于cursor.duplicateNonDurableCursor("analyze-backlog-" + UUID)创建的副本游标，避免干扰订阅的真实消费游标；扫描完成后异步删除该临时游标（asyncDeleteCursor），不留垃圾。
从 mark-delete 位置起扫：游标副本的getMarkDeletedPosition()即"最后已处理位置"，未指定起点时从该处开始。
逐条读取并解析元数据：对每个 entry，若内存中无元数据则通过Commands.peekMessageMetadata解析；有markerType的消息（内部 marker，如事务标记）单独计入markerMessages。
批处理展开：若messageMetadata.hasNumMessagesInBatch()，则取getNumMessagesInBatch()作为该 entry 包含的逻辑消息数，用于累计messages及各类 filter 消息计数。
过滤器裁决：通过entryFilterSupport.runFiltersForEntry(entry, messageMetadata, null)执行订阅过滤器（EntryFilterSupport来自 dispatcher 或新建实例），结果按EntryFilter.FilterResult分为ACCEPT/REJECT/RESCHEDULE三档分别计数，对应响应模型中的三组 accepted/rejected/rescheduled 统计。
硬性保护限制：maxEntries与timeOutMs直接取自上述两个配置，目的是"防止拒绝服务"——这与 PIP-187 中"prevent huge (and useless) scans"的初衷一致。
进度日志：每处理 1000 条 entry 打印一条Scan running日志（含已耗时与条数），结束后打印Scan complete（含总耗时与结果），供运维在 Broker 日志中定位中止原因。

分区主题与不支持场景

分区主题不可直接分析：测试 AnalyzeBacklogSubscriptionTest.partitionedTopicNotAllowed 验证了对分区主题调用会抛出PulsarAdminException.NotAllowedException，但可以分别对topic-partition-N单分区执行分析。
非持久主题不支持：NonPersistentSubscription.analyzeBacklog 直接抛出UnsupportedOperationException("Unsupported operation analyzeBacklog for NonPersistentSubscription")——该特性只针对持久订阅。

与 individuallyDeletedMessages 的交互

PIP-187 实现说明中专门提到"它考虑 individuallyDeletedMessages"（单条确认的消息）。scan基于游标副本扫描，而游标的individuallyDeletedMessages（单条删除范围）会被扫描逻辑跳过，因此已被单独确认的消息不会计入 backlog。测试 analyzeBacklogWithIndividualAck 与simpleAnalyzeBacklogTest（L59-L151）对这一行为做了验证，例如非批处理场景下单条 ack 后entries与messages各减 1；批处理场景下 ack 单条消息不减少 entry 计数（entry 仍存在），这与底层存储语义一致。

测试验证：从源码看行为保证

仓库中的核心测试文件是 pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AnalyzeBacklogSubscriptionTest.java，覆盖了以下关键场景：

测试方法	验证点
`simpleAnalyzeBacklogTest`/`simpleAnalyzeBacklogTestWithBatching`	无批处理与批处理（batchSize=5）下 entries/messages 的精确计数；消费 ack 后统计随之减少
`partitionedTopicNotAllowed`	分区主题整体不可用，单分区可用
`analyzeBacklogServerReturnFalseAbortedFlagWithoutLoop`	积压小于服务端上限时`aborted=false`
`analyzeBacklogMaxEntriesExceedWithoutLoop`	积压超过服务端上限时返回`aborted=true`，只统计到上限
`analyzeBacklogServerReturnFalseAbortedFlagWithLoop`	客户端循环下可拿到完整积压
`analyzeBacklogMaxEntriesExceedWithLoop`	客户端`backlogScanMaxEntries`控制循环终止；Broker 单次 15 条、循环 3 次共返回 45 条（客户端阈值 40）
`analyzeBacklogWithTopicUnload`	扫描期间主题被 unload/重新加载后循环续扫仍正确
`analyzeBacklogWithIndividualAck`	单条 ack（individuallyDeletedMessages）场景下计数正确

此外，tests/integration/src/test/java/org/apache/pulsar/tests/integration/cli/topic/AnalyzeBacklogTest.java 还提供了针对pulsar-admin topics analyze-backlog命令的集成测试。

为何放弃"旁路计数"方案：Rejected Alternatives

PIP-187 在"拒绝的备选方案"一节中解释了为什么不能通过更廉价的方式获得精确值，这对理解该 API 的定位至关重要：

写入时维护逻辑消息计数器。不可行的原因有四：
- 写入时无法预知未来会有哪些订阅（订阅在写入之后才可能创建）；
- 订阅可以从过去（Earliest）创建，计数器无法回溯覆盖；
- 订阅过滤器通常配置在 Subscription Properties 上、是动态可变的，计数器无法跟随过滤器变化；
- 在写入路径上执行过滤器计算会拖垮写入延迟与吞吐。
用客户端克隆订阅并消费数据来计算。同样不可行：这要求把海量积压数据从 Broker 传输到客户端，工作量大且严重浪费带宽与资源。

这两条理由也反向解释了为什么该功能必须是"按需、扫描式、读路径"的昂贵操作——这是权衡所有方案后唯一既能保证精确、又不对写入路径造成影响的可行路径。

实践建议：如何用它做监控与告警

PIP-187 给出的监控策略非常明确，可以作为生产环境落地该 API 的准则：

常规告警用 stats：日常监控、自动告警应继续基于常规的订阅统计（backlog entries 等）配置阈值，保持低开销；
精确分析走 analyzeBacklog：当常规统计出现异常（积压疑似超标、怀疑过滤器误过滤、需要评估消费延迟的真实工作量）时，通过pulsar-admin topics analyze-backlog或 Java API 发起一次精确分析，作为人工介入的深度排查手段；
控制扫描成本：由于该操作昂贵，建议使用-b客户端循环限制或依赖 Broker 默认上限，避免单次请求扫描过深；对超大积压可先观察aborted标志与lastMessageId判断扫描是否完整；
配合过滤器排查：对比filterRejectedMessages与filterAcceptedMessages可快速判断积压中多少消息实际不会投递，从而校准真实消费压力；
注意适用边界：仅持久主题、非分区主题（或单分区）可用；Broker 配置subscriptionBacklogScanMaxTimeMs/subscriptionBacklogScanMaxEntries控制单次扫描的硬上限，调整时需权衡 HTTP 请求超时风险。

小结

PIP-187 通过"按需扫描 + 服务端过滤 + 批处理展开"的组合，为 Apache Pulsar 提供了唯一准确的订阅 backlog 视图：ManagedCursor.scan提供存储层扫描原语，Broker REST 与 Java PulsarAdmin 暴露分析入口，客户端循环机制突破单次扫描上限，CLI 命令让运维人员一条命令即可洞察积压真相。理解其"昂贵但必要"的设计取舍，以及 entries/messages/过滤器裁决等多维统计口径，是正确使用这一能力进行监控、告警与容量评估的关键。相关实现与测试均可直接在本仓库对应路径中查阅，便于深入研读。