ControllerBrokerRequestBatch分析

为了提高KafkaController Leader和集群其他broker的通信效率，实现批量发送请求的功能

leaderAndIsrRequestMap：保存了发往指定broker的LeaderAndIsrRequest请求相关的信息

stopReplicaRequestMap：保存了发往指定broker的StopReplicaRequest请求相关的信息

updateMetadataRequestMap：保存了发往指定broker的UpdateMetadataCacheRequest请求相关的信息

// 将保存相关请求的集合清空
def clear() {
leaderAndIsrRequestMap.clear()
stopReplicaRequestMap.clear()
updateMetadataRequestMap.clear()
}
// 往LeaderAndIsrRequest请求的map添加相关信息
def addLeaderAndIsrRequestForBrokers(brokerIds: Seq[Int], topic: String, partition: Int,
leaderIsrAndControllerEpoch: LeaderIsrAndControllerEpoch,
replicas: Seq[Int], callback: AbstractRequestResponse => Unit = null) {
val topicPartition = new TopicPartition(topic, partition)

brokerIds.filter(_ >= 0).foreach { brokerId =>
val result = leaderAndIsrRequestMap.getOrElseUpdate(brokerId, mutable.Map.empty)
result.put(topicPartition, PartitionStateInfo(leaderIsrAndControllerEpoch, replicas.toSet))
}

addUpdateMetadataRequestForBrokers(controllerContext.liveOrShuttingDownBrokerIds.toSeq,
                                     Set(TopicAndPartition(topic, partition)))
}
// 往StopReplicaRequest请求的map添加相关信息
def addStopReplicaRequestForBrokers(brokerIds: Seq[Int], topic: String, partition: Int, deletePartition: Boolean,
                                    callback: (AbstractRequestResponse, Int) => Unit = null) {
brokerIds.filter(b => b >= 0).foreach { brokerId =>
    stopReplicaRequestMap.getOrElseUpdate(brokerId, Seq.empty[StopReplicaRequestInfo])
    val v = stopReplicaRequestMap(brokerId)
    if(callback != null)
      stopReplicaRequestMap(brokerId) = v :+ StopReplicaRequestInfo(PartitionAndReplica(topic, partition, brokerId),
        deletePartition, (r: AbstractRequestResponse) => callback(r, brokerId))
    else
      stopReplicaRequestMap(brokerId) = v :+ StopReplicaRequestInfo(PartitionAndReplica(topic, partition, brokerId),
        deletePartition)
}
}

// 往UpdateMetadataRequest请求的map添加相关信息
def addUpdateMetadataRequestForBrokers(brokerIds: Seq[Int],
                                       partitions: collection.Set[TopicAndPartition] = Set.empty[TopicAndPartition],
                                       callback: AbstractRequestResponse => Unit = null) {
// 定义回调函数
def updateMetadataRequestMapFor(partition: TopicAndPartition, beingDeleted: Boolean) {
    // 找到controllerContext中保存的该分区的leader
    val leaderIsrAndControllerEpochOpt = controllerContext.partitionLeadershipInfo.get(partition)
    leaderIsrAndControllerEpochOptmatch {
      case Some(leaderIsrAndControllerEpoch) =>
        // 获取该分区的AR集合
        val replicas = controllerContext.partitionReplicaAssignment(partition).toSet
        val partitionStateInfo = if (beingDeleted) {
          val leaderAndIsr = new LeaderAndIsr(LeaderAndIsr.LeaderDuringDelete, leaderIsrAndControllerEpoch.leaderAndIsr.isr)
          PartitionStateInfo(LeaderIsrAndControllerEpoch(leaderAndIsr, leaderIsrAndControllerEpoch.controllerEpoch), replicas)
        } else {
          PartitionStateInfo(leaderIsrAndControllerEpoch, replicas)
        }
        // 向updateMetadataCacaheRequestMap添加数据
        brokerIds.filter(b => b >= 0).foreach { brokerId =>
          updateMetadataRequestMap.getOrElseUpdate(brokerId, mutable.Map.empty[TopicPartition, PartitionStateInfo])
          updateMetadataRequestMap(brokerId).put(new TopicPartition(partition.topic, partition.partition), partitionStateInfo)
        }
      case None =>
        info("Leader not yet assigned for partition %s. Skip sendingUpdateMetadataRequest.".format(partition))
    }
}
// 如果指定的分区集合为空，则需要全部更新分区
val filteredPartitions = {
    // 如果指定的分区集合为空
    val givenPartitions = if (partitions.isEmpty)
      // 返回获取所有分区的leader信息
      controllerContext.partitionLeadershipInfo.keySet
    else
      partitions // 直接返回分区
    // 过滤即将被删除的partition
    if (controller.deleteTopicManager.partitionsToBeDeleted.isEmpty)
      givenPartitions
    else
      givenPartitions-- controller.deleteTopicManager.partitionsToBeDeleted
}
// 将过滤后的分区信息添加到updateMetadataCacheRequestMap集合，等待发送
if (filteredPartitions.isEmpty)
    brokerIds.filter(b => b >= 0).foreach { brokerId =>
      updateMetadataRequestMap.getOrElseUpdate(brokerId, mutable.Map.empty[TopicPartition, PartitionStateInfo])
    }
else
    filteredPartitions.foreach(partition => updateMetadataRequestMapFor(partition, beingDeleted = false))

controller.deleteTopicManager.partitionsToBeDeleted.foreach(partition => updateMetadataRequestMapFor(partition, beingDeleted = true))
}

def sendRequestsToBrokers(controllerEpoch: Int) {try {// 首先处理leaderAndIsrRequestMapleaderAndIsrRequestMap.foreach { case (broker, partitionStateInfos) =>partitionStateInfos.foreach { case (topicPartition, state) =>val typeOfRequest = if (broker == state.leaderIsrAndControllerEpoch.leaderAndIsr.leader) "become-leader" else "become-follower"stateChangeLogger.trace(("Controller %d epoch %d sending %s LeaderAndIsr request %s to broker %d " +"for partition [%s,%d]").format(controllerId, controllerEpoch, typeOfRequest,state.leaderIsrAndControllerEpoch, broker,topicPartition.topic, topicPartition.partition))}val leaderIds = partitionStateInfos.map(_._2.leaderIsrAndControllerEpoch.leaderAndIsr.leader).toSetval leaders = controllerContext.liveOrShuttingDownBrokers.filter(b => leaderIds.contains(b.id)).map {_.getNode(controller.config.interBrokerSecurityProtocol)}val partitionStates = partitionStateInfos.map { case (topicPartition, partitionStateInfo) =>val LeaderIsrAndControllerEpoch(leaderIsr, controllerEpoch) = partitionStateInfo.leaderIsrAndControllerEpochval partitionState = new requests.PartitionState(controllerEpoch, leaderIsr.leader,leaderIsr.leaderEpoch, leaderIsr.isr.map(Integer.valueOf).asJava, leaderIsr.zkVersion,partitionStateInfo.allReplicas.map(Integer.valueOf).asJava)topicPartition -> partitionState}// 创建LeaderAndIsrRequest对象val leaderAndIsrRequest = new LeaderAndIsrRequest(controllerId, controllerEpoch, partitionStates.asJava, leaders.asJava)// 添加到队列等到被发送controller.sendRequest(broker, ApiKeys.LEADER_AND_ISR, None, leaderAndIsrRequest, null)}// 将集合清空leaderAndIsrRequestMap.clear()// 处理updateMetadataRequestMap信息updateMetadataRequestMap.foreach { case (broker, partitionStateInfos) =>partitionStateInfos.foreach(p => stateChangeLogger.trace(("Controller %d epoch %d sending UpdateMetadata request %s " +"to broker %d for partition %s").format(controllerId, controllerEpoch, p._2.leaderIsrAndControllerEpoch,broker, p._1)))val partitionStates = partitionStateInfos.map { case (topicPartition, partitionStateInfo) =>val LeaderIsrAndControllerEpoch(leaderIsr, controllerEpoch) = partitionStateInfo.leaderIsrAndControllerEpochval partitionState = new requests.PartitionState(controllerEpoch, leaderIsr.leader,leaderIsr.leaderEpoch, leaderIsr.isr.map(Integer.valueOf).asJava, leaderIsr.zkVersion,partitionStateInfo.allReplicas.map(Integer.valueOf).asJava)topicPartition -> partitionState}val version = if (controller.config.interBrokerProtocolVersion >= KAFKA_0_10_0_IV1) 2: Shortelse if (controller.config.interBrokerProtocolVersion >= KAFKA_0_9_0) 1: Shortelse 0: Short// 创建UpdateMetadataRequestval updateMetadataRequest =if (version == 0) {val liveBrokers = controllerContext.liveOrShuttingDownBrokers.map(_.getNode(SecurityProtocol.PLAINTEXT))new UpdateMetadataRequest(controllerId, controllerEpoch, liveBrokers.asJava, partitionStates.asJava)}else {val liveBrokers = controllerContext.liveOrShuttingDownBrokers.map { broker =>val endPoints = broker.endPoints.map { case (securityProtocol, endPoint) =>securityProtocol -> new UpdateMetadataRequest.EndPoint(endPoint.host, endPoint.port)}new UpdateMetadataRequest.Broker(broker.id, endPoints.asJava, broker.rack.orNull)}new UpdateMetadataRequest(version, controllerId, controllerEpoch, partitionStates.asJava, liveBrokers.asJava)}// 添加到队列等到被发送controller.sendRequest(broker, ApiKeys.UPDATE_METADATA_KEY, Some(version), updateMetadataRequest, null)}// 将集合清空updateMetadataRequestMap.clear()// 处理stopReplicaRequestMap信息stopReplicaRequestMap.foreach { case (broker, replicaInfoList) =>val stopReplicaWithDelete = replicaInfoList.filter(_.deletePartition).map(_.replica).toSetval stopReplicaWithoutDelete = replicaInfoList.filterNot(_.deletePartition).map(_.replica).toSetdebug("The stop replica request (delete = true) sent to broker %d is %s".format(broker, stopReplicaWithDelete.mkString(",")))debug("The stop replica request (delete = false) sent to broker %d is %s".format(broker, stopReplicaWithoutDelete.mkString(",")))replicaInfoList.foreach { r =>// 创建StopReplicaRequest对象val stopReplicaRequest = new StopReplicaRequest(controllerId, controllerEpoch, r.deletePartition,Set(new TopicPartition(r.replica.topic, r.replica.partition)).asJava)// 添加到队列等到被发送controller.sendRequest(broker, ApiKeys.STOP_REPLICA, None, stopReplicaRequest, r.callback)}}// 将集合清空stopReplicaRequestMap.clear()} catch {case e : Throwable => {if (leaderAndIsrRequestMap.nonEmpty) {error("Haven't been able to send leader and isr requests, current state of " +s"the map is $leaderAndIsrRequestMap. Exception message: $e")}if (updateMetadataRequestMap.nonEmpty) {error("Haven't been able to send metadata update requests, current state of " +s"the map is $updateMetadataRequestMap. Exception message: $e")}if (stopReplicaRequestMap.nonEmpty) {error("Haven't been able to send stop replica requests, current state of " +s"the map is $stopReplicaRequestMap. Exception message: $e")}throw new IllegalStateException(e)}}
}

ControllerBrokerRequestBatch分析相关推荐

Kafka源码分析10：副本状态机ReplicaStateMachine详解（图解+秒懂+史上最全）
文章很长,建议收藏起来,慢慢读! Java 高并发发烧友社群:疯狂创客圈奉上以下珍贵的学习资源: 免费赠送经典图书:<Java高并发核心编程(卷1)> 面试必备 + 大厂必备 +涨薪 ...
【Golang源码分析】Go Web常用程序包gorilla/mux的使用与源码简析
目录[阅读时间:约10分钟] 一.概述二.对比: gorilla/mux与net/http DefaultServeMux 三.简单使用四.源码简析 1.NewRouter函数 2.HandleF ...
2022-2028年中国自动驾驶系统行业现状调研分析报告
[报告类型]产业研究 [报告价格]4500起 [出版时间]即时更新(交付时间约3个工作日) [发布机构]智研瞻产业研究院 [报告格式]PDF版本报告介绍了中国自动驾驶系统行业市场行业相关概述.中国自 ...
2022-2028年中国阻尼涂料市场研究及前瞻分析报告
[报告类型]产业研究 [报告价格]4500起 [出版时间]即时更新(交付时间约3个工作日) [发布机构]智研瞻产业研究院 [报告格式]PDF版本报告介绍了中国阻尼涂料行业市场行业相关概述.中国阻尼涂 ...
2021-2028年中国阻燃装饰行业市场需求与投资规划分析报告
[报告类型]产业研究 [报告价格]4500起 [出版时间]即时更新(交付时间约3个工作日) [发布机构]智研瞻产业研究院 [报告格式]PDF版本报告介绍了中国阻燃装饰行业市场行业相关概述.中国阻燃装 ...
2022-2028年全球与中国漂白吸水棉市场研究及前瞻分析报告
[报告类型]产业研究 [报告价格]4500起 [出版时间]即时更新(交付时间约3个工作日) [发布机构]智研瞻产业研究院 [报告格式]PDF版本报告介绍了全球与中国漂白吸水棉行业市场行业相关概述.全 ...
2022-2028年全球与中国青苔清洗剂市场研究及前瞻分析报告
[报告类型]产业研究 [报告价格]4500起 [出版时间]即时更新(交付时间约3个工作日) [发布机构]智研瞻产业研究院 [报告格式]PDF版本报告介绍了全球与中国青苔清洗剂行业市场行业相关概述.全 ...
2022-2028年全球与中国氢碘化物市场智研瞻分析报告
[报告类型]产业研究 [报告价格]4500起 [出版时间]即时更新(交付时间约3个工作日) [发布机构]智研瞻产业研究院 [报告格式]PDF版本报告介绍了全球与中国氢碘化物行业市场行业相关概述.全球 ...
2022-2028年全球与中国人字拖市场研究及前瞻分析报告
[报告类型]产业研究 [报告价格]4500起 [出版时间]即时更新(交付时间约3个工作日) [发布机构]智研瞻产业研究院 [报告格式]PDF版本报告介绍了全球与中国人字拖行业市场行业相关概述.全球与 ...

ControllerBrokerRequestBatch分析

ControllerBrokerRequestBatch分析相关推荐

最新文章

热门文章