Kafka 的多副本机制提升了数据容灾能力。

副本通常分为数据副本与服务副本。数据副本是指在不同的节点上持久化同一份数据；服务副本指多个节点提供同样的服务，每个节点都有能力接收来自外部的请求并进行相应的处理。

1 副本刨析

1.1 相关概念

AR：Assigned Replicas,分区中的所有副本。

ISR：In-Sync Replicas,与leader副本保持同步状态的副本集合。

LEO：Log End Offset,分区中最后一条消息的下一个位置。

HW：High Watermark，高水位。标识了一个特定的消息偏移量，消费者只能拉取到这个偏移量之前的消息。HW是ISR集合中最小的LEO。

1.2 失效副本

在ISR集合之外的副本称为失效副本。即处于同步失效的状态。

broker端参数replica.lag.time.max.ms 配置了一个follower副本滞后于leader副本的最长时间间隔。（默认值30s）

当follower副本将leader副本LEO之前的日志全部同步过来的间隔时间超过这个值时，该副本处于同步失效状态，会从ISR集合中移除。

当副本追赶上leader时，会更新该副本的lastCaughtUpTimeMs。如果副本还未追赶上leader，则用当前时间-lastCaughtUpTimeMs计算差值，如果差值大于上面配置的值，那么该副本处于同步失效状态。

追赶上leader副本的判定标准是此副本的LEO是否不小于leader副本的HW。

1.2.1 ISR的伸缩

“isr-expiration”任务用于周期性地检测每个分区是否需要缩减其ISR集合。周期是replica.lag.time.max.ms配置参数的一半。当检测到ISR集合中有副本失效时，就会收缩ISR集合。

当ISR集合发生变更时，会将变更后的记录缓存到isrChangeSet中。isr-change-propagation会周期性（固定值为2500ms）地检查isrChangeSet。如果发现了变更记录，它会在Zookeeper的/isr_change_notification路径下创建一个保存isrChangeSet信息的节点。Kafka为/isr_change_notification添加了一个Watcher，当这个节点中有子节点发生变化时会触发Watcher的动作。

注意，频繁触发Watcher会影响性能，Kafka为避免这种情况，当检测到ISR集合发生变化时，还需要检查以下两个条件：

1. 上次ISR集合发生变化距离现在已超过5s。
2. 上次写入Zookeeper的时间距离现在已超过60s。

1.3 副本LEO与HW的变化

图 副本同步过程中LEO与HW的变化

follower 向leader拉取消息时，LEO与HW的变更步骤如下：

1. follower向leader拉取消息时，请求会携带自身的LEO信息，即fetch_offset。
2. leader收到请求时，会先检查该副本是否在ISR中，如果在，则将自身的HW值更新为所有ISR中的follower的LEO最小的值（leader会保存其他follower副本的LEO，会在返回响应之前更新对应follower的LEO）。然后连同消息和HW一起返回FetchResponse给follower。
3. follower 在收到FetchResponse响应后，更新LEO，然后取自身LEO及返回的HW中的最小值来更新自身的HW。

1.3.1 LEO和HW的持久化

Kafka 会周期性的将所有分区的LEO刷写到recovery-point-offset-checkpoint中（恢复点文件）。将所有分区的HW刷写到replication-offset-checkpoint中（复制点文件）。

1.4 同步机制

在0.11.0版本之前，Kafka使用的是基于HW的同步机制，这样可能会出现数据丢失或数据不一致的问题。

1.4.1 数据丢失

图 副本宕机及恢复过程中数据丢失

1. 如图，某刻A副本为follower副本，LEO=5，HW=3。B为leader。此时A发生宕机。
2. A恢复，并且根据HW，对日志进行阶段，使LEO=3。
3. B发生宕机，A被选举为leader。
4. B恢复，成为follower。因为follower的HW不能大于leader的HW。所以B会更改HW，并进行日志阶段，使HW=3,LEO=3。
5. 丢失2条消息。

1.4.2 数据不一致

图 副本宕机及恢复过程中数据不一致

1. A与leader B 同时宕机。随后A先恢复，成为leader。
2. 有1条消息写入到该分区，leader A 的LEO变为4，HW也变更为4。
3. 此时B也恢复成为follower，因为其HW不大于leader的HW，且等于LEO，所以其不要解答日志，同时也不会拉取leader的数据。
4. B与leader A 的最新一条消息不一致。

1.4.3 Leader Epoch

为了解决上述两种问题，从0.11.0版本开始引入leader epoch的概念。

leader epoch 代表leader的纪元信息，初始值为0，每当leader变更一次，该值就会加1。

每个副本都会增设一个矢量<LeaderEpoch => StartOffset>，其中StartOffset表示当前LeaderEpoch下写入的第一条消息的偏移量（LEO）。在发生leader epoch变更时，每个副本会将对应的矢量追加到其Log下的leader-epoch-checkpoint文件中。

follower副本从宕机状态恢复后，会先发送OffsetsForLeaderEpochRequest请求给leader。将携带follower当前的Leader Epoch值。leader 收到该请求后会返回当前的LEO。如果follower的Leader Epoch值和leader的不相同，那么leader将会查找 Leader epoch 为 follower 的Leader Epoch 值 + 1对应的StartOffset，并返回。

follower在收到响应后，根据返回值与自身的LEO作对比，来决定是否需要将日志阶段截断使LEO等于返回值。

图 副本宕机及恢复过程Leader epoch的变化

1. A为leader，此时副本的LE（Leader epoch）都为0。B发生宕机，然后A发生宕机，此时C被选举为leader，并且C的LE+1,变更为1。
2. B 恢复，并且向C发送OffsetsForLeaderEpochRequest请求，C返回3，B收到响应后，将日志截断，使得LEO=3。

注意：当leader epoch 发送变更时，leader将会通知其他非宕机副本，使得它们来更新自己的<LeaderEpoch => StartOffset>矢量信息。

1. C发生宕机，B被选举为leader，并且B的LE+1,变更为2。随后B被写入两条新的消息，LEO变为5。
2. A恢复，并且向B发送OffsetsForLeaderEpochRequest请求，B返回LE为1的StartOffset，即为3。A收到响应后，将日志截断，使得其LEO=3.
3. 随后A变更LE为2。并且向B拉取消息。