文章摘要

本文整理自 ApacheCon Asia 2022 大会上，StreamNative 工程师赵延的分享《深入探究 Apache Pulsar 双阶段删除协议的工作原理》。目前，Apache Pulsar 对于数据的删除存在两个步骤，即删除元数据与实际存储数据。由于这两个步骤互相分开，我们无法保证实际存储数据一定能被删除，因此可能存在元数据已删除而实际存储数据依然存在的现象。现有 Pulsar 的用户在生产环境也遇到了该问题，存在大量的脏数据无法删除。对此，我们引入双阶段删除协议来解决该问题。本文将详细介绍两段数据删除的工作原理。

作者介绍

赵延，StreamNative 软件工程师，Apache Pulsar Contributor。

背景

关于 Apache Pulsar

Apache Pulsar 是云原生的消息和流平台。云原生的定义是“将计算与存储分离，可与云基础设施和 Kubernetes 一起工作的多层方案”，Pulsar 计算存储分离，存储有状态而计算无状态，可以方便地启动或者停止 Broker 运行在 K8s 上。

上图是 Apache Pulsar 的架构简介。生产者将消息发送给 Broker，而 Broker 并不负责消息的存储，消息存储在底层的 Apache BookKeeper 上。Broker 收到消息后通过 Pulsar 客户端将消息发送给 Bookie。消费者将订阅请求发送给 Broker，后者从 BookKeeper 中读取数据并发送给消费者。可见 Broker 和 BookKeeper 层分别负责计算和存储，是存算分离的设计。

负责存储的 BookKeeper 是多副本协议存储系统，具备以下特性：

高数据存储容量，可以通过增加服务端 Bookie 节点来无限扩容； 高一致性（数据重复读取时保证一致性）；读写高可用性；I/O 隔离。Broker 收到消息后，存储数据的操作分为以下几步：

1. 通过 BookKeeper 客户端，在 BookKeeper 中创建 Ledger handler 处理器（负责处理所有写入请求）；

2. 将上述 Ledger handler 的唯一标识 ID 添加到元数据存储中；

3. 通过 Ledger handler 将数据发送给 BookKeeper 服务端完成数据写入。

上述任何一步失败，本次数据存储操作都会失败。

关于 Broker 数据删除

与存储数据相对，Broker 删除数据的操作分为以下两步：

1. 在元数据存储中删除 Ledger handler 的 ID；

2. 第一步成功后，通过 BookKeeper 客户端删除 Ledger handler。

如果先完成第二步的操作再操作第一步，而第一步又因为 Broker 重启而失败，那么Broker 就会误认为已经被删除的数据仍然存在。所以第一步和第二步的操作顺序不能调换。

但这里又产生了一个问题：考虑到两步操作分别由计算和存储端执行，就可能出现第一步的操作成功而第二步操作失败的情况。在第二步的操作失败后，Pulsar 会重试三次，三次重试全部失败就不会继续重试，数据就会永久保存在 BookKeeper 中，占据额外的存储空间。

目前 Apache Pulsar 提供了一种工具来探测此类状况，通过元数据存储和真实数据存储的对比来发现未能成功删除的数据，但这一过程较为复杂，还会产生误删数据的风险，并带来更多运维压力。

基于 System Topic 的双阶段数据删除协议

针对上述问题，Apache Pulsar 新引入了基于 System Topic 的双阶段数据删除协议。该协议的基本工作流程如下：

1. 删除数据时，System Topic Producer 会向 System Topic Consumer 发送一个删除事件消息。

2. 后者收到消息后给 Broker 发送删除命令，要求后者执行删除操作。

3. 如果数据的元数据 Ledger 已被删除，System Topic Consumer 会自己到数据存储中执行删除操作，删除真实数据。

Pulsar 引入了以下三个新的 System Topic：

pulsar/system/persistent/__ledger_deletion：存储删除 Ledger 的事件；pulsar/system/persistent/__ledger_deletion-RETRY：存储重试事件，当消费者消费消息失败，或删除 Ledger 失败后，Pulsar 会将删除事件发送给该 Topic，该 Topic 在一段时间后将删除事件重新投递给消费者； pulsar/system/persistent/__ledger_deletion-DLQ：当消费者重复删除某 Ledger 达到一定次数后，删除事件会发送给该死信队列，不再消费。Pulsar 会认为该 Ledger 短时间内无法被删除。用户可以查看归档死信队列中的数据，判断哪些数据没有被真正删除。

在 Broker 启动时，Pulsar 会通过 pulsarAdmin 来创建分区的 System Topic，即上述三个 Topic。之所以是分区 Topic，是因为 Pulsar 会自动创建一些 Topic；当删除数据时，Broker 收到消息后会自动创建一个非分区 Topic。删除 Ledger 的事件都会写入该非分区 Topic。

但当 Broker 配置变化时，例如用户将 allowAutoTopicCreationType=non-partitioned 改为 allowAutoTopicCreationType=partitioned，Broker 就会自动创建一个分区 Topic 来存储消息。之前非分区 Topic 中的消息不会再有消费者订阅，对应的 Ledger 也无法删除。

双阶段数据删除详解

第一阶段

Broker 启动时会启动一个生产者，为之后的数据删除做准备。

client.newProducer(Schema.AVRO(PendingDeleteLedgerInfo.class))    .topic("pulsar/system/persistent/__ledger deletion")    .enableBatching(false)    .createAsync();

需要注意这里禁止了 Batching，因为如果删除多个 Ledger 的消息事件 Batch 到同一消息中，当消费者收到消息后会很难 ack 消息。

第一阶段中，删除一个 Ledger 分为以下步骤：

1. 向 System Topic 发送删除 Ledger 事件消息；

2. 从元数据存储中删除 Ledgerid。 如果第一步失败，第二步则不能进行，这样才能保证消费者可以收到消息删除 Ledger。只有元数据 Ledgerid 删除成功，第一阶段才完成。

第二阶段

在 Broker 启动时，会启动一个消费者来消费删除 Ledger 事件消息。

client.newConsumer(Schema-AVRO(PendingDeleteLedgerInfo.class))    .topic("pulsar/system/persistent/_ledger_deletion")    .subscriptionName("ledger-deletion-worker")    .subscriptionType(SubscriptionType.Shared)    .enableRetry(true)    .deadLetterPolicy(DeadLetterPolicy.builder()        .retryLetterTopic("pulsar/system/persistent/_ledger_deletion-RETRY")        .deadLetterTopic("pulsar/system/persistent/_ledger_deletion-DLQ")        .maxRedeliverCount(10).build())    .subscribeAsync()

上述配置中，ledger-deletion-worker 是统一的消息订阅名称，有多个 Broker 时会共享同一订阅，消费进度一致。一个消费事件消费失败 10 次后进入死信队列。

第二阶段流程如下：

System Topic Consumer 启动完成后，会收到生产者发送的删除事件。然后检查 Ledger 所属的 Topic 是否存在，如果存在就将删除操作交给 Broker 处理，否则在 Consumer 一侧直接删除数据。

在 Broker 删除 Ledger 时会检查 Ledger 是否仍在使用中，Ledgerid 是否存在于元数据中。如果存在，这个 Ledger 可能还在被使用，所以不能删除，需要等待 10 分钟后再重新消费删除事件。如果不存在，就正常删除数据并 ack 消息。

安全性

使用 System Topic 来存储要删除 Ledger 的消息会带来安全性问题。因为用户可以通过生产者来向 System Topic 直接发送消息，所以可以被恶意利用，导致数据误删除。所以在删除数据时会做元数据校验。

当前存储在 BookKeeper 中的 Ledger 保存了 Ledger 所属的 Topic 信息。删除某个 Ledger 时会拉取 Ledger 对应的 Topic，和本次要删除的 Ledger Topic 参数进行校验，通过后才会删除。

但上述操作会有一定性能损耗，所以在 Broker 侧引入了一个删除缓存，避免每次删除都拉取元数据。生产者发送删除事件消息后，待删除 Ledgerid 会被缓存到内存中。当 Broker 收到删除命令，会检查 Ledgerid 是否在缓存中，如果是，会认为本次删除活动是受信任的；如果不是，则拉取元数据进行校验。Broker 重启后缓存才会被清空。

如何删除 Topic

引入双阶段删除后，要删除一个 Topic 时，需要通过双阶段删除操作逐个删除所有 Ledger 才能删除 Topic。

一些中间状态

删除 Ledger 事件消息成功发送，元数据却因为 Broker 宕机而未删除。消费者收到事件消息后，会将删除事件重发给 Broker。Broker 会检查对应的 Ledgerid 是否存在于元数据中，如果还存在就不会删除数据。删除 Ledger 事件消息成功发送，元数据却因为 Broker 宕机而未删除。片刻后，生产者重发消息给 System Topic，且这一次删除成功。这相当于给 System Topic 发送了两次删除同一 Ledger 的消息，消费者也会给 Broker 发送两次删除命令。第一次过程中 Broker 已经删除了数据，第二次过程中 BookKeeper 会发现要删除的数据已不存在，则会发送一个特殊状态码给客户端，此时认为第二次删除操作也是成功的。消费者收到消息并成功删除 Ledger，但未 ack 消息就宕机。这时删除事件消息会重新投递给消费者，再次执行删除操作。由于删除操作是幂等的，第二次操作会被认为成功并 ack 成功。总结

目前，Apache Pulsar 对于数据的删除的两个步骤，删除元数据与实际存储数据互相分开，可能导致存在大量的脏数据无法删除的情况。对此，我们引入 [PIP-186][1] 双阶段删除协议来解决该问题。

本文详细介绍了双阶段数据删除的工作原理。在第一阶段，Broker 启动时会启动一个生产者，如果第一步 Ledger 删除失败，第二步则不能进行，以此保证消费者可以收到消息删除 Ledger。在第二阶段，启动 Broker 时，会启动一个消费者来消费删除 Ledger 事件消息。该特性相关 PR 正在审核中，功能还未上线，关注 Apache Pulsar 邮件列表以及后续的版本发布获得最新信息。

引用链接

[1] [PIP-186]: