Apache Kafka源码分析

在broker的配置中，auto.leader.rebalance.enable (false)

那么这个leader是如何进行rebalance的？

首先在controller启动的时候会打开一个scheduler，

if (config.autoLeaderRebalanceEnable) { //如果打开outoLeaderRebalance，需要把partiton leader由于dead而发生迁徙的，重新迁徙回去
        info("starting the partition rebalance scheduler")
        autoRebalanceScheduler.startup()
        autoRebalanceScheduler.schedule("partition-rebalance-thread", checkAndTriggerPartitionRebalance,
          5, config.leaderImbalanceCheckIntervalSeconds, TimeUnit.SECONDS)
      }

定期去做,

checkAndTriggerPartitionRebalance

这个函数逻辑，就是找出所有发生过迁移的replica，即

topicsNotInPreferredReplica

并且判断如果满足imbalance比率，即自动触发leader rebalance，将leader迁回perfer replica

关键要理解什么是preferred replicas？

preferredReplicasForTopicsByBrokers =
          controllerContext.partitionReplicaAssignment.filterNot(p => deleteTopicManager.isTopicQueuedUpForDeletion(p._1.topic)).groupBy {
            case(topicAndPartition, assignedReplicas) => assignedReplicas.head
          }

 partitionReplicaAssignment: mutable.Map[TopicAndPartition, Seq[Int]] 

TopicAndPartition可以通过topic name和partition id来唯一标识一个partition，Seq[int],表示brokerids，表明这个partition的replicas在哪些brokers上面

从partition的ReplicaAssignment里面过滤掉delete的topic，然后按照assignedReplicas.head进行groupby，就是按照Seq中的第一个brokerid
意思就是说，默认每个partition的preferred replica就是第一个被assign的replica

groupby的结果就是，每个broker，和应该以该broker作为leader的所有partition，即

case(leaderBroker, topicAndPartitionsForBroker)

那么找出里面当前leader不是preferred的，即发生过迁移的，
很简单，直接和leaderAndIsr里面的leader进行比较，如果不相等就说明发生过迁徙

topicsNotInPreferredReplica =
              topicAndPartitionsForBroker.filter {
                case(topicPartition, replicas) => {
                  controllerContext.partitionLeadershipInfo.contains(topicPartition) &&
                  controllerContext.partitionLeadershipInfo(topicPartition).leaderAndIsr.leader != leaderBroker
                }
              }

并且只有当某个broker上的imbalanceRatio大于10%的时候，才会触发rebalance

imbalanceRatio = totalTopicPartitionsNotLedByBroker.toDouble / totalTopicPartitionsForBroker

对每个partition的迁移过程，
首先preferred的broker要是活着的，并且当前是没有partition正在进行reassign或replica election的，说明这个过程是不能并行的，同时做reassign很容易冲突

// do this check only if the broker is live and there are no partitions being reassigned currently
                  // and preferred replica election is not in progress
                  if (controllerContext.liveBrokerIds.contains(leaderBroker) &&
                      controllerContext.partitionsBeingReassigned.size == 0 &&
                      controllerContext.partitionsUndergoingPreferredReplicaElection.size == 0 &&
                      !deleteTopicManager.isTopicQueuedUpForDeletion(topicPartition.topic) &&
                      controllerContext.allTopics.contains(topicPartition.topic)) {
                    onPreferredReplicaElection(Set(topicPartition), true)

onPreferredReplicaElection

还是通过partitionStateMachine，来改变partition的状态

partitionStateMachine.handleStateChanges(partitions, OnlinePartition, preferredReplicaPartitionLeaderSelector)

partitionStateMachine会另外分析，这里只需要知道，当前partition的状态是，OnlinePartition –> OnlinePartition
并且是以preferredReplicaPartitionLeaderSelector，作为leaderSelector的策略

 

PreferredReplicaPartitionLeaderSelector

策略很简单，就是把leader换成preferred replica

def selectLeader(topicAndPartition: TopicAndPartition, currentLeaderAndIsr: LeaderAndIsr): (LeaderAndIsr, Seq[Int]) = {
    val assignedReplicas = controllerContext.partitionReplicaAssignment(topicAndPartition)
    val preferredReplica = assignedReplicas.head  //取AR第一个replica作为preferred
    // check if preferred replica is the current leader
    val currentLeader = controllerContext.partitionLeadershipInfo(topicAndPartition).leaderAndIsr.leader
    if (currentLeader == preferredReplica) { //如果当前leader就是preferred就不需要做了
      throw new LeaderElectionNotNeededException("Preferred replica %d is already the current leader for partition %s" .format(preferredReplica, topicAndPartition))
    } else {
      info("Current leader %d for partition %s is not the preferred replica.".format(currentLeader, topicAndPartition) + " Trigerring preferred replica leader election")
      // check if preferred replica is not the current leader and is alive and in the isr
      if (controllerContext.liveBrokerIds.contains(preferredReplica) && currentLeaderAndIsr.isr.contains(preferredReplica)) { //判断当前preferred replica所在broker是否活，是否在isr中
        (new LeaderAndIsr(preferredReplica, currentLeaderAndIsr.leaderEpoch + 1, currentLeaderAndIsr.isr, currentLeaderAndIsr.zkVersion + 1), assignedReplicas) //产生新的leaderAndIsr
      } else {
        throw new StateChangeFailedException("Preferred replica %d for partition ".format(preferredReplica) +
          "%s is either not alive or not in the isr. Current leader and ISR: [%s]".format(topicAndPartition, currentLeaderAndIsr))
      }
    }
  }
}