在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)) } } } }
2015-10-27
