Apache Pulsar Transaction Buffer 多快照机制解析：基于 PIP-178 的设计与源码实现

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

【免费下载链接】pulsar

Apache Pulsar - distributed pub-sub messaging system

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

点击查看免费下载

导读

Transaction Buffer（事务缓冲区）是 Apache Pulsar 实现事务消息的核心组件，它需要持久化"已中止事务 ID（aborted transaction IDs）"，以便在消息消费时过滤掉被中止事务写入的消息。本文以 PIP-178: Multiple snapshots for transaction buffer 为核心骨架，完整讲解该提案面临的技术挑战、多快照（multiple-snapshot）的键设计、快照写入与恢复流程，并结合当前仓库中pulsar-broker模块的真实实现（单快照处理器与分段快照处理器）进行源码级佐证。读完本文，你将理解：为什么 aborted transaction ID 数量不可控时会撑爆单个 BookKeeper entry；PIP-178 如何用"多 entry + 有序键 + maxReadPosition 语义"解决压缩覆盖与部分写入问题；以及该提案最终在 Pulsar 中以transactionBufferSegmentedSnapshotEnabled开关落地为 v2 分段快照架构的演进过程。

一、背景：为什么需要多快照（Multiple Snapshots）

Pulsar 中每条 Topic 都维护一个 Transaction Buffer，用于记录该 Topic 上的中止事务 ID。当消费者读取消息时，Transaction Buffer 会利用这份"中止列表"过滤掉属于中止事务的消息。为了在 Broker 重启后能够恢复，Transaction Buffer 会定期把 aborted transaction IDs 写入快照（snapshot），持久化到 BookKeeper 中。

问题在于：aborted transaction IDs 的数量是不受限制的。在长生命周期、高中止率的 Topic 上，中止事务 ID 集合会持续增长。而一个 BookKeeper entry 能够容纳的数据量是有限的，当 aborted transaction IDs 的总大小超过单个 entry 的容量时，Transaction Buffer 就必须把一份快照拆成多条快照（multiple-snapshot），分别写入多个 entry。

PIP-178 正是为了解决"单 entry 放不下"这一现实约束而提出的多快照方案。核心思路可以概括为：用带有序语义的多个键（key）写入多个快照 entry，并把 maxReadPosition 只记录在最后一个快照中。

写入键设计（提案原始方案）： topicName-1, topicName-2, ..., topicName-end

其中前 N-1 个快照只承载 aborted transaction IDs（maxReadPosition取最早值），最后一个以topicName-end为键的快照才承载真正的maxReadPosition。这样既突破了单 entry 容量限制，又保证恢复时总能找到一个携带正确maxReadPosition的快照。

二、挑战：两个必须解决的现实问题

提案原文明确指出，由于"压缩（compression）"和"写入不完整（incomplete sending）"的存在，实现多快照面临两大挑战：

2.1 Broker 重启导致只写入了部分快照

Transaction Buffer 需要把多快照 (1, 2, 3) 依次写入。如果只写完了快照 1、2，Broker 就发生重启，快照 3 永远不会被写入。此时恢复逻辑必须能容忍"快照集合不完整"的状态，仍然可以正确还原 aborted transaction IDs，而不是因为缺失最后一个快照而报错或丢数据。

2.2 压缩导致新快照覆盖同键旧快照

Pulsar 的 System Topic / compacted topic 具有压缩（compaction）语义：以相同 key 写入的新消息会覆盖旧消息。当多快照 (1, 2, 3) 因重试等原因被第二次写入时，可能出现"快照 1、2 是第二次写入的新版本，而快照 3 仍是第一次写入的旧版本"的错配状态。恢复时如果按 key 读到新旧混合的快照集合，就可能把错误的 abort 数据与错误的 maxReadPosition 组合在一起。

这两个挑战共同决定了：多快照方案必须保证"任意部分写入 + 任意覆盖"状态下，恢复结果依然一致且正确。PIP-178 通过"按 aborted marker 位置排序 + 只保留最大 maxReadPosition"给出了答案。

三、核心方案（Approach）

PIP-178 的实施方案分为三步：

1. 将aborts的数据结构从LinkedMap改为ConcurrentSkipListMap，使 aborted transaction IDs 天然按 aborted marker 的持久化位置（ledger/entry 位置）有序。提案的 Goal 之一即是"aborted transaction IDs 按 aborted marker 位置排序，从而让 aborts 成为一个 FIFO map"。
2. 多快照使用键topicName-1, topicName-2, ..., topicName-end写入，普通（单）快照使用键topicName-end写入。即键中携带序号，末尾哨兵键topicName-end是快照集合的收尾标记。
3. 只有键为topicName-end的快照才存储真正的maxReadPosition，其余快照的maxReadPosition取最早值（earliest）。这是保证"始终存在一个带正确 maxReadPosition 的快照用于恢复"的关键。

三个目标相互支撑：

• Goal 1：aborted transaction IDs 按 aborted marker 位置有序 → aborts 变成 FIFO 队列，便于按序清理。
• Goal 2：新快照覆盖旧快照不会造成错误 → 因为同键快照的 abort 集合在有序化后内容一致，覆盖是幂等的。
• Goal 3：始终存在一个携带正确 maxReadPosition 的快照 → 无论部分写入还是压缩覆盖，恢复都有可靠锚点。

四、三种运行场景推演（Examples）

4.1 正常流程（Normal Flow）

第一个快照在新建 Producer 发送消息时触发，因此必然存在一个键为topicName-end、携带 maxReadPosition 的快照可供恢复。正常流程下多快照完整写入，恢复时按序合并所有 entry 即可得到完整 abort 集合与最新的 maxReadPosition。

4.2 写入不完整（Write incompletely）

当 aborted transaction IDs 已按 aborted marker 位置排序、且尚未从aborts中删除时，同一个键对应的快照中存储的 txn IDs 是相同的（键为topicName-end的快照除外）。这意味着：即使某个中间快照未被写入，或者旧快照被新快照覆盖，恢复出的 abort 集合依然一致——因为有序集合的内容由"已存在的中止事务"唯一确定，与"被哪一次写入"无关。

4.3 写入不完整 + 因 ledger 删除导致部分 txn 被移除

Pulsar 中若中止事务所在的 ledger 已被删除（过期清理），对应的事务 ID 也会从aborts中移除。由于删除也是按 aborted marker 位置从小到大依次进行的，新快照压缩时不会丢失任何有效消息。例如：txn0~25 所在 ledger 被删除后，txn0~25 从 aborts 中移除，但这对快照中已持久化的信息没有任何影响——恢复时仍能找到一个有效的 maxReadPosition。

这四条场景共同验证了方案的幂等性与可恢复性：乱序覆盖、缺失中间快照、过期清理，都不会破坏最终恢复结果。

五、代码实现（Code Implement）

5.1 恢复端：handleSnapshot

提案给出了恢复阶段的处理逻辑：读取快照后，仅当快照中的 maxReadPosition 大于当前值时更新（取最大）；然后把快照中的 aborted transaction IDs 逐个放入abortsmap：

public void handleSnapshot(TransactionBufferSnapshot snapshot) { PositionImpl newMaxReadPosition = PositionImpl.get(snapshot.getMaxReadPositionLedgerId(), snapshot.getMaxReadPositionEntryId()); if (newMaxReadPosition.compareTo(maxReadPosition) > 0) { maxReadPosition = newMaxReadPosition; } if (snapshot.getAborts() != null) { snapshot.getAborts().forEach(abortTxnMetadata -> aborts.put(new TxnID(abortTxnMetadata.getTxnIdMostBits(), abortTxnMetadata.getTxnIdLeastBits()), PositionImpl.get(abortTxnMetadata.getLedgerId(), abortTxnMetadata.getEntryId()))); } }

注意newMaxReadPosition.compareTo(maxReadPosition) > 0这个条件：多个快照中只有键为topicName-end的快照携带真实 maxReadPosition，取最大值可以天然跳过中间快照的 earliest 值，保证恢复锚点不后退。这与提案中"Always there is a snapshot with the right maxReadPosition to recover"的目标一一对应。

5.2 写入端：takeSnapshot

提案给出的写入逻辑把超大 abort 集合按maxSize分批，每一批生成一个TransactionBufferSnapshot，最后一批携带 maxReadPosition，全部通过writer.writeAsync异步写入并等待完成：

private CompletableFuture<Void> takeSnapshot() { changeMaxReadPositionAndAddAbortTimes.set(0); return takeSnapshotWriter.thenCompose(writer -> { TransactionBufferSnapshot snapshot = new TransactionBufferSnapshot(); ArrayList<TransactionBufferSnapshot> snapshots = new ArrayList<>(); synchronized (TopicTransactionBuffer.this) { List<AbortTxnMetadata> list = new ArrayList<>(); aborts.forEach((k, v) -> { AbortTxnMetadata abortTxnMetadata = new AbortTxnMetadata(); abortTxnMetadata.setTxnIdMostBits(k.getMostSigBits()); abortTxnMetadata.setTxnIdLeastBits(k.getLeastSigBits()); abortTxnMetadata.setLedgerId(v.getLedgerId()); abortTxnMetadata.setEntryId(v.getEntryId()); list.add(abortTxnMetadata); }); while (list.size() > maxSize) { List<AbortTxnMetadata> newList = new ArrayList<>(); while (newList.size() < maxSize) { newList.add(list.remove(0)); } snapshot.setAborts(newList); snapshot.setTopicName(topic.getName()); snapshots.add(snapshot); snapshot = new TransactionBufferSnapshot(); } snapshot.setMaxReadPositionLedgerId(maxReadPosition.getLedgerId()); snapshot.setMaxReadPositionEntryId(maxReadPosition.getEntryId()); snapshot.setAborts(list); snapshot.setTopicName(topic.getName()); snapshots.add(snapshot); } List<CompletableFuture<Void>> completableFutures = new LinkedList<>(); snapshots.forEach(snapshot_ -> { completableFutures.add(writer.writeAsync(snapshot_).thenAccept(messageId -> { this.lastSnapshotTimestamps = System.currentTimeMillis(); if (log.isDebugEnabled()) { log.debug("[{}]Transaction buffer take snapshot success! " + "messageId : {}", topic.getName(), messageId); } }).exceptionally(e -> { log.warn("[{}]Transaction buffer take snapshot fail! ", topic.getName(), e); return null; })); }); return FutureUtil.waitForAll(completableFutures); }); }

这段代码完整体现了"拆分（分批）+ 锚点（最后一批携带 maxReadPosition）+ 容错（exceptionally 只告警不阻塞）+ 汇合（waitForAll）"四个要点，是理解多快照写入语义的最佳入口。

5.3 快照数据结构：TransactionBufferSnapshot

提案中多快照承载的数据结构在仓库中已有落地的元数据类，见 TransactionBufferSnapshot.java，其字段与提案完全一致：

public class TransactionBufferSnapshot { private String topicName; private long maxReadPositionLedgerId; private long maxReadPositionEntryId; private List<AbortTxnMetadata> aborts; }

• topicName：快照所属 Topic；
• maxReadPositionLedgerId/maxReadPositionEntryId：恢复锚点（仅末尾快照携带真实值）；
• aborts：一组AbortTxnMetadata（包含txnIdMostBits、txnIdLeastBits、ledgerId、entryId），记录中止事务 ID 及其 aborted marker 的持久化位置。

六、仓库中的最终落地：从多快照到分段快照（v2）

PIP-178 提出的多快照思想在后续演进中发展为分段快照（Snapshot Segment）架构。这一点可以从仓库源码结构得到印证：pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/metadata/v2/目录下新增了一整套 v2 元数据类：

• TransactionBufferSnapshotSegment.java：一个快照段，包含topicName、sequenceId（段序号）、persistentPositionLedgerId/EntryId（段内最后中止标记的持久化位置）与aborts列表；
• TransactionBufferSnapshotIndex.java：段索引，记录sequenceID、abortedMarkLedgerID/EntryID、segmentLedgerID/EntryID；
• TransactionBufferSnapshotIndexesMetadata.java：携带maxReadPosition与未封段事务列表的元数据（对应原方案的"末尾快照"角色）；
• TransactionBufferSnapshotIndexes.java：由indexList（所有段索引）与snapshot（元数据）组成的总索引。

6.1 处理器选择：单快照 vs 分段快照

在 TopicTransactionBuffer.java 中，通过配置开关选择两种处理器实现：

private static AbortedTxnProcessor createSnapshotProcessor(PersistentTopic topic) { return topic.getBrokerService().getPulsar().getConfiguration().isTransactionBufferSegmentedSnapshotEnabled() ? new SnapshotSegmentAbortedTxnProcessorImpl(topic) : new SingleSnapshotAbortedTxnProcessorImpl(topic); }

• SingleSnapshotAbortedTxnProcessorImpl：即提案中"普通单快照"路径，把整个 abort 集合写进一个快照（见其takeAbortedTxnsSnapshot与handleSnapshot实现，SingleSnapshotAbortedTxnProcessorImpl.java）；
• SnapshotSegmentAbortedTxnProcessorImpl：即多快照/分段路径，默认关闭。

6.2 分段快照处理器核心实现

SnapshotSegmentAbortedTxnProcessorImpl.java 完整实现了提案的多快照思想，并做了工程化演进：

（1）段容量按配置动态计算。构造函数中根据transactionBufferSnapshotSegmentSize配置减去头部开销（空段 5 字节、空链表 +1、topic 名7 + topicName.length()等）后除以每条 abort 记录 3 字节的增量，得出每个段可容纳的 abort 数量：

this.snapshotSegmentCapacity = (topic.getBrokerService().getPulsar() .getConfiguration().getTransactionBufferSnapshotSegmentSize() - 8 - topic.getName().length()) / 3;

（2）aborts 采用有序 LinkedMap 并按段封存。putAbortedTxnAndPosition在累积数量达到段容量时，把当前段unsealedTxnIds交给PersistentWorker异步写入（L169-L184），这正是提案"多个 entry 存储多个快照"的落地形式。

（3）键设计演进。提案原方案使用topicName-1...topicName-end；当前实现中段写入键为SNAPSHOT_PREFIX + sequenceId + "-" + topicName，即multiple-<seq>-<topic>（buildKey 方法），索引写入键为topic.getName()。从源码结构看，键中引入了递增sequenceId，从而在压缩（compaction）语义下让新段覆盖旧段时保持幂等。

（4）索引 + 元数据双写保证锚点。PersistentWorker.updateSnapshotIndex每次把最新的indexList与携带maxReadPosition的元数据一并写入索引系统 topic（L752-L770），恢复时doRecoverFromSnapshot先读最新索引、再按索引回放所有段（readSegmentEntries），即"始终存在带正确 maxReadPosition 的快照可恢复"的工程实现。

（5）过期清理按序进行。trimExpiredAbortedTxns利用segmentIndex（段内最后中止标记位置 → 段内最后事务 ID）判断段是否过期，一旦 ledger 不存在，就按段从前到后批量删除 abort 与段文件（L198-L220），完全符合提案"按 aborted marker 位置顺序删除、压缩不丢消息"的设计目标。

6.3 相关配置项

分段快照架构对应以下配置项，均已在 ServiceConfiguration.java 中定义，并可在 conf/broker.conf 中查看默认值：

配置项	默认值	说明
transactionBufferSegmentedSnapshotEnabled	false	是否启用分段快照（即多快照/多 entry 方案）处理大量中止事务
transactionBufferSnapshotSegmentSize	262144（256 KB）	单个快照段的字节上限，决定段容量
transactionBufferSnapshotMaxTransactionCount	1000	启用分段快照后，事务操作数达到该值即更新快照元数据
transactionBufferSnapshotMinTimeInMillis	5000	快照间隔；分段快照模式下为更新元数据的间隔

启用示例（修改conf/broker.conf）：

transactionBufferSegmentedSnapshotEnabled=true transactionBufferSnapshotSegmentSize=262144 transactionBufferSnapshotMaxTransactionCount=1000 transactionBufferSnapshotMinTimeInMillis=5000

配置的默认值是false，说明分段快照是面向"aborted transaction ID 规模超过单 entry 容量"的高压力场景的可选能力，普通场景仍走单快照路径。

七、被否决的备选方案（Reject Alternatives）

PIP-178 还系统性地评估并否决了另外两条技术路线，理解它们有助于把握最终方案的设计取舍。

7.1 方案 A：在 TransactionBufferSnapshot 中增加 snapshotEntryCounts 字段

为每个快照增加"快照条目数"字段：普通快照为 1，多快照为其实际 entry 数：

public class TransactionBufferSnapshot { private String topicName; private long maxReadPositionLedgerId; private long maxReadPositionEntryId; private long snapshotEntryCount; private List<AbortTxnMetadata> aborts; }

否决原因（从提案表述可以推断）：该方案依赖"记录总数"来判定快照集合完整性，无法应对"压缩导致同键覆盖"与"部分写入"叠加时条目数不可靠的问题，且计数本身增加额外维护负担，不如"键内携带序号 + 末尾哨兵键"来得自洽。

7.2 方案 B：用 null 字段标记多快照

多快照的中间 entry 只写 aborts 和 maxReadPosition 数据，不设置 topicName，仅最后一个 entry 设置 topicName。读取时topicName == null表示多快照开始，topicName != null表示多快照结束：

entry: topicName=null, aborts=..., maxReadPosition=... (多快照开始) entry: topicName=null, aborts=..., maxReadPosition=... entry: topicName=xxx, aborts=..., maxReadPosition=... (多快照结束)

否决原因：以"null 字段"作为边界标记过于脆弱——一旦最后一个 entry 缺失或乱序，边界判定就会失效；同时 topicName 作为通用字段被赋予双重语义，破坏了数据模型的一致性。相比之下，"键中带序号 +-end哨兵键"将边界信息放在消息键（key）上，天然兼容压缩主题的覆盖规则，更稳健。

7.3 备选方案的 API 变更讨论

方案 A/B 都伴随对 Writer API 的改造讨论——为每条消息显式指定写入的 topic（key）：

interface Writer<T> { /** * Write event to the system topic. * @param t pulsar event * @param topic the topicName for the pulsar event * @return message id * @throws PulsarClientException exception while write event cause */ MessageId write(T t, String topic) throws PulsarClientException; /** * Async write event to the system topic. * @param t pulsar event * @param topic the topicName for the pulsar event * @return message id future */ CompletableFuture<MessageId> writeAsync(T t, String topic); }

其对应实现按 key 写入消息，正是提案中多快照"以 key 区分 entry"的基础设施：

public CompletableFuture<MessageId> writeAsync(TransactionBufferSnapshot transactionBufferSnapshot, String topicName) { return producer.newMessage().key(topicName) .value(transactionBufferSnapshot).sendAsync(); }

最终方案沿用了"按 key 写入"的思路（这也是 System Topic 压缩语义的前提），但放弃了 null 标记与计数字段，改用有序键 + 哨兵键。

八、总结

PIP-178 解决了一个非常具体的工程问题：aborted transaction IDs 无限增长导致单 BookKeeper entry 无法容纳一份完整快照时，如何安全地拆分与恢复。其核心贡献可以归纳为三点：

1. 数据有序化：将 aborts 改为按 aborted marker 位置有序的映射结构，使中止事务集合成为 FIFO 序列，为幂等覆盖与按序清理奠定基础；
2. 键语义化：用topicName-1...topicName-end的键体系承载多 entry，并把 maxReadPosition 锚点收敛到-end键，保证"始终存在正确的恢复锚点"；
3. 容错归一化：无论 broker 重启导致的部分写入，还是压缩导致的同键覆盖，恢复结果都保持一致，配合按序的过期清理，任何时刻都不会丢失有效数据。

该提案在 Apache Pulsar 中的最终落地形态是"分段快照"架构（v2 元数据 +SnapshotSegmentAbortedTxnProcessorImpl+transactionBufferSegmentedSnapshotEnabled开关），读者可以在pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/目录下按"元数据类 → 处理器实现 → 配置项"的线索继续深入研读源码。

赞分享

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

【免费下载链接】pulsar

Apache Pulsar - distributed pub-sub messaging system

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

点击查看免费下载

相关推荐

• Apache Pulsar PIP-36 详解：max_message_size 动态协商机制的设计与实现

  Apache Pulsar PIP 36 详解：max_message_size 动态协商机制的设计与实现 本文基于 Apache Pulsar 仓库中的 PI

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

• Apache Pulsar 无状态代理（Pulsar Proxy）架构解析：基于 PIP-1 的二进制协议代理设计与实现

  Apache Pulsar 无状态代理（Pulsar Proxy）架构解析：基于 PIP 1 的二进制协议代理设计与实现 导读 本文以 Apache Pulsa

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

• Apache Pulsar PIP-33 复制订阅（Replicated Subscriptions）设计与实现解析

  Apache Pulsar PIP 33 复制订阅（Replicated Subscriptions）设计与实现解析 本篇围绕 pip/pip 33.md ht

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

上一篇：brpc 高性能哈希容器 FlatMap 全解析：原理、API 使用与基准测试

下一篇：TanStack Table 核心 API 深度解析：Column 接口（Column_Column）的完整指南