Kafka -- 关于高水位和Leader Epoch的讨论

Posted 2022-02-05 果汁华

什么是高水位？

        在 Kafka 的世界中，水位的概念有一点不同。Kafka 的水位不是时间戳，更与时间无关。它是和位置信息绑定的，具体来说，它是用消息位移来表征的。用来区分已消费和未消费数据。 （Kafka 中也有低水位（Low Watermark），它是与 Kafka 删除消息相关联的概念）

高水位的作用

        在 Kafka 中，高水位的作用主要有 2 个。

1、定义消息可见性，即用来标识分区下的哪些消息是可以被消费者消费的。

2、帮助 Kafka 完成副本同步。

        我们假设这是某个分区 Leader 副本的高水位图。首先，请你注意图中的“已提交消息”和“未提交消息”。我们之前在专栏第 11 讲谈到 Kafka 持久性保障的时候，特意对两者进行了区分。现在，我借用高水位再次强调一下。在分区高水位以下的消息被认为是已提交消息，反之就是未提交消息。消费者只能消费已提交消息，即图中位移小于 8 的所有消息。注意，这里我们不讨论 Kafka 事务，因为事务机制会影响消费者所能看到的消息的范围，它不只是简单依赖高水位来判断。它依靠一个名为 LSO（Log Stable Offset）的位移值来判断事务型消费者的可见性。

        另外，需要关注的是，位移值等于高水位的消息也属于未提交消息。也就是说，高水位上的消息是不能被消费者消费的。

        图中还有一个日志末端位移的概念，即 Log End Offset，简写是 LEO。它表示副本写入下一条消息的位移值。注意，数字 15 所在的方框是虚线，这就说明，这个副本当前只有 15 条消息，位移值是从 0 到 14，下一条新消息的位移是 15。显然，介于高水位和 LEO 之间的消息就属于未提交消息。这也从侧面告诉了我们一个重要的事实，那就是：同一个副本对象，其高水位值不会大于 LEO 值。

        高水位和 LEO 是副本对象的两个重要属性。Kafka 所有副本都有对应的高水位和 LEO 值，而不仅仅是 Leader 副本。只不过 Leader 副本比较特殊，Kafka 使用 Leader 副本的高水位来定义所在分区的高水位。换句话说，分区的高水位就是其 Leader 副本的高水位。

高水位更新机制

        现在，我们知道了每个副本对象都保存了一组高水位值和 LEO 值，但实际上，在 Leader 副本所在的 Broker 上，还保存了其他 Follower 副本的 LEO 值。我们一起来看看下面这张图。

        在这张图中，我们可以看到，Broker 0 上保存了某分区的 Leader 副本和所有 Follower 副本的 LEO 值，而 Broker 1 上仅仅保存了该分区的某个 Follower 副本。Kafka 把 Broker 0 上保存的这些 Follower 副本又称为远程副本（Remote Replica）。Kafka 副本机制在运行过程中，会更新 Broker 1 上 Follower 副本的高水位和 LEO 值，同时也会更新 Broker 0 上 Leader 副本的高水位和 LEO 以及所有远程副本的 LEO，但它不会更新远程副本的高水位值，也就是我在图中标记为灰色的部分。（Broker 0 所有远程副本 HW不会更新 ，一轮请求仅仅是 拉取消息，更新 follwer的 leo 在follower的二轮请求中更新leader保存的远程副本的leo和 leader的 hw和。 在二轮请求的响应中，follwer 更新其 hw）

 

为什么要在 Broker 0 上保存这些远程副本呢？其实，它们的主要作用是，帮助 Leader 副本确定其高水位，也就是分区高水位。

下面，我们分别从 Leader 副本和 Follower 副本两个维度，来总结一下高水位和 LEO 的更新机制。

Leader 副本

处理生产者请求的逻辑如下：

1、写入消息到本地磁盘。

2、更新分区高水位值。

   i.获取 Leader 副本所在 Broker 端保存的所有远程副本 LEO 值（LEO-1，LEO-2，……，LEO-n）。

  ii: 获取 Leader 副本高水位值：currentHW。

  iii: 更新 currentHW = maxcurrentHW, min（LEO-1, LEO-2, ……，LEO-n）。

处理 Follower 副本拉取消息的逻辑如下：

1、读取磁盘（或页缓存）中的消息数据。

2、使用 Follower 副本发送请求中的位移值更新远程副本 LEO 值。

3、更新分区高水位值（具体步骤与处理生产者请求的步骤相同）。

Follower 副本

从 Leader 拉取消息的处理逻辑如下：

1、写入消息到本地磁盘。

2、更新 LEO 值。

3、更新高水位值。

   i. 获取 Leader 发送的高水位值：currentHW。

  ii. 获取步骤 2 中更新过的 LEO 值：currentLEO。

  iii. 更新高水位为 min(currentHW, currentLEO)。

Leader Epoch 登场

        故事讲到这里似乎很完美，依托于高水位，Kafka 既界定了消息的对外可见性，又实现了异步的副本同步机制。不过，我们还是要思考一下这里面存在的问题。

        从刚才的分析中，我们知道，Follower 副本的高水位更新需要一轮额外的拉取请求才能实现。如果把上面那个例子扩展到多个 Follower 副本，情况可能更糟，也许需要多轮拉取请求。也就是说，Leader 副本高水位更新和 Follower 副本高水位更新在时间上是存在错配的。这种错配是很多“数据丢失”或“数据不一致”问题的根源。基于此，社区在 0.11 版本正式引入了 Leader Epoch 概念，来规避因高水位更新错配导致的各种不一致问题。

        Leader Epoch Offset 规避单纯使用 LEO 同步可能造成的数据丢失。

        所谓 Leader Epoch，我们大致可以认为是 Leader 版本。它由两部分数据组成。

1、Epoch。一个单调增加的版本号。每当副本领导权发生变更时，都会增加该版本号。小版本号的 Leader 被认为是过期 Leader，不能再行使 Leader 权力。

2、起始位移（Start Offset）。Leader 副本在该 Epoch 值上写入的首条消息的位移。

        我举个例子来说明一下 Leader Epoch。假设现在有两个 Leader Epoch<0, 0> 和 <1, 120>，那么，第一个 Leader Epoch 表示版本号是 0，这个版本的 Leader 从位移 0 开始保存消息，一共保存了 120 条消息。之后，Leader 发生了变更，版本号增加到 1，新版本的起始位移是 120。

        Kafka Broker 会在内存中为每个分区都缓存 Leader Epoch 数据，同时它还会定期地将这些信息持久化到一个 checkpoint 文件中。当 Leader 副本写入消息到磁盘时，Broker 会尝试更新这部分缓存。如果该 Leader 是首次写入消息，那么 Broker 会向缓存中增加一个 Leader Epoch 条目，否则就不做更新。这样，每次有 Leader 变更时，新的 Leader 副本会查询这部分缓存，取出对应的 Leader Epoch 的起始位移，以避免数据丢失和不一致的情况。

总结

• 高水位的2个作用：定义消息的可见性；帮助kafka完成副本同步；
• 在分区高水位以下的消息被认为是已提交的消息，反之就是未提交的消息。消费者只能消费已提交的消息。
• 日志末端位移（LEO: log end offset）：表示副本写入下一条消息的位移值。
• Leader Epoch:  我们大致认为是Leader版本。它由两部分数据组成。一个是Epoch ， 一个单调增加的版本号。每当副本领导权发生变更时，都会增加该版本号。另一个是起始位移Leader副本在该Epoch值上写入的首条消息的位移。