文章插图
其中 follower1 同步了 2 条数据，而 follower2 同步了 3 条数据 。
而 follower 的 HW = min(自己的LEO, 同步回来的HW)
3.follower 再次同步数据，同时 leader 又被写入了 5 条消息
文章插图
leader 更新了 HW
4.leader 给 follower 返回 fetch response
文章插图
根据公式，follower 更新 HW = 3
在一个分区中，leader 所在 broker 会记录所有副本的 LEO 和 自己的 HW；而 follower 所在的 broker 只会记录自己的 LEO 和 HW 。因此，在逻辑层面上，我们可以得到下图：
文章插图
0.11.0.0版本之前，Kafka 是基于 HW 的同步机制 ， 但是这个设计有可能出现数据丢失和数据不一致的问题 。Kafka 后面的版本通过 leader epoch 来进行优化 。3.3 数据丢失 & 数据不一致的解决方案3.2小节说到了 LEO 与 HW 的更新机制，并且提到这种设计可能会出现数据丢失和数据不一致 。我们先一起来看下这两个问题是如何产生的 。3.3.1 数据丢失假设某一分区在某一时刻的状态如下图（L 代表是 leader）：
文章插图
可以看见副本A的 LEO 是 2，HW 是 1；副本B的 LEO 是 2，HW 是 2 。显然 ， 哪怕没有新的消息写入副本B中，副本A也要过一小段时间才能追上副本A，并更新 HW 。
假设在副本A更新 HW = 2之前 ， A宕机了，随后立马就恢复 。这里会有一个截断机制——根据宕机之前持久化的HW 恢复消息 。也就是说，A只恢复了 m1，m2 丢失了 。
再假设 A 刚恢复，B 也宕机了 ， A 成为了 leader 。这时 B 又恢复了 ， 并成为了 follower 。由于 follower 的 HW 不能比 leader 的 HW 高，所以 B 的 m2 也丢失了 。
文章插图
总结：这里大家可以发现 follower 的 HW 更新是有一定间隙的，像我这个例子其实 follower 是拿到 m2 了，只不过 HW 要追上 leader 需要等下一次的 fetch request 。除非配置 acks=-1 并且配置min.insync.replicas 大于 1 ， unclean.leader.election.enable = true 才行 。3.3.2 数据不一致
文章插图
假设某一分区在某一时刻，副本A 的 HW = 2，LEO = 2；副本B 的 HW = 1，LEO = 1 。
又假设它们同时挂了 ， B 先恢复 。这时 ， B 会成为 leader，如下图：
文章插图
此时，B 写入新消息 m3 ， 并将 HW、LEO 更新为 2 。此时，A 也恢复了 。由于 A 的 HW 也是 2，所以没有截断消息 。如下图：
文章插图
这样一来，A 中 offset = 1 的消息是 m2 ， B 中 offset = 1 的消息是 m3，数据不一致了 。3.3.3 leader epoch为了解决 3.3.1 和 3.3.2 的问题，Kafka 从 0.11.0.0 开始引入 leader epoch，在需要截断时使用 leader epoch 作为依据，而不再是 HW 。
如果看框架代码比较多的同学应该知道 epoch 是相当于版本的这么一个概念 。leader epoch 的初始值是 0，每变更一次 leader ， leader epoch 就会增加 1 。另外，每个副本中还会增加一个矢量<LeaderEpoch => StartOffset> ， 其中 StartOffset 是当前 leader epoch 下写入第一条消息的偏移量 。每个副本的 Log 下都有一个 leader-epoch-checkpoint 文件，在发生 leader 变更时，会将对应的矢量追加到这个文件中 。3.3.3.1 解决数据丢失问题
文章插图
还是3.3.1的例子，只不过多了 leader epoch 矢量信息 。
副本A：HW=1，LEO=2，LE(leader epoch)=0，Offset(StartOffset)=0
leader-副本B：HW=2，LEO=2，LE=0，Offset(StartOffset)=0
假设在副本A更新 HW = 2之前，A宕机了，随后立马就恢复 。不过这里不会立马进行截断日志操作，而是会发送一个 OffsetsForLeaderEpochRequest 请求给 B，B 作为目前的 leader 在收到请求之后会返回 OffsetsForLeaderEpochResponse 给 A 。
我们先来看下 OffsetsForLeaderEpochRequest 和 OffsetsForLeaderEpochResponse 的数据结构 。如下图：
