主题

private def assignReplicasToBrokersRackUnaware(
    nPartitions: Int,		//分区数 
	replicationFactor: Int, //副本因子 
	brokerList: Seq[Int],   //集群中broker列表
	fixedStartIndex: Int,   //起始索引，即第一个副本分配的位置，默认值为-1
	startPartitionId: Int): Map[Int, Seq[Int]] = //起始分区编号，默认值为-1
{
    // ret表示 <partition,Seq[replica所在brokerId]> 的关系
    val ret = mutable.MapInt, Seq[Int]//保存分配结果的集合
    val brokerArray = brokerList.toArray //brokerId的列表
    // 如果起始索引fixedStartIndex小于0，则根据broker列表长度随机生成一个，以此来保证是有效的brokerId
    val startIndex = if (fixedStartIndex >= 0) fixedStartIndex else rand.nextInt(brokerArray.length)
    // 确保起始分区号不小于0
    var currentPartitionId = math.max(0, startPartitionId)
    // 指定了副本的间隔，目的是为了更均匀地将副本分配到不同的broker上
    var nextReplicaShift = if (fixedStartIndex >= 0) fixedStartIndex else rand.nextInt(brokerArray.length)
    // 轮询所有分区，将每个分区的副本分配到不同的broker上
    for (_ <- 0 until nPartitions) {
        //只有分区编号大于0且刚好分区编号已经轮流一遍broker时才递增下一个副本的间隔
        if (currentPartitionId > 0 && (currentPartitionId % brokerArray.length == 0))
            nextReplicaShift += 1
        val firstReplicaIndex = (currentPartitionId + startIndex) % brokerArray.length
        val replicaBuffer = mutable.ArrayBuffer(brokerArray(firstReplicaIndex))
        // 保存该分区所有副本分配的broker集合
        for (j <- 0 until replicationFactor - 1)
            // 为其余的副本分配broker
            replicaBuffer += brokerArray(replicaIndex(firstReplicaIndex, nextReplicaShift, j, brokerArray.length))
        // 保存该分区所有副本的分配信息
        ret.put(currentPartitionId, replicaBuffer)
        // 继续为下一个分区分配副本
        currentPartitionId += 1
    }
    ret
}
// 该方法是基于第一个副本分配的broker位置，再根据偏移量计算出后续副本被分配到的broker位置。
private def replicaIndex(
    firstReplicaIndex : Int, 
    secondReplicaShift : Int,
	replicaIndex : Int, 
    nBrokers : Int) : Int = 
{
    val shift = 1 + (secondReplicaShift + replicaIndex ) % ( nBrokers - 1 )
    (firstReplicaIndex + shift) % nBrokers
}

private def assignReplicasToBrokersRackAware(
    nPartitions: Int,
	replicationFactor: Int,
	brokerMetadatas: Seq[BrokerMetadata],
	fixedStartIndex: Int,
	startPartitionId: Int): Map[Int, Seq[Int]] = 
{
    val brokerRackMap = brokerMetadatas.collect { 
        case BrokerMetadata(id, Some(rack)) => id -> rack
    }.toMap
    // 统计机架个数
    val numRacks = brokerRackMap.values.toSet.size
    // 基于机架信息生成一个Broker列表，不同机架上的Broker交替出现
    val arrangedBrokerList = getRackAlternatedBrokerList(brokerRackMap)
    // 统计broker个数
    val numBrokers = arrangedBrokerList.size
    val ret = mutable.Map[Int, Seq[Int]]()
    val startIndex = if (fixedStartIndex >= 0) fixedStartIndex else rand.nextInt(arrangedBrokerList.size)
    var currentPartitionId = math.max(0, startPartitionId)
    var nextReplicaShift = if (fixedStartIndex >= 0) fixedStartIndex else rand.nextInt(arrangedBrokerList.size)
    for (_ <- 0 until nPartitions) {
        if (currentPartitionId > 0 && (currentPartitionId % arrangedBrokerList.size == 0))
            nextReplicaShift += 1
        val firstReplicaIndex = (currentPartitionId + startIndex) % arrangedBrokerList.size
        // 第一个副本所在的broker即leader副本的broker
        val leader = arrangedBrokerList(firstReplicaIndex)
        // 每个分区的副本分配列表
        val replicaBuffer = mutable.ArrayBuffer(leader)
        // 每个分区中所分配的机架的列表集
        val racksWithReplicas = mutable.Set(brokerRackMap(leader))
        // 每个分区所分配的brokerId的列表集，和racksWithReplicas一起用来做一层筛选处理
        val brokersWithReplicas = mutable.Set(leader)
        var k = 0
        for (_ <- 0 until replicationFactor - 1) {
            var done = false
            while (!done) {
                val broker = arrangedBrokerList(replicaIndex(firstReplicaIndex, nextReplicaShift * numRacks, k, arrangedBrokerList.size))
                val rack = brokerRackMap(broker)
                if ((!racksWithReplicas.contains(rack) || racksWithReplicas.size == numRacks) && (!brokersWithReplicas.contains(broker) || brokersWithReplicas.size == numBrokers))    {
                    replicaBuffer += broker
                    racksWithReplicas += rack
                    brokersWithReplicas += broker
                    done = true
                }
                k += 1
            }
        }
        ret.put(currentPartitionId, replicaBuffer)
        currentPartitionId += 1
    }
    ret
}
// 其中
private def replicaIndex(
    firstReplicaIndex : Int, 
    secondReplicaShift : Int,
	replicaIndex : Int, nBrokers : Int) : Int = 
{
    val shift = 1 + (secondReplicaShift + replicaIndex ) % ( nBrokers - 1 )
    (firstReplicaIndex + shift) % nBrokers
}