Heaven.Blog

def createTopic(zkUtils: ZkUtils, opts: TopicCommandOptions) {

val topic = opts.options.valueOf(opts.topicOpt)

val configs = parseTopicConfigsToBeAdded(opts)

val ifNotExists = opts.options.has(opts.ifNotExistsOpt)

if (Topic.hasCollisionChars(topic))

println(“WARNING: Due to limitations in metric names, topics with a period (‘.’) or underscore (‘_’) could collide. To avoid issues it is best to use either, but not both.”)

try {

if (opts.options.has(opts.replicaAssignmentOpt)) {//note: 指定 replica 的分配,直接向 zk 更新即可

val assignment = parseReplicaAssignment(opts.options.valueOf(opts.replicaAssignmentOpt))

AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkUtils, topic, assignment, configs, update = false)

} else {//note: 未指定 replica 的分配,调用自动分配算法进行分配

CommandLineUtils.checkRequiredArgs(opts.parser, opts.options, opts.partitionsOpt, opts.replicationFactorOpt)

val partitions = opts.options.valueOf(opts.partitionsOpt).intValue

val replicas = opts.options.valueOf(opts.replicationFactorOpt).intValue

val rackAwareMode = if (opts.options.has(opts.disableRackAware)) RackAwareMode.Disabled

else RackAwareMode.Enforced

AdminUtils.createTopic(zkUtils, topic, partitions, replicas, configs, rackAwareMode)

}

println(“Created topic \”%s\”.”.format(topic))

} catch  {

case e: TopicExistsException => if (!ifNotExists) throw e

}

nPartitions: Int,

replicationFactor: Int,

fixedStartIndex: Int = -1,

startPartitionId: Int = -1): Map[Int, Seq[Int]] = {

if (nPartitions <= 0) // note: 要增加的 partition 数需要大于0

throw new InvalidPartitionsException(“number of partitions must be larger than 0”)

if (replicationFactor <= 0) //note: replicas 应该大于0

throw new InvalidReplicationFactorException(“replication factor must be larger than 0″)

if (replicationFactor > brokerMetadatas.size) //note: replicas 超过了 broker 数

throw new InvalidReplicationFactorException(s”replication factor: $replicationFactor larger than available brokers: ${brokerMetadatas.size}”)

if (brokerMetadatas.forall(_.rack.isEmpty))//note: 没有开启机架感知

assignReplicasToBrokersRackUnaware(nPartitions, replicationFactor, brokerMetadatas.map(_.id), fixedStartIndex,

startPartitionId)

else { //note: 机架感知的情况

if (brokerMetadatas.exists(_.rack.isEmpty)) //note: 并不是所有的机架都有机架感知

throw new AdminOperationException(“Not all brokers have rack information for replica rack aware assignment”)

assignReplicasToBrokersRackAware(nPartitions, replicationFactor, brokerMetadatas, fixedStartIndex,

startPartitionId)

}

}

replicationFactor: Int,

brokerList: Seq[Int],

fixedStartIndex: Int,

startPartitionId: Int): Map[Int, Seq[Int]] = {

val ret = mutable.Map[Int, Seq[Int]]()

val brokerArray = brokerList.toArray

val startIndex = if (fixedStartIndex >= 0) fixedStartIndex else rand.nextInt(brokerArray.length) //note: 随机选择一个Broker

var currentPartitionId = math.max(0, startPartitionId) //note: 开始增加的第一个 partition

var nextReplicaShift = if (fixedStartIndex >= 0) fixedStartIndex else rand.nextInt(brokerArray.length)

for (_ <- 0 until nPartitions) { //note: 对每个 partition 进行分配

if (currentPartitionId > 0 && (currentPartitionId % brokerArray.length == 0))

nextReplicaShift += 1 //note: 防止 partition 过大时,其中某些 partition 的分配（leader、follower）完全一样

val firstReplicaIndex = (currentPartitionId + startIndex) % brokerArray.length //note: partition 的第一个 replica

val replicaBuffer = mutable.ArrayBuffer(brokerArray(firstReplicaIndex))

for (j <- 0 until replicationFactor – 1) //note: 其他 replica 的分配

replicaBuffer += brokerArray(replicaIndex(firstReplicaIndex, nextReplicaShift, j, brokerArray.length))

ret.put(currentPartitionId, replicaBuffer)

currentPartitionId += 1

}

ret

}

//note: 为 partition 设置完第一个 replica 后,其他 replica 分配的计算

private def replicaIndex(firstReplicaIndex: Int, secondReplicaShift: Int, replicaIndex: Int, nBrokers: Int): Int = {

val shift = 1 + (secondReplicaShift + replicaIndex) % (nBrokers – 1)//note: 在 secondReplicaShift 的基础上增加一个 replicaIndex

(firstReplicaIndex + shift) % nBrokers

}

partition 为0时，那第一副本在 (0+0)%5=0，第二个副本在 (0+(1+(0+0)%4)))%5=1，第三副本在 (0+(1+(0+1)%4)))%5=2；

partition 为2时，那第一副本在 (0+2)%5=2，第二个副本在 (2+(1+(0+0)%4)))%5=3，第三副本在 (2+(1+(0+1)%4)))%5=4；

partition 为5时，那第一副本在 (0+5)%5=0，第二个副本在 (0+(1+(1+0)%4)))%5=2，第三副本在 (0+(1+(1+1)%4)))%5=3（partition 数是 Broker 数一倍时，nextReplicaShift 值会增加1）；

partition 为8时，那第一副本在 (0+8)%5=3，第二个副本在 (3+(1+(1+0)%4)))%5=0，第三副本在 (3+(1+(1+1)%4)))%5=1。

info(“New partition creation callback for %s”.format(newPartitions.mkString(“,”)))

partitionStateMachine.handleStateChanges(newPartitions, NewPartition)

replicaStateMachine.handleStateChanges(controllerContext.replicasForPartition(newPartitions), NewReplica)

partitionStateMachine.handleStateChanges(newPartitions, OnlinePartition, offlinePartitionSelector)

replicaStateMachine.handleStateChanges(controllerContext.replicasForPartition(newPartitions), OnlineReplica)

}

//note: 新建一个 partition

assertValidPreviousStates(topicAndPartition, List(NonExistentPartition), NewPartition)

partitionState.put(topicAndPartition, NewPartition) //note: 缓存 partition 的状态

val assignedReplicas = controllerContext.partitionReplicaAssignment(topicAndPartition).mkString(“,”)

stateChangeLogger.trace(“Controller %d epoch %d changed partition %s state from %s to %s with assigned replicas %s”

.format(controllerId, controller.epoch, topicAndPartition, currState, targetState, assignedReplicas))

assertValidPreviousStates(partitionAndReplica, List(NonExistentReplica), targetState)  //note: 验证

// start replica as a follower to the current leader for its partition

val leaderIsrAndControllerEpochOpt = ReplicationUtils.getLeaderIsrAndEpochForPartition(zkUtils, topic, partition)

leaderIsrAndControllerEpochOpt match {

case Some(leaderIsrAndControllerEpoch) =>

if(leaderIsrAndControllerEpoch.leaderAndIsr.leader == replicaId)//note: 这个状态的 Replica 不能作为 leader

throw new StateChangeFailedException(“Replica %d for partition %s cannot be moved to NewReplica”

.format(replicaId, topicAndPartition) + “state as it is being requested to become leader”)

//note: 向所有 replicaId 发送 LeaderAndIsr 请求,这个方法同时也会向所有的 broker 发送 updateMeta 请求

brokerRequestBatch.addLeaderAndIsrRequestForBrokers(List(replicaId),

topic, partition, leaderIsrAndControllerEpoch,

replicaAssignment)

assertValidPreviousStates(topicAndPartition, List(NewPartition, OnlinePartition, OfflinePartition), OnlinePartition)

partitionState(topicAndPartition) match {

case NewPartition =>

// initialize leader and isr path for new partition

initializeLeaderAndIsrForPartition(topicAndPartition) //note: 为新建的 partition 初始化 leader 和 isr

case OfflinePartition =>

electLeaderForPartition(topic, partition, leaderSelector)

case OnlinePartition => // invoked when the leader needs to be re-elected

electLeaderForPartition(topic, partition, leaderSelector)

case _ => // should never come here since illegal previous states are checked above

}