Kafka 3.0 源码笔记(12)-Kafka 服务端分区异常恢复机制的源码分析

系统管理员 2023-10-01 21:56 36阅读 0赞

#### 文章目录 ####

*  1. HW 同步机制的问题
 *   *   *  1.1 数据丢失问题
         *  1.2 主从数据不一致
 *  2. Kafka 的异常恢复机制
 *   *   *  2.1 解决数据丢失问题
         *  2.2 解决数据不一致问题
 *  3. 异常恢复原理源码分析
 *   *   *  3.1 LeaderEpoch 的维护更新
         *   *  3.1.1 分区 Leader 当选时的更新
             *  3.1.2 消息写入时的维护
         *  3.2 异常恢复的处理

## 1. HW 同步机制的问题 ##

在 [Kafka 3.0 源码笔记(10)-Kafka 服务端消息数据的主从同步源码分析][Kafka 3.0 _10_-Kafka] 中，笔者介绍了消息数据在主从副本之间同步的流程，从中可以知道主从同步的核心是依赖 HW 来标识副本备份进度。而在这套机制中， Follower 至少需要进行两次 Fetch 请求才能成功更新本地 HW，也就是说 Follower 的 HW 在某个时间段内总是落后于 Leader 的 HW。在这样的情况下，一旦发生宕机等异常情况，节点重启后副本再根据 HW 值进行异常恢复时就有可能产生以下两个问题：

> 1.  数据丢失问题
> 2.  主从数据不一致

#### 1.1 数据丢失问题 ####

下图详细标示了数据丢失问题发生的全过程，可分为 5 个步骤进行简单描述：

> 1.  `新消息写入 Leader`  
>     初始时分区 Leader 副本位于 Broker1 上，分区 Follower 副本位于 Broker2，此时主从副本上的 HW 和 LEO 完全一致，\[HW = 1，LEO =1\]。当有新消息写入时，率先写入分区 Leader 副本， 此时分区 Leader 的日志尾部偏移量发生改变 LEO = 2
> 2.  `主从第一次 Fetch 请求交互`  
>     Follower 副本节点通过 Fetcher 线程定时发送 Fetch 请求到 Leader 副本同步消息，发起的请求中会携带本地分区副本的 LEO=1。 Leader 副本所在节点接收到请求后，更新目标分区 remoteReplicasMap 中保存的该 Follower 副本状态，尝试更新本地日志HW。此时只要当前 Follower 副本不是最后一个来同步消息的，Leader 副本就不会更新本地 HW，仅仅返回消息记录。 Follower 副本节点在处理 Fetch 响应时，仅能将消息追加到本地日志，并将 LEO 指向 Offset = 2 的位置
> 3.  `主从第二次 Fetch 交互，Follower 宕机`  
>     与第一次 Fetch 请求交互类似，只不过这时 Fetcher 线程发起的请求中会携带本地分区副本的 LEO=2。假设此时所有 Follower 副本都保存了新消息，那么 Leader 副本节点在尝试更新本地日志 HW 时会成功更新本地 HW 指向 Offset=2 的位置，并在 Fetch 响应中将当前 HW 返回给 Follower。然而此时 Follower 所在的节点 Broker2 宕机，没有处理这个 Fetch 响应，最终导致主从副本之间 HW 出现不一致，Leader \[HW=2, LEO=2\]，Follower \[HW=1, LEO=2\]
> 4.  `Follower 异常恢复，Leader 却宕机`  
>     Broker2 节点重启后，分布在其上的分区副本开始依据本地日志文件进行异常恢复。此时 Follower 根据本地的 HW = 1 进行日志截断操作，将偏移量大于等于 HW 的消息数据清除掉，则此时 Follower 日志数据发生变化 Follower \[HW=1, LEO=1\]，原 Offset = 1 的消息数据在 Follower 副本上丢失。接下来 Follower 重新开始和 Leader 进行 Fetch 交互，结果 Leader 所在节点 Broker1 宕机，Leader 副本上 Offset = 1 的消息数据并未同步到 Follower 副本
> 5.  `Broker2 上的 Follower 被选举为新 Leader，Broker1 重启同步数据`  
>     Broker1重启，该节点上原 Leader 副本异常恢复，成为 Follower \[HW=2, LEO=2\]。假如此时 Broker2 上的 Follower 被选举为新的 Leader，则当前 Leader \[HW=1, LEO=1\]。接下来当前 Follower 通过 Fetch 请求同步当前 Leader 数据，结果从 Fetch 响应中拿到 Leader 的 HW = 1，则根据该值在本地进行日志截断操作，至此原 Offset = 1 的消息数据彻底丢失

![在这里插入图片描述][b4c9d0e7dfbc410d83b592ed6d7a9239.jpeg_pic_center]

#### 1.2 主从数据不一致 ####

下图展示了主从数据不一致发生的全过程，与上一节类似，笔者将其分为 5 个步骤进行简单描述：

> 1.  `新消息写入 Leader`  
>     初始时分区 Leader 副本位于 Broker1 上，分区 Follower 副本位于 Broker2，此时主从副本上的 HW 和 LEO 完全一致，\[HW = 1，LEO =1\]。当有新消息写入时，率先写入分区 Leader 副本， 此时分区 Leader 的日志尾部偏移量发生改变 LEO = 2
> 2.  `主从第一次 Fetch 请求交互`  
>     Follower 副本节点通过 Fetcher 线程定时发送 Fetch 请求到 Leader 副本同步消息，发起的请求中会携带本地分区副本的 LEO = 1。 Leader 副本所在节点接收到请求后，更新目标分区 remoteReplicasMap 中保存的该 Follower 副本状态，尝试更新本地日志HW。此时只要当前 Follower 副本不是最后一个来同步消息的，Leader 副本就不会更新本地 HW，仅仅返回消息记录。 Follower 副本节点在处理 Fetch 响应时，仅能将消息追加到本地日志，并将 LEO 指向 Offset = 2 的位置
> 3.  `主从第二次 Fetch 交互，Follower 宕机`  
>     与第一次 Fetch 请求交互类似，只不过这时 Fetcher 线程发起的请求中会携带本地分区副本的 LEO=2。假设此时所有 Follower 副本都保存了新消息，那么 Leader 副本节点在尝试更新本地日志 HW 时会成功更新本地 HW 指向 Offset=2 的位置，并在 Fetch 响应中将当前 HW 返回给 Follower。然而此时 Follower 所在的节点 Broker2 宕机，没有处理这个 Fetch 响应，最终导致主从副本之间 HW 出现不一致，Leader \[HW=2, LEO=2\]，Follower \[HW=1, LEO=2\]
> 4.  `Follower 异常恢复，Leader 却宕机`  
>     Broker2 节点重启后，分布在其上的分区副本开始依据本地日志文件进行异常恢复。此时 Follower 根据本地的 HW = 1 进行日志截断操作，将偏移量大于等于 HW 的消息数据清除掉，则此时 Follower 日志数据发生变化 Follower \[HW=1, LEO=1\]，原 Offset = 1 的消息数据在 Follower 副本上丢失。接下来 Follower 重新开始和 Leader 进行 Fetch 交互，结果 Leader 所在节点 Broker1 宕机，Leader 副本上 Offset = 1 的消息数据并未同步到 Follower 副本
> 5.  `Broker2 上的 Follower 被选举为新 Leader，写入新消息`  
>     Broker1重启，该节点上原 Leader 副本异常恢复，成为 Follower \[HW=2, LEO=2\]。假如此时 Broker2 上的 Follower 被选举为新的 Leader，则当前 Leader \[HW=1, LEO=1\]。接下来新消息写入的请求先一步到达当前 Leader 节点，则将新消息写入本地 Offset = 1 的位置后当前 Leader \[HW=1, LEO=2\]。此后 Follower 的 Fetch 请求到来，请求中携带的 LEO = 2 促使当前 Leader 处理完请求后进行 HW 更新，即 Leader \[HW=2, LEO=2\]。至此，新的主从副本 HW 和 LEO 完全一致，\[HW = 2，LEO =2\]，然而在 Follower 上 Offset = 1 位置的消息为 MessageB，在 Leader 上 Offset = 1 位置的消息则为 MessageC，出现主从数据不一致问题

![在这里插入图片描述][50ab6359affa4370a734df5c67a9432e.jpeg_pic_center]

## 2. Kafka 的异常恢复机制 ##

导致以上两个问题的根本原因在于，被用作衡量副本备份进度的 HW 也被用于异常恢复的日志截断，但 HW 的更新是异步延迟的，因此这中间发生的任何崩溃都可能导致 HW 过期。有鉴于此，Kafka 引入 `LeaderEpoch` 来取代 HW 的异常恢复功能，通过缓存每个分区版本信息的方式来尽可能规避上面的两个场景

> LeaderEpoch 实际是一个 `[epoch，startOffset]` 键值对，用于表示整个分区的版本偏移量信息。当然 LeaderEpoch 机制也并非万无一失，在 `min.insync.replicas = 1`且`unclean.leader.election.enabled = true` 时，依然无法保证数据的可靠性
> 
>  *  `epoch`  
>     表示整个分区的版本号，分区 Leader 每次变更都自增1，从 0 开始
>  *  `startOffset`  
>     对应版本的 Leader 当选时的本地 LEO，从 0 开始
> 
> --------------------
> 
> 假设有两对值：\[0, 0\]，\[1, 128\]，即表示第一个 Leader 当选时版本号为 0，从 Offset = 0 处开始写入消息；第二个 Leader 版本号是1，从 Offset = 128 处开始写入消息。据此也可以计算得出，第一个 Leader 总共写入了 128 条消息

#### 2.1 解决数据丢失问题 ####

下图是 Kafka 异常恢复机制解决数据丢失问题的流程示意，依然可分为 5 个步骤：

> 1.  `新消息写入 Leader`  
>     初始时分区 Leader 副本位于 Broker1 上，分区 Follower 副本位于 Broker2，此时 LeaderEpoch \[epoch = 0, startOffset = 0\]，主从副本上的 HW 和 LEO 完全一致，\[HW = 1，LEO =1\]。当有新消息写入时，率先写入分区 Leader 副本， 此时分区 Leader 的日志尾部偏移量发生改变 LEO = 2
> 2.  `主从第一次 Fetch 请求交互`  
>     Follower 副本节点通过 Fetcher 线程定时发送 Fetch 请求到 Leader 副本同步消息。 Leader 副本所在节点接收到请求后，更新目标分区 remoteReplicasMap 中保存的该 Follower 副本状态，尝试更新本地日志HW。此时只要当前 Follower 副本不是最后一个来同步消息的，Leader 副本就不会更新本地 HW，仅仅返回消息记录。 Follower 副本节点在处理 Fetch 响应时，仅能将消息追加到本地日志，并将 LEO 指向 Offset = 2 的位置
> 3.  `主从第二次 Fetch 交互，Follower 宕机`  
>     与第一次 Fetch 请求交互类似，只不过这时 Fetcher 线程发起的请求中会携带本地分区副本的 LEO = 2。假设此时所有 Follower 副本都保存了新消息，那么 Leader 副本节点在尝试更新本地日志 HW 时会成功更新本地 HW 指向 Offset=2 的位置，并在 Fetch 响应中将当前 HW 返回给 Follower。然而此时 Follower 所在的节点 Broker2 宕机，没有处理这个 Fetch 响应，最终导致主从副本之间 HW 出现不一致，Leader \[HW=2, LEO=2\]，Follower \[HW=1, LEO=2\]
> 4.  `Follower 尝试异常恢复，Leader 却宕机`  
>     Broker2 节点重启后，分布在其上的分区副本首先向 Leader 发送 OffsetForLeaderEpoch 请求，以便确认异常恢复的位点。然而 Leader 所在节点 Broker1 宕机，Follower 副本并未进行日志截断，本地消息数据状态依然是 Follower \[HW=1, LEO=2\]
> 5.  `Broker2 上的 Follower 被选举为新 Leader，更新 LeaderEpoch`  
>     假如 Broker2 上的 Follower 被选举为新的 Leader，则在当选的那一刻它将更新 LeaderEpoch 并尝试更新本地水位，此处当前副本本该在以上第3步骤更新的 HW 得到更新，则当前副本状态为 Leader \[HW=2, LEO=2\]，LeaderEpoch \[epoch = 1, startOffset = 2\]。至此，新的主从副本上数据是完全一致的，offset = 1 位置的消息数据并未丢失

![在这里插入图片描述][7673dd1aa8544967b54085a95ac092ad.jpeg_pic_center]

#### 2.2 解决数据不一致问题 ####

数据不一致的问题通常发生在 Follower 副本落后 Leader 副本比较多的情况下，下图大致分析了相关流程：

> 1.  `Follower 不在 ISR 列表，同步时 Leader 宕机`  
>     初始时 Follower 不在 ISR 列表，则该 Follower 无法影响到 Leader 的 HW 更新，会存在 Follower 副本物理上的消息数据落后于 Leader 的情形。这种情况下，数据落后的 Follower 向 Leader 发送 Fetch 请求同步数据， Leader 却故障宕机，则当前 Follower 未同步到最新的消息
> 2.  `允许脏选举，Broker2 选举为新 Leader，写入新消息`  
>     unclean.leader.election.enabled 参数设置为 true 时，原本不在 ISR 列表的分区副本也有可能被选举为 Leader。此时，新的 Leader 产生，在它当选的时候更新本地 LeaderEpoch，则当前副本状态为 Leader \[HW=1, LEO=1\]，LeaderEpoch \[epoch = 1, startOffset = 1\]。。该副本成为 Leader 后接受 Produce 请求，写入一条新消息，则本地日志尾部偏移量指针发生变化  LEO = 2 
> 3.  `Broker1 重启，原Leader成为Follower异常恢复`  
>     Follower 重启后首先向当前 Leader 发送 OffsetForLeaderEpoch 请求，确认宕机期间是否发生分区主从切换。Leader 将本地的 LeaderEpoch \[epoch = 1, startOffset = 1\] 返回给 Follower，此时 Follower 自检发现本地 LEO 大于新 Leader 版本的 startOffset ，则执行日志截断， 将本地原 Offset = 1 位置的消息丢弃。此时 Follower 上的日志数据只是落后于 Leader，并没有不一致产生，则后续可以通过 Fetch 请求追上 Leader，主从数据不一致的问题不复存在

![在这里插入图片描述][d5f3e96eb157406db4b7897bb6a84275.jpeg_pic_center]

## 3. 异常恢复原理源码分析 ##

异常恢复的源码时序图如下所示，由于该功能的核心在于 LeaderEpoch 缓存数据，所以本节内容分为了两个部分：

> 1.  LeaderEpoch 的维护更新
> 2.  异常恢复的处理

![在这里插入图片描述][4aead4c354884cdcb560fc014702113b.jpeg_pic_center]

#### 3.1 LeaderEpoch 的维护更新 ####

在 Kafka 3.0 版本的源码中，涉及到 LeaderEpoch 维护更新的也有以下两个部分，下文笔者将详细分析

> 1.  分区 Leader 当选时的更新
> 2.  消息数据写入时的维护

##### 3.1.1 分区 Leader 当选时的更新 #####

1.  分区当选后需要调用 `Partition.scala#makeLeader()` 方法进行选主完成的处理，这部分笔者在 [Kafka 3.0 源码笔记(10)-Kafka 服务端消息数据的主从同步源码分析][Kafka 3.0 _10_-Kafka 1] 也有提及，此处只关注和 LeaderEpoch 更新有关的部分，也就是调用 `Log.scala#maybeAssignEpochStartOffset()` 方法将本地 LEO 作为新分区版本的 startOffset 缓存起来
    
        def makeLeader(partitionState: LeaderAndIsrPartitionState,
                      highWatermarkCheckpoints: OffsetCheckpoints,
                      topicId: Option[Uuid]): Boolean = {
              
         val (leaderHWIncremented, isNewLeader) = inWriteLock(leaderIsrUpdateLock) {
              
           // record the epoch of the controller that made the leadership decision. This is useful while updating the isr
           // to maintain the decision maker controller's epoch in the zookeeper path
           
           ......
        
           //We cache the leader epoch here, persisting it only if it's local (hence having a log dir)
           leaderEpoch = partitionState.leaderEpoch
           leaderEpochStartOffsetOpt = Some(leaderEpochStartOffset)
           zkVersion = partitionState.zkVersion
        
           // In the case of successive leader elections in a short time period, a follower may have
           // entries in its log from a later epoch than any entry in the new leader's log. In order
           // to ensure that these followers can truncate to the right offset, we must cache the new
           // leader epoch and the start offset since it should be larger than any epoch that a follower
           // would try to query.
           leaderLog.maybeAssignEpochStartOffset(leaderEpoch, leaderEpochStartOffset)
        
           ......
        
           // we may need to increment high watermark since ISR could be down to 1
           (maybeIncrementLeaderHW(leaderLog), isNewLeader)
         }
         // some delayed operations may be unblocked after HW changed
         if (leaderHWIncremented)
           tryCompleteDelayedRequests()
         isNewLeader
        }
2.  `Log.scala#maybeAssignEpochStartOffset()` 方法只是入口，核心的调用在于 `LeaderEpochFileCache.scala#assign()` 方法
    
        def maybeAssignEpochStartOffset(leaderEpoch: Int, startOffset: Long): Unit = {
              
         leaderEpochCache.foreach {
               cache =>
           cache.assign(leaderEpoch, startOffset)
         }
        }
3.  `LeaderEpochFileCache.scala#assign()` 方法比较简单，首先是将新的分区版本号和对应的开始偏移量封装到键值对 `EpochEntry` 中，调用 `LeaderEpochFileCache.scala#assign()` 重载方法将其缓存，如果缓存成功则调用 `LeaderEpochFileCache.scala#flush()` 方法将其写入本地文件 `LeaderEpochCheckpointFile`，至此
    
        def assign(epoch: Int, startOffset: Long): Unit = {
              
         val entry = EpochEntry(epoch, startOffset)
         if (assign(entry)) {
              
           debug(s"Appended new epoch entry $entry. Cache now contains ${epochs.size} entries.")
           flush()
         }
        }
        
        // 将 EpochEntry 存入内部 TreeMap
        private def assign(entry: EpochEntry): Boolean = {
              
         if (entry.epoch < 0 || entry.startOffset < 0) {
              
           throw new IllegalArgumentException(s"Received invalid partition leader epoch entry $entry")
         }
        
         def isUpdateNeeded: Boolean = {
              
           latestEntry match {
              
             case Some(lastEntry) =>
               entry.epoch != lastEntry.epoch || entry.startOffset < lastEntry.startOffset
             case None =>
               true
           }
         }
        
         // Check whether the append is needed before acquiring the write lock
         // in order to avoid contention with readers in the common case
         if (!isUpdateNeeded)
           return false
        
         inWriteLock(lock) {
              
           if (isUpdateNeeded) {
              
             maybeTruncateNonMonotonicEntries(entry)
             epochs.put(entry.epoch, entry)
             true
           } else {
              
             false
           }
         }
        }
        
        private def flush(): Unit = {
              
         checkpoint.write(epochs.values.asScala)
        }

##### 3.1.2 消息写入时的维护 #####

1.  Follower 通过 Fetch 请求同步 Leader 的数据到本地，写入消息到本地日志文件需调用 `Partition.scala#appendRecordsToFollowerOrFutureReplica()`方法，可以看到这里最终的核心动作其实是调用 `Log.scala#appendAsFollower()` 方法
    
        def appendRecordsToFollowerOrFutureReplica(records: MemoryRecords, isFuture: Boolean): Option[LogAppendInfo] = {
              
         try {
              
           doAppendRecordsToFollowerOrFutureReplica(records, isFuture)
         } catch {
              
           case e: UnexpectedAppendOffsetException =>
             val log = if (isFuture) futureLocalLogOrException else localLogOrException
             val logEndOffset = log.logEndOffset
             if (logEndOffset == log.logStartOffset &&
                 e.firstOffset < logEndOffset && e.lastOffset >= logEndOffset) {
              
               // This may happen if the log start offset on the leader (or current replica) falls in
               // the middle of the batch due to delete records request and the follower tries to
               // fetch its first offset from the leader.
               // We handle this case here instead of Log#append() because we will need to remove the
               // segment that start with log start offset and create a new one with earlier offset
               // (base offset of the batch), which will move recoveryPoint backwards, so we will need
               // to checkpoint the new recovery point before we append
               val replicaName = if (isFuture) "future replica" else "follower"
               info(s"Unexpected offset in append to $topicPartition. First offset ${e.firstOffset} is less than log start offset ${log.logStartOffset}." +
                    s" Since this is the first record to be appended to the $replicaName's log, will start the log from offset ${e.firstOffset}.")
               truncateFullyAndStartAt(e.firstOffset, isFuture)
               doAppendRecordsToFollowerOrFutureReplica(records, isFuture)
             } else
               throw e
         }
        }
        
        private def doAppendRecordsToFollowerOrFutureReplica(records: MemoryRecords, isFuture: Boolean): Option[LogAppendInfo] = {
              
         if (isFuture) {
              
           // The read lock is needed to handle race condition if request handler thread tries to
           // remove future replica after receiving AlterReplicaLogDirsRequest.
           inReadLock(leaderIsrUpdateLock) {
              
             // Note the replica may be undefined if it is removed by a non-ReplicaAlterLogDirsThread before
             // this method is called
             futureLog.map {
               _.appendAsFollower(records) }
           }
         } else {
              
           // The lock is needed to prevent the follower replica from being updated while ReplicaAlterDirThread
           // is executing maybeReplaceCurrentWithFutureReplica() to replace follower replica with the future replica.
           futureLogLock.synchronized {
              
             Some(localLogOrException.appendAsFollower(records))
           }
         }
        }
2.  `Log.scala#appendAsFollower()` 方法只是入口，这里的关键逻辑是调用 `Log.scala#append()` 方法将消息追加到本地日志文件
    
        def appendAsFollower(records: MemoryRecords): LogAppendInfo = {
              
         append(records,
           origin = AppendOrigin.Replication,
           interBrokerProtocolVersion = ApiVersion.latestVersion,
           validateAndAssignOffsets = false,
           leaderEpoch = -1,
           None,
           // disable to check the validation of record size since the record is already accepted by leader.
           ignoreRecordSize = true)
        }
3.  `Log.scala#append()` 方法很长，这个方法是主从副本公用的写消息方法，因此会根据方法入参数做区分处理
    
    > 1.  如果是写消息到 Leader 副本，则 `validateAndAssignOffsets` 参数为 true，后续处理时会将 leaderEpoch 填充写入到消息记录中，并调用 `Log.scala#maybeAssignEpochStartOffset()` 方法尝试维护更新本地 LeaderEpoch 缓存
    > 2.  如果是写消息到 Follower 副本，则 `validateAndAssignOffsets` 参数为 false，不需要填充 leaderEpoch，直接从同步自 Leader 的消息记录中读取分区版本信息，之后调用 `Log.scala#maybeAssignEpochStartOffset()` 方法尝试维护更新本地 LeaderEpoch 缓存
    > 
    > --------------------
    > 
    > 此处 `Log.scala#maybeAssignEpochStartOffset()` 方法后续处理参考上一节3.1.1步骤2即可
    
        private def append(records: MemoryRecords,
                          origin: AppendOrigin,
                          interBrokerProtocolVersion: ApiVersion,
                          validateAndAssignOffsets: Boolean,
                          leaderEpoch: Int,
                          requestLocal: Option[RequestLocal],
                          ignoreRecordSize: Boolean): LogAppendInfo = {
              
         // We want to ensure the partition metadata file is written to the log dir before any log data is written to disk.
         // This will ensure that any log data can be recovered with the correct topic ID in the case of failure.
         maybeFlushMetadataFile()
        
         val appendInfo = analyzeAndValidateRecords(records, origin, ignoreRecordSize, leaderEpoch)
        
         // return if we have no valid messages or if this is a duplicate of the last appended entry
         if (appendInfo.shallowCount == 0) appendInfo
         else {
              
        
           // trim any invalid bytes or partial messages before appending it to the on-disk log
           var validRecords = trimInvalidBytes(records, appendInfo)
        
           // they are valid, insert them in the log
           lock synchronized {
              
             maybeHandleIOException(s"Error while appending records to $topicPartition in dir ${dir.getParent}") {
              
               checkIfMemoryMappedBufferClosed()
               if (validateAndAssignOffsets) {
              
                 // assign offsets to the message set
                 val offset = new LongRef(nextOffsetMetadata.messageOffset)
                 appendInfo.firstOffset = Some(LogOffsetMetadata(offset.value))
                 val now = time.milliseconds
                 val validateAndOffsetAssignResult = try {
              
                   LogValidator.validateMessagesAndAssignOffsets(validRecords,
                     topicPartition,
                     offset,
                     time,
                     now,
                     appendInfo.sourceCodec,
                     appendInfo.targetCodec,
                     config.compact,
                     config.recordVersion.value,
                     config.messageTimestampType,
                     config.messageTimestampDifferenceMaxMs,
                     leaderEpoch,
                     origin,
                     interBrokerProtocolVersion,
                     brokerTopicStats,
                     requestLocal.getOrElse(throw new IllegalArgumentException(
                       "requestLocal should be defined if assignOffsets is true")))
                 } catch {
              
                   case e: IOException =>
                     throw new KafkaException(s"Error validating messages while appending to log $name", e)
                 }
                 validRecords = validateAndOffsetAssignResult.validatedRecords
                 appendInfo.maxTimestamp = validateAndOffsetAssignResult.maxTimestamp
                 appendInfo.offsetOfMaxTimestamp = validateAndOffsetAssignResult.shallowOffsetOfMaxTimestamp
                 appendInfo.lastOffset = offset.value - 1
                 appendInfo.recordConversionStats = validateAndOffsetAssignResult.recordConversionStats
                 if (config.messageTimestampType == TimestampType.LOG_APPEND_TIME)
                   appendInfo.logAppendTime = now
        
                 // re-validate message sizes if there's a possibility that they have changed (due to re-compression or message
                 // format conversion)
                 if (!ignoreRecordSize && validateAndOffsetAssignResult.messageSizeMaybeChanged) {
              
                   validRecords.batches.forEach {
               batch =>
                     if (batch.sizeInBytes > config.maxMessageSize) {
              
                       // we record the original message set size instead of the trimmed size
                       // to be consistent with pre-compression bytesRejectedRate recording
                       brokerTopicStats.topicStats(topicPartition.topic).bytesRejectedRate.mark(records.sizeInBytes)
                       brokerTopicStats.allTopicsStats.bytesRejectedRate.mark(records.sizeInBytes)
                       throw new RecordTooLargeException(s"Message batch size is ${batch.sizeInBytes} bytes in append to" +
                         s"partition $topicPartition which exceeds the maximum configured size of ${config.maxMessageSize}.")
                     }
                   }
                 }
               } else {
              
                 // we are taking the offsets we are given
                 if (!appendInfo.offsetsMonotonic)
                   throw new OffsetsOutOfOrderException(s"Out of order offsets found in append to $topicPartition: " +
                     records.records.asScala.map(_.offset))
        
                 if (appendInfo.firstOrLastOffsetOfFirstBatch < nextOffsetMetadata.messageOffset) {
              
                   // we may still be able to recover if the log is empty
                   // one example: fetching from log start offset on the leader which is not batch aligned,
                   // which may happen as a result of AdminClient#deleteRecords()
                   val firstOffset = appendInfo.firstOffset match {
              
                     case Some(offsetMetadata) => offsetMetadata.messageOffset
                     case None => records.batches.asScala.head.baseOffset()
                   }
        
                   val firstOrLast = if (appendInfo.firstOffset.isDefined) "First offset" else "Last offset of the first batch"
                   throw new UnexpectedAppendOffsetException(
                     s"Unexpected offset in append to $topicPartition. $firstOrLast " +
                       s"${appendInfo.firstOrLastOffsetOfFirstBatch} is less than the next offset ${nextOffsetMetadata.messageOffset}. " +
                       s"First 10 offsets in append: ${records.records.asScala.take(10).map(_.offset)}, last offset in" +
                       s" append: ${appendInfo.lastOffset}. Log start offset = $logStartOffset",
                     firstOffset, appendInfo.lastOffset)
                 }
               }
        
               // update the epoch cache with the epoch stamped onto the message by the leader
               validRecords.batches.forEach {
               batch =>
                 if (batch.magic >= RecordBatch.MAGIC_VALUE_V2) {
              
                   maybeAssignEpochStartOffset(batch.partitionLeaderEpoch, batch.baseOffset)
                 } else {
              
                   // In partial upgrade scenarios, we may get a temporary regression to the message format. In
                   // order to ensure the safety of leader election, we clear the epoch cache so that we revert
                   // to truncation by high watermark after the next leader election.
                   leaderEpochCache.filter(_.nonEmpty).foreach {
               cache =>
                     warn(s"Clearing leader epoch cache after unexpected append with message format v${batch.magic}")
                     cache.clearAndFlush()
                   }
                 }
               }
        
               // check messages set size may be exceed config.segmentSize
               if (validRecords.sizeInBytes > config.segmentSize) {
              
                 throw new RecordBatchTooLargeException(s"Message batch size is ${validRecords.sizeInBytes} bytes in append " +
                   s"to partition $topicPartition, which exceeds the maximum configured segment size of ${config.segmentSize}.")
               }
        
               // maybe roll the log if this segment is full
               val segment = maybeRoll(validRecords.sizeInBytes, appendInfo)
        
               val logOffsetMetadata = LogOffsetMetadata(
                 messageOffset = appendInfo.firstOrLastOffsetOfFirstBatch,
                 segmentBaseOffset = segment.baseOffset,
                 relativePositionInSegment = segment.size)
        
               // now that we have valid records, offsets assigned, and timestamps updated, we need to
               // validate the idempotent/transactional state of the producers and collect some metadata
               val (updatedProducers, completedTxns, maybeDuplicate) = analyzeAndValidateProducerState(
                 logOffsetMetadata, validRecords, origin)
        
               maybeDuplicate match {
              
                 case Some(duplicate) =>
                   appendInfo.firstOffset = Some(LogOffsetMetadata(duplicate.firstOffset))
                   appendInfo.lastOffset = duplicate.lastOffset
                   appendInfo.logAppendTime = duplicate.timestamp
                   appendInfo.logStartOffset = logStartOffset
                 case None =>
                   // Before appending update the first offset metadata to include segment information
                   appendInfo.firstOffset = appendInfo.firstOffset.map {
               offsetMetadata =>
                     offsetMetadata.copy(segmentBaseOffset = segment.baseOffset, relativePositionInSegment = segment.size)
                   }
        
                   segment.append(largestOffset = appendInfo.lastOffset,
                     largestTimestamp = appendInfo.maxTimestamp,
                     shallowOffsetOfMaxTimestamp = appendInfo.offsetOfMaxTimestamp,
                     records = validRecords)
        
                   // Increment the log end offset. We do this immediately after the append because a
                   // write to the transaction index below may fail and we want to ensure that the offsets
                   // of future appends still grow monotonically. The resulting transaction index inconsistency
                   // will be cleaned up after the log directory is recovered. Note that the end offset of the
                   // ProducerStateManager will not be updated and the last stable offset will not advance
                   // if the append to the transaction index fails.
                   updateLogEndOffset(appendInfo.lastOffset + 1)
        
                   // update the producer state
                   updatedProducers.values.foreach(producerAppendInfo => producerStateManager.update(producerAppendInfo))
        
                   // update the transaction index with the true last stable offset. The last offset visible
                   // to consumers using READ_COMMITTED will be limited by this value and the high watermark.
                   completedTxns.foreach {
               completedTxn =>
                     val lastStableOffset = producerStateManager.lastStableOffset(completedTxn)
                     segment.updateTxnIndex(completedTxn, lastStableOffset)
                     producerStateManager.completeTxn(completedTxn)
                   }
        
                   // always update the last producer id map offset so that the snapshot reflects the current offset
                   // even if there isn't any idempotent data being written
                   producerStateManager.updateMapEndOffset(appendInfo.lastOffset + 1)
        
                   // update the first unstable offset (which is used to compute LSO)
                   maybeIncrementFirstUnstableOffset()
        
                   trace(s"Appended message set with last offset: ${appendInfo.lastOffset}, " +
                     s"first offset: ${appendInfo.firstOffset}, " +
                     s"next offset: ${nextOffsetMetadata.messageOffset}, " +
                     s"and messages: $validRecords")
        
                   if (unflushedMessages >= config.flushInterval) flush()
               }
               appendInfo
             }
           }
         }
        }

#### 3.2 异常恢复的处理 ####

1.  异常恢复的核心功能在分区 Fetcher 线程启动时就开始工作，入口在 `AbstractFetcherThread.scala#doWork()` 方法中，可以看到此处与本文相关的部分是 `AbstractFetcherThread.scala#maybeTruncate()` 方法调用
    
        override def doWork(): Unit = {
              
         maybeTruncate()
         maybeFetch()
        }
2.  `AbstractFetcherThread.scala#maybeTruncate()` 方法的核心逻辑分为如下两个步骤：
    
    > 1.  首先调用 `AbstractFetcherThread.scala#fetchTruncatingPartitions()` 方法根据分区状态来确定分区是否带有版本
    > 2.  如果判断本地日志恢复的处理需依赖 LeaderEpoch，则调用 `AbstractFetcherThread.scala#truncateToEpochEndOffsets()` 方法进行
    
        private def maybeTruncate(): Unit = {
              
         val (partitionsWithEpochs, partitionsWithoutEpochs) = fetchTruncatingPartitions()
         if (partitionsWithEpochs.nonEmpty) {
              
           truncateToEpochEndOffsets(partitionsWithEpochs)
         }
         if (partitionsWithoutEpochs.nonEmpty) {
              
           truncateToHighWatermark(partitionsWithoutEpochs)
         }
        }
3.  `AbstractFetcherThread.scala#fetchTruncatingPartitions()` 方法判断当前分区状态 PartitionFetchState如果处于 `Truncating` 状态，则根据本地日志是否具有分区版本缓存来构建返回值
    
    > `AbstractFetcherThread.scala#partitionFetchState()` 方法会在分区 Fetcher 线程创建的时候为分区设置初始状态
    
        private def fetchTruncatingPartitions(): (Map[TopicPartition, EpochData], Set[TopicPartition]) = inLock(partitionMapLock) {
              
         val partitionsWithEpochs = mutable.Map.empty[TopicPartition, EpochData]
         val partitionsWithoutEpochs = mutable.Set.empty[TopicPartition]
        
         partitionStates.partitionStateMap.forEach {
               (tp, state) =>
           if (state.isTruncating) {
              
             latestEpoch(tp) match {
              
               case Some(epoch) if isOffsetForLeaderEpochSupported =>
                 partitionsWithEpochs += tp -> new EpochData()
                   .setPartition(tp.partition)
                   .setCurrentLeaderEpoch(state.currentLeaderEpoch)
                   .setLeaderEpoch(epoch)
               case _ =>
                 partitionsWithoutEpochs += tp
             }
           }
         }
        
         (partitionsWithEpochs, partitionsWithoutEpochs)
        }
4.  `AbstractFetcherThread.scala#truncateToEpochEndOffsets()` 方法的关键逻辑分为两个部分：
    
    > 1.  首先调用 `AbstractFetcherThread.scala#fetchEpochEndOffsets()` 同步 Leader 副本的 LeaderEpoch 版本信息，以便决定当前副本的 LEO
    > 2.  取得最新的 LeaderEpoch 信息后就确定了截断的位点，调用 `AbstractFetcherThread.scala#maybeTruncateToEpochEndOffsets()` 方法尝试进行截断
    
        /**
         * - Build a leader epoch fetch based on partitions that are in the Truncating phase
         * - Send OffsetsForLeaderEpochRequest, retrieving the latest offset for each partition's
         *   leader epoch. This is the offset the follower should truncate to ensure
         *   accurate log replication.
         * - Finally truncate the logs for partitions in the truncating phase and mark the
         *   truncation complete. Do this within a lock to ensure no leadership changes can
         *   occur during truncation.
         */
        private def truncateToEpochEndOffsets(latestEpochsForPartitions: Map[TopicPartition, EpochData]): Unit = {
              
         val endOffsets = fetchEpochEndOffsets(latestEpochsForPartitions)
         //Ensure we hold a lock during truncation.
         inLock(partitionMapLock) {
              
           //Check no leadership and no leader epoch changes happened whilst we were unlocked, fetching epochs
           val epochEndOffsets = endOffsets.filter {
               case (tp, _) =>
             val curPartitionState = partitionStates.stateValue(tp)
             val partitionEpochRequest = latestEpochsForPartitions.getOrElse(tp, {
              
               throw new IllegalStateException(
                 s"Leader replied with partition $tp not requested in OffsetsForLeaderEpoch request")
             })
             val leaderEpochInRequest = partitionEpochRequest.currentLeaderEpoch
             curPartitionState != null && leaderEpochInRequest == curPartitionState.currentLeaderEpoch
           }
        
           val ResultWithPartitions(fetchOffsets, partitionsWithError) = maybeTruncateToEpochEndOffsets(epochEndOffsets, latestEpochsForPartitions)
           handlePartitionsWithErrors(partitionsWithError, "truncateToEpochEndOffsets")
           updateFetchOffsetAndMaybeMarkTruncationComplete(fetchOffsets)
         }
        }
5.  `AbstractFetcherThread.scala#fetchEpochEndOffsets()` 方法由子类 `ReplicaFetcherThread.scala#fetchEpochEndOffsets()` 实现，核心逻辑比较简单，就是向分区副本 Leader 发送 OffsetsForLeaderEpoch 请求并获取响应
    
        override def fetchEpochEndOffsets(partitions: Map[TopicPartition, EpochData]): Map[TopicPartition, EpochEndOffset] = {
              
        
         if (partitions.isEmpty) {
              
           debug("Skipping leaderEpoch request since all partitions do not have an epoch")
           return Map.empty
         }
        
         val topics = new OffsetForLeaderTopicCollection(partitions.size)
         partitions.forKeyValue {
               (topicPartition, epochData) =>
           var topic = topics.find(topicPartition.topic)
           if (topic == null) {
              
             topic = new OffsetForLeaderTopic().setTopic(topicPartition.topic)
             topics.add(topic)
           }
           topic.partitions.add(epochData)
         }
        
         val epochRequest = OffsetsForLeaderEpochRequest.Builder.forFollower(
           offsetForLeaderEpochRequestVersion, topics, brokerConfig.brokerId)
         debug(s"Sending offset for leader epoch request $epochRequest")
        
         try {
              
           val response = leaderEndpoint.sendRequest(epochRequest)
           val responseBody = response.responseBody.asInstanceOf[OffsetsForLeaderEpochResponse]
           debug(s"Received leaderEpoch response $response")
           responseBody.data.topics.asScala.flatMap {
               offsetForLeaderTopicResult =>
             offsetForLeaderTopicResult.partitions.asScala.map {
               offsetForLeaderPartitionResult =>
               val tp = new TopicPartition(offsetForLeaderTopicResult.topic, offsetForLeaderPartitionResult.partition)
               tp -> offsetForLeaderPartitionResult
             }
           }.toMap
         } catch {
              
           case t: Throwable =>
             warn(s"Error when sending leader epoch request for $partitions", t)
        
             // if we get any unexpected exception, mark all partitions with an error
             val error = Errors.forException(t)
             partitions.map {
               case (tp, _) =>
               tp -> new EpochEndOffset()
                 .setPartition(tp.partition)
                 .setErrorCode(error.code)
             }
         }
        }
6.  `AbstractFetcherThread.scala#maybeTruncateToEpochEndOffsets()` 方法也比较简单，核心逻辑为最终调用子类实现 `ReplicaFetcherThread.scala#truncate()` 进行截断操作
    
        private def maybeTruncateToEpochEndOffsets(fetchedEpochs: Map[TopicPartition, EpochEndOffset],
                                                  latestEpochsForPartitions: Map[TopicPartition, EpochData]): ResultWithPartitions[Map[TopicPartition, OffsetTruncationState]] = {
              
         val fetchOffsets = mutable.HashMap.empty[TopicPartition, OffsetTruncationState]
         val partitionsWithError = mutable.HashSet.empty[TopicPartition]
        
         fetchedEpochs.forKeyValue {
               (tp, leaderEpochOffset) =>
           if (partitionStates.contains(tp)) {
              
             Errors.forCode(leaderEpochOffset.errorCode) match {
              
               case Errors.NONE =>
                 val offsetTruncationState = getOffsetTruncationState(tp, leaderEpochOffset)
                 if (doTruncate(tp, offsetTruncationState))
                   fetchOffsets.put(tp, offsetTruncationState)
        
               case Errors.FENCED_LEADER_EPOCH =>
                 val currentLeaderEpoch = latestEpochsForPartitions.get(tp)
                   .map(epochEndOffset => Int.box(epochEndOffset.currentLeaderEpoch)).asJava
                 if (onPartitionFenced(tp, currentLeaderEpoch))
                   partitionsWithError += tp
        
               case error =>
                 info(s"Retrying leaderEpoch request for partition $tp as the leader reported an error: $error")
                 partitionsWithError += tp
             }
           } else {
              
             // Partitions may have been removed from the fetcher while the thread was waiting for fetch
             // response. Removed partitions are filtered out while holding `partitionMapLock` to ensure that we
             // don't update state for any partition that may have already been migrated to another thread.
             trace(s"Ignoring epoch offsets for partition $tp since it has been removed from this fetcher thread.")
           }
         }
        
         ResultWithPartitions(fetchOffsets, partitionsWithError)
        }
        
        private def doTruncate(topicPartition: TopicPartition, truncationState: OffsetTruncationState): Boolean = {
              
         try {
              
           truncate(topicPartition, truncationState)
           true
         }
         catch {
              
           case e: KafkaStorageException =>
             error(s"Failed to truncate $topicPartition at offset ${truncationState.offset}", e)
             markPartitionFailed(topicPartition)
             false
           case t: Throwable =>
             error(s"Unexpected error occurred during truncation for $topicPartition "
               + s"at offset ${truncationState.offset}", t)
             markPartitionFailed(topicPartition)
             false
         }
        }
7.  `ReplicaFetcherThread.scala#truncate()` 的逻辑非常清晰，就是调用 `Partition.scala#truncateTo()` 进行日志截断操作
    
        override def truncate(tp: TopicPartition, offsetTruncationState: OffsetTruncationState): Unit = {
              
         val partition = replicaMgr.getPartitionOrException(tp)
         val log = partition.localLogOrException
        
         partition.truncateTo(offsetTruncationState.offset, isFuture = false)
        
         if (offsetTruncationState.offset < log.highWatermark)
           warn(s"Truncating $tp to offset ${offsetTruncationState.offset} below high watermark " +
             s"${log.highWatermark}")
        
         // mark the future replica for truncation only when we do last truncation
         if (offsetTruncationState.truncationCompleted)
           replicaMgr.replicaAlterLogDirsManager.markPartitionsForTruncation(brokerConfig.brokerId, tp,
             offsetTruncationState.offset)
        }
8.  `Partition.scala#truncateTo()` 只是个入口转发，此处核心在于 `LogManager.scala#truncateTo()` 方法调用
    
        def truncateTo(offset: Long, isFuture: Boolean): Unit = {
              
         // The read lock is needed to prevent the follower replica from being truncated while ReplicaAlterDirThread
         // is executing maybeReplaceCurrentWithFutureReplica() to replace follower replica with the future replica.
         inReadLock(leaderIsrUpdateLock) {
              
           logManager.truncateTo(Map(topicPartition -> offset), isFuture = isFuture)
         }
        }
9.  `LogManager.scala#truncateTo()` 方法会根据 Leader 的 LeaderEpoch 信息判断是否真的需要进行截断操作，如果需要则调用 `Log.scala#truncateTo()` 方法进行
    
        def truncateTo(partitionOffsets: Map[TopicPartition, Long], isFuture: Boolean): Unit = {
              
         val affectedLogs = ArrayBuffer.empty[Log]
         for ((topicPartition, truncateOffset) <- partitionOffsets) {
              
           val log = {
              
             if (isFuture)
               futureLogs.get(topicPartition)
             else
               currentLogs.get(topicPartition)
           }
           // If the log does not exist, skip it
           if (log != null) {
              
             // May need to abort and pause the cleaning of the log, and resume after truncation is done.
             val needToStopCleaner = truncateOffset < log.activeSegment.baseOffset
             if (needToStopCleaner && !isFuture)
               abortAndPauseCleaning(topicPartition)
             try {
              
               if (log.truncateTo(truncateOffset))
                 affectedLogs += log
               if (needToStopCleaner && !isFuture)
                 maybeTruncateCleanerCheckpointToActiveSegmentBaseOffset(log, topicPartition)
             } finally {
              
               if (needToStopCleaner && !isFuture)
                 resumeCleaning(topicPartition)
             }
           }
         }
        
         for (dir <- affectedLogs.map(_.parentDirFile).distinct) {
              
           checkpointRecoveryOffsetsInDir(dir)
         }
        }
10. `Log.scala#truncateTo()` 真正开始底层日志的阶段操作，此处核心逻辑分为以下几步，至此异常恢复的处理基本结束
    
    > 1.  取当前副本最新的日志片段，调用 `LogSegment.scala#truncateTo()` 进行日志截断，也就是开始操作底层物理日志文件及索引文件
    > 2.  调用 `LeaderEpochFileCache.scala#truncateFromEnd()` 尝试截断本地的 LeaderEpoch，因为如果发生截断的话，这个缓存也可能是过期的
    > 3.  调用 `Log.scala#completeTruncation()` 更新本地 HW
    
        private[kafka] def truncateTo(targetOffset: Long): Boolean = {
              
        maybeHandleIOException(s"Error while truncating log to offset $targetOffset for $topicPartition in dir ${dir.getParent}") {
              
          if (targetOffset < 0)
            throw new IllegalArgumentException(s"Cannot truncate partition $topicPartition to a negative offset (%d).".format(targetOffset))
          if (targetOffset >= logEndOffset) {
              
            info(s"Truncating to $targetOffset has no effect as the largest offset in the log is ${logEndOffset - 1}")
        
            // Always truncate epoch cache since we may have a conflicting epoch entry at the
            // end of the log from the leader. This could happen if this broker was a leader
            // and inserted the first start offset entry, but then failed to append any entries
            // before another leader was elected.
            lock synchronized {
              
              leaderEpochCache.foreach(_.truncateFromEnd(logEndOffset))
            }
        
            false
          } else {
              
            info(s"Truncating to offset $targetOffset")
            lock synchronized {
              
              checkIfMemoryMappedBufferClosed()
              if (segments.firstSegment.get.baseOffset > targetOffset) {
              
                truncateFullyAndStartAt(targetOffset)
              } else {
              
                val deletable = logSegments.filter(segment => segment.baseOffset > targetOffset)
                removeAndDeleteSegments(deletable, asyncDelete = true, LogTruncation)
                activeSegment.truncateTo(targetOffset)
                leaderEpochCache.foreach(_.truncateFromEnd(targetOffset))
        
                completeTruncation(
                  startOffset = math.min(targetOffset, logStartOffset),
                  endOffset = targetOffset
                )
              }
              true
            }
          }
         }
        }
        
        private def completeTruncation(
        startOffset: Long,
        endOffset: Long
        ): Unit = {
              
         logStartOffset = startOffset
         nextOffsetMetadata = LogOffsetMetadata(endOffset, activeSegment.baseOffset, activeSegment.size)
         recoveryPoint = math.min(recoveryPoint, endOffset)
         rebuildProducerState(endOffset, producerStateManager)
         updateHighWatermark(math.min(highWatermark, endOffset))
        }

[Kafka 3.0 _10_-Kafka]: https://blog.csdn.net/weixin_45505313/article/details/124049923#1__9
[b4c9d0e7dfbc410d83b592ed6d7a9239.jpeg_pic_center]: https://img-blog.csdnimg.cn/b4c9d0e7dfbc410d83b592ed6d7a9239.jpeg#pic_center
[50ab6359affa4370a734df5c67a9432e.jpeg_pic_center]: https://img-blog.csdnimg.cn/50ab6359affa4370a734df5c67a9432e.jpeg#pic_center
[7673dd1aa8544967b54085a95ac092ad.jpeg_pic_center]: https://img-blog.csdnimg.cn/7673dd1aa8544967b54085a95ac092ad.jpeg#pic_center
[d5f3e96eb157406db4b7897bb6a84275.jpeg_pic_center]: https://img-blog.csdnimg.cn/d5f3e96eb157406db4b7897bb6a84275.jpeg#pic_center
[4aead4c354884cdcb560fc014702113b.jpeg_pic_center]: https://img-blog.csdnimg.cn/4aead4c354884cdcb560fc014702113b.jpeg#pic_center
[Kafka 3.0 _10_-Kafka 1]: https://blog.csdn.net/weixin_45505313/article/details/124049923#22__195