Kafka控制器-分区状态机（PartitionStateMachine）

/**
 * 初始化所有分区的状态（从ZK获取）, 然后对于 New/Offline 触发选主（选主成功后, 变为 OnlinePartition）
 */
def startup() {
  info("Initializing partition state")
  /** 初始化分区的状态, 如果leader所在的broker是存活的, 那么状态为OnlinePartition; 否则为 OfflinePartition */
  initializePartitionState()
  info("Triggering online partition state changes")
  /** 为所有处理 NewPartition 或 OnlinePartition 状态的分区选举 leader */
  triggerOnlinePartitionStateChange()
  debug(s"Started partition state machine with initial state -> ${controllerContext.partitionStates}")
}

/**
 * 根据从ZK获取的所有分区, 进行状态初始化
 */
private def initializePartitionState() {
  for (topicPartition <- controllerContext.allPartitions) {
    /** 检查分区是否有LeaderAndIsr信息, 如果没有则设置为New */
    controllerContext.partitionLeadershipInfo.get(topicPartition) match {
      case Some(currentLeaderIsrAndEpoch) =>
        if (controllerContext.isReplicaOnline(currentLeaderIsrAndEpoch.leaderAndIsr.leader, topicPartition))
          /** 有LeaderAndIsr信息, 且leader存活, 设置为OnlinePartition */
          controllerContext.putPartitionState(topicPartition, OnlinePartition)
        else
          /** 有LeaderAndIsr信息, 但leader不存活, 设置为OfflinePartition */
          controllerContext.putPartitionState(topicPartition, OfflinePartition)
      case None =>
        /** 没有LeaderAndIsr信息, 设置为NewPartition（表示分区不存在）*/
        controllerContext.putPartitionState(topicPartition, NewPartition)
    }
  }
}

/**
 * 这个方法是在控制器选举后或broker上线/下线时触发的
 */
def triggerOnlinePartitionStateChange(): Unit = {
  val partitions = controllerContext.partitionsInStates(Set(OfflinePartition, NewPartition))
  triggerOnlineStateChangeForPartitions(partitions)
}

private def triggerOnlineStateChangeForPartitions(partitions: collection.Set[TopicPartition]): Unit = {
  /** 过滤掉将要被删除的分区，只保留状态为NewPartition/OfflinePartition的分区 */
  val partitionsToTrigger = partitions.filter { partition =>
    !controllerContext.isTopicQueuedUpForDeletion(partition.topic)
  }.toSeq

  handleStateChanges(partitionsToTrigger, OnlinePartition, Some(OfflinePartitionLeaderElectionStrategy(false)))
}

/**
 * 分区状态机改变处理方法, 可以处理多个分区的状态转换, 这样就可以采用批量方式发送请求给多个Broker
 * 用partitionLeaderElectionStrategyOpt指定的策略去选举Leader
 */
override def handleStateChanges(partitions: Seq[TopicPartition], targetState: PartitionState,
  partitionLeaderElectionStrategyOpt: Option[PartitionLeaderElectionStrategy]
): Map[TopicPartition, Either[Throwable, LeaderAndIsr]] = {
  if (partitions.nonEmpty) {
    try {
      controllerBrokerRequestBatch.newBatch()
      /** 为处在OfflinePartition/NewPartition状态的分区选Leader, 完成后转换为OnlinePartition */
      val result = doHandleStateChanges(
        partitions,
        targetState,
        partitionLeaderElectionStrategyOpt
      )
      /** 发送请求给所有broker, 包括LeaderAndIsr请求和UpdateMetadata请求 */
      controllerBrokerRequestBatch.sendRequestsToBrokers(controllerContext.epoch)
      result
    } catch {
    }
  } else {
    Map.empty
  }
}

/** 如果分区的状态不存在, 设置为NonExistentPartition */
case NewPartition =>
  validPartitions.foreach { partition =>
    stateChangeLog.trace(s"Changed partition $partition state from ${partitionState(partition)} to $targetState with " +
      s"assigned replicas ${controllerContext.partitionReplicaAssignment(partition).mkString(",")}")
    controllerContext.putPartitionState(partition, NewPartition)
  }
  Map.empty

case OnlinePartition =>
  /** 获取NewPartition状态的分区集合 */
  val uninitializedPartitions = validPartitions.filter(partition => partitionState(partition) == NewPartition)
  /** 获取OfflinePartition|OnlinePartition的分区集合 */
  val partitionsToElectLeader = validPartitions.filter(partition => partitionState(partition) == OfflinePartition || partitionState(partition) == OnlinePartition)
  if (uninitializedPartitions.nonEmpty) {
    /** 选举leader, 更新到ZK和控制器上下文中, 如果没有存活的副本, 则抛出异常 */
    val successfulInitializations = initializeLeaderAndIsrForPartitions(uninitializedPartitions)
    successfulInitializations.foreach { partition =>
      stateChangeLog.trace(s"Changed partition $partition from ${partitionState(partition)} to $targetState with state " +
        s"${controllerContext.partitionLeadershipInfo(partition).leaderAndIsr}")
      controllerContext.putPartitionState(partition, OnlinePartition)
    }
  }
  if (partitionsToElectLeader.nonEmpty) {
    /** 进行leader选举, 更新到ZK及控制器上下文中, 失败抛出异常 */
    val electionResults = electLeaderForPartitions(
      partitionsToElectLeader,
      partitionLeaderElectionStrategyOpt.getOrElse(
        throw new IllegalArgumentException("Election strategy is a required field when the target state is OnlinePartition")
      )
    )

    electionResults.foreach {
      case (partition, Right(leaderAndIsr)) =>
        stateChangeLog.trace(
          s"Changed partition $partition from ${partitionState(partition)} to $targetState with state $leaderAndIsr"
        )
        controllerContext.putPartitionState(partition, OnlinePartition)
      case (_, Left(_)) => // Ignore; no need to update partition state on election error
    }

    electionResults
  } else {
    Map.empty
  }

case OfflinePartition =>
  validPartitions.foreach { partition =>
    stateChangeLog.trace(s"Changed partition $partition state from ${partitionState(partition)} to $targetState")
    controllerContext.putPartitionState(partition, OfflinePartition)
  }
  Map.empty

case NonExistentPartition =>
  validPartitions.foreach { partition =>
    stateChangeLog.trace(s"Changed partition $partition state from ${partitionState(partition)} to $targetState")
    controllerContext.putPartitionState(partition, NonExistentPartition)
  }
  Map.empty