private def doElectLeaderForPartitions(partitions: Seq[TopicPartition], partitionLeaderElectionStrategy: PartitionLeaderElectionStrategy
  ): (Map[TopicPartition, Either[Exception, LeaderAndIsr]], Seq[TopicPartition]) = {
    
    // 请求 Zookeeper 获取 partition 当前状态  
    val getDataResponses = try {
      zkClient.getTopicPartitionStatesRaw(partitions)
    } catch {
      case e: Exception =>
        return (partitions.iterator.map(_ -> Left(e)).toMap, Seq.empty)
    }

    val failedElections = mutable.Map.empty[TopicPartition, Either[Exception, LeaderAndIsr]]
    val validLeaderAndIsrs = mutable.Buffer.empty[(TopicPartition, LeaderAndIsr)]

    getDataResponses.foreach { getDataResponse =>
      val partition = getDataResponse.ctx.get.asInstanceOf[TopicPartition]
      val currState = partitionState(partition)
      if (getDataResponse.resultCode == Code.OK) {
        // 剔除状态已失效或不存在的 partition
        TopicPartitionStateZNode.decode(getDataResponse.data, getDataResponse.stat) match {
          case Some(leaderIsrAndControllerEpoch) =>
            if (leaderIsrAndControllerEpoch.controllerEpoch > controllerContext.epoch) {
              val failMsg = s"Aborted leader election for partition $partition since the LeaderAndIsr path was " +
                s"already written by another controller. This probably means that the current controller $controllerId went through " +
                s"a soft failure and another controller was elected with epoch ${leaderIsrAndControllerEpoch.controllerEpoch}."
              failedElections.put(partition, Left(new StateChangeFailedException(failMsg)))
            } else { 
              validLeaderAndIsrs += partition -> leaderIsrAndControllerEpoch.leaderAndIsr
            }

          case None =>
            val exception = new StateChangeFailedException(s"LeaderAndIsr information doesn't exist for partition $partition in $currState state")
            failedElections.put(partition, Left(exception))
        }

      } else if (getDataResponse.resultCode == Code.NONODE) {
        val exception = new StateChangeFailedException(s"LeaderAndIsr information doesn't exist for partition $partition in $currState state")
        failedElections.put(partition, Left(exception))
      } else {
        failedElections.put(partition, Left(getDataResponse.resultException.get))
      }
    }

    // 如果全部 partition 均失效，则跳过此次选举
    if (validLeaderAndIsrs.isEmpty) {
      return (failedElections.toMap, Seq.empty)
    }

    // 根据指定的选举策略选择 partition leader
    val (partitionsWithoutLeaders, partitionsWithLeaders) = partitionLeaderElectionStrategy match {
      // Elect leaders for new or offline partitions.
      case OfflinePartitionLeaderElectionStrategy(allowUnclean) =>
        val partitionsWithUncleanLeaderElectionState = collectUncleanLeaderElectionState(validLeaderAndIsrs, allowUnclean)
        leaderForOffline(controllerContext, partitionsWithUncleanLeaderElectionState).partition(_.leaderAndIsr.isEmpty)
      // Elect leaders for partitions that are undergoing reassignment.
      case ReassignPartitionLeaderElectionStrategy =>
        leaderForReassign(controllerContext, validLeaderAndIsrs).partition(_.leaderAndIsr.isEmpty)
      // Elect preferred leaders.
      case PreferredReplicaPartitionLeaderElectionStrategy =>
        leaderForPreferredReplica(controllerContext, validLeaderAndIsrs).partition(_.leaderAndIsr.isEmpty)
      // Elect leaders for partitions whose current leaders are shutting down.
      case ControlledShutdownPartitionLeaderElectionStrategy =>
        leaderForControlledShutdown(controllerContext, validLeaderAndIsrs).partition(_.leaderAndIsr.isEmpty)
    }

    partitionsWithoutLeaders.foreach { electionResult =>
      val partition = electionResult.topicPartition
      val failMsg = s"Failed to elect leader for partition $partition under strategy $partitionLeaderElectionStrategy"
      failedElections.put(partition, Left(new StateChangeFailedException(failMsg)))
    }

    // 将选举结果同步到 TopicPartitionStateZNode 对应的 Zookeeper 节点
    val recipientsPerPartition = partitionsWithLeaders.map(result => result.topicPartition -> result.liveReplicas).toMap
    val adjustedLeaderAndIsrs = partitionsWithLeaders.map(result => result.topicPartition -> result.leaderAndIsr.get).toMap
    val UpdateLeaderAndIsrResult(finishedUpdates, updatesToRetry) = zkClient.updateLeaderAndIsr(adjustedLeaderAndIsrs, controllerContext.epoch, controllerContext.epochZkVersion)

    finishedUpdates.forKeyValue { (partition, result) =>
      result.foreach { leaderAndIsr =>
        val replicaAssignment = controllerContext.partitionFullReplicaAssignment(partition)
        val leaderIsrAndControllerEpoch = LeaderIsrAndControllerEpoch(leaderAndIsr, controllerContext.epoch)
        controllerContext.putPartitionLeadershipInfo(partition, leaderIsrAndControllerEpoch)
        controllerBrokerRequestBatch.addLeaderAndIsrRequestForBrokers(recipientsPerPartition(partition), partition,
          leaderIsrAndControllerEpoch, replicaAssignment, isNew = false)
      }
    }

    (finishedUpdates ++ failedElections, updatesToRetry)
  }

日志复制

复制流程

日志复制中的一些重要偏移概念：

• 起始位移base offset：副本所含第一条消息的 offset
• 高水位值high watermark：副本最新一条己提交消息的 offset
• 日志末端位移log end offset：副本中下一条待写入消息的 offset

每个副本会同时维护 HW 与 LEO 值：

• leader 保证只有 HW 及其之前的消息，才对消费者是可见的。
• follower 宕机后重启时会对其日志截断，只保留 HW 及其之前的日志消息（新版本有改动）。

Kafka 中的复制流程大致如下：

• leader 会将接收到的消息写入日志文件，同时更新 \(\tiny \textsf{leader LEO}\)
• follower 发送 fetch 请求指定 offset 的消息
• 接收到请求后 leader 会根据 offset 更新下面两个值
  • \(\tiny \textsf{follower LEO} = \texttt{offset}\)
  • \(\tiny \textsf{HW} = \min(\textsf{leader LEO}, \min(\textsf{follower LEO of ISR}))\)
• leader 返回消息的同时会附带上最新的 HW
• follower 接收到响应后会将消息写入日志文件，并同时更新 HW

leader epoch

在前面我们提到 follower 在重启后会对日志进行截断，这可能导致消息会丢失：

假设某个分区分布在 A 和 B 两个 broker 上，且最开始时 B 是分区 leader

• 某个时刻，follower A 从 leader B 同步消息 m2，但此时并未收到 HW 更新
• 就在此时，follower A 发生了重启，此时它会截断 m2 所在的日志，然后才向 leader B 重新请求数据
• 不巧，此时 leader B 也发生了宕机，此时 follower A 会被选为新的 leader A，这意味着消息 m2 已经永久丢失了

为了解决这一问题，Kafka 为每一届 leader 分配了一个唯一的 epoch，由其追加到日志的消息都会包含这个 epoch。

然后每个副本都在本地维护一个 epoch 快照文件，并在其中保存 (epoch, offset)：

• epoch 表示 leader 的版本号，当 leader 变更一次 epoch 就会加 1
• offset 则对应 epoch 版本的 leader 写入第一条消息的对于的位移

回到之前的场景，增加了 leader epoch 之后的行为如下：

• follower A 发生重启后，会向 leader B 发送 LeaderEpochRequest 请求最新的 leader epoch
• leader B 会在响应中返回自己的 LEO
• follower A 接收到响应后发现无需对日志进行截断，从而避免了消息 m2 丢失

更多的细节可以参考这篇文章。

文件格式

创建主题时，Kafka 会为主题的每个分区在文件系统中创建了一个对应的子目录，命名格式为主题名-分区号，每个日志子目录的文件构成如下：

[lhop@localhost log]$ tree my-topic-*
my-topic-0
├── 00000000000050209130.index
├── 00000000000050209130.log
├── 00000000000050209130.snapshot
├── 00000000000050209130.timeindex
└── leader-epoch-checkpoint
my-topic-1
├── 00000000000048329826.index
├── 00000000000048329826.log
├── 00000000000048329826.timeindex
└── leader-epoch-checkpoint