Kafka源码阅读（11）-Controller之元数据与通信

class ControllerContext {
  // 控制器状态相关
  var epoch: Int = KafkaController.InitialControllerEpoch // 当前epoch，用于表示controller的选举轮次
  var epochZkVersion: Int = KafkaController.InitialControllerEpochZkVersion // epoch在zk中的版本号，用于乐观锁控制
  
  // broker相关
  private var liveBrokers: Set[Broker] = Set.empty // 当前活跃的broker集合
  private var liveBrokerEpochs: Map[Int, Long] = Map.empty // 当前活跃broker id到其epoch的映射
  var shuttingDownBrokerIds: mutable.Set[Int] = mutable.Set.empty // 正在关闭的broker id集合
  val replicasOnOfflineDirs: mutable.Map[Int, Set[TopicPartition]] = mutable.Map.empty // broker id到该broker上离线分区的集合
  
  // topic相关
  var allTopics: Set[String] = Set.empty // 所有topic的集合
  val topicsToBeDeleted = mutable.Set.empty[String] // 待删除topic的集合
  val topicsWithDeletionStarted = mutable.Set.empty[String] // 正在被删除的topic集合
  val topicsIneligibleForDeletion = mutable.Set.empty[String] // 不符合删除条件的topic集合
  
  // partition相关
  val partitionAssignments = mutable.Map.empty[String, mutable.Map[Int, ReplicaAssignment]] // 各主题分区副本的分配情况
  val partitionLeadershipInfo = mutable.Map.empty[TopicPartition, LeaderIsrAndControllerEpoch] // 分区leader、ISR及其控制器epoch的信息
  val partitionsBeingReassigned = mutable.Set.empty[TopicPartition] // 正在进行副本重分配的分区集合
  val partitionStates = mutable.Map.empty[TopicPartition, PartitionState] // 分区状态
  val replicaStates = mutable.Map.empty[PartitionAndReplica, ReplicaState] // 分区副本状态
  var offlinePartitionCount = 0 // 离线分区的数量

  // 监控指标相关
  val stats = new ControllerStats
}

case class ReplicaAssignment private (
  replicas: Seq[Int],
  addingReplicas: Seq[Int],
  removingReplicas: Seq[Int]
)

def partitionsOnBroker(brokerId: Int): Set[TopicPartition] = {
  partitionAssignments.flatMap {
    case (topic, topicReplicaAssignment) => topicReplicaAssignment.filter {
      case (_, partitionAssignment) => partitionAssignment.replicas.contains(brokerId)
    }.map {
      case (partition, _) => new TopicPartition(topic, partition)
    }
  }.toSet
}

def partitionsForTopic(topic: String): collection.Set[TopicPartition] = {
  partitionAssignments.getOrElse(topic, mutable.Map.empty).map {
    case (partition, _) => new TopicPartition(topic, partition)
  }.toSet
}

def handle(request: RequestChannel.Request): Unit = {
  try {
    request.header.apiKey match {
      case ApiKeys.LEADER_AND_ISR => handleLeaderAndIsrRequest(request)
      case ApiKeys.STOP_REPLICA => handleStopReplicaRequest(request)
      case ApiKeys.UPDATE_METADATA => handleUpdateMetadataRequest(request)
      
      // ...

public abstract class AbstractControlRequest extends AbstractRequest {
  public static abstract class Builder<T extends AbstractRequest> extends AbstractRequest.Builder<T> {
    protected final int controllerId; // controller 所在的 broker id
    protected final int controllerEpoch; // controller epoch
    protected final long brokerEpoch; // 目标broker的epoch
  }
}

private def isBrokerEpochStale(brokerEpochInRequest: Long): Boolean = {
  if (brokerEpochInRequest == AbstractControlRequest.UNKNOWN_BROKER_EPOCH) false
  else {
    // 如果当前broker的epoch更大，说明这个请求已经过时
    brokerEpochInRequest < controller.brokerEpoch
  }
}

protected val brokerStateInfo = new HashMap[Int, ControllerBrokerStateInfo] // broker id -> ControllerBrokerStateInfo

case class ControllerBrokerStateInfo(
  networkClient: NetworkClient, // 与目标broker通信的网络工具类
  brokerNode: Node, // 目标broker的连接信息
  messageQueue: BlockingQueue[QueueItem], // 请求队列
  requestSendThread: RequestSendThread, // 请求处理线程
  queueSizeGauge: Gauge[Int],
  requestRateAndTimeMetrics: Timer,
  reconfigurableChannelBuilder: Option[Reconfigurable]
)

// 启动ControllerChannelManager
// 其实就是创建ControllerBrokerStateInfo，然后启动所有请求处理线程
def startup() = {
  controllerContext.liveOrShuttingDownBrokers.foreach(addNewBroker)

  brokerLock synchronized {
    brokerStateInfo.foreach(brokerState => startRequestSendThread(brokerState._1))
  }
}

// 添加broker，创建ControllerBrokerStateInfo
private def addNewBroker(broker: Broker): Unit = {
  // 创建请求队列
  val messageQueue = new LinkedBlockingQueue[QueueItem]
  debug(s"Controller ${config.brokerId} trying to connect to broker ${broker.id}")
  // 创建listener和node
  val controllerToBrokerListenerName = config.controlPlaneListenerName.getOrElse(config.interBrokerListenerName)
  val controllerToBrokerSecurityProtocol = config.controlPlaneSecurityProtocol.getOrElse(config.interBrokerSecurityProtocol)
  val brokerNode = broker.node(controllerToBrokerListenerName)
  val logContext = new LogContext(s"[Controller id=${config.brokerId}, targetBrokerId=${brokerNode.idString}] ")
  // 创建网络通信类
  val (networkClient, reconfigurableChannelBuilder) = {
    val channelBuilder = ChannelBuilders.clientChannelBuilder(
      controllerToBrokerSecurityProtocol,
      JaasContext.Type.SERVER,
      config,
      controllerToBrokerListenerName,
      config.saslMechanismInterBrokerProtocol,
      time,
      config.saslInterBrokerHandshakeRequestEnable,
      logContext
    )
    val reconfigurableChannelBuilder = channelBuilder match {
      case reconfigurable: Reconfigurable =>
        config.addReconfigurable(reconfigurable)
        Some(reconfigurable)
      case _ => None
    }
    // 创建KSelector
    // 配置：响应大小无限制、不关闭空闲连接
    val selector = new Selector(
      NetworkReceive.UNLIMITED,
      Selector.NO_IDLE_TIMEOUT_MS,
      metrics,
      time,
      "controller-channel",
      Map("broker-id" -> brokerNode.idString).asJava,
      false,
      channelBuilder,
      logContext
    )
    // 创建网络通信工具类
    val networkClient = new NetworkClient(
      selector,
      new ManualMetadataUpdater(Seq(brokerNode).asJava),
      config.brokerId.toString,
      1,
      0,
      0,
      Selectable.USE_DEFAULT_BUFFER_SIZE,
      Selectable.USE_DEFAULT_BUFFER_SIZE,
      config.requestTimeoutMs,
      ClientDnsLookup.DEFAULT,
      time,
      false,
      new ApiVersions,
      logContext
    )
    (networkClient, reconfigurableChannelBuilder)
  }
  val threadName = threadNamePrefix match {
    case None => s"Controller-${config.brokerId}-to-broker-${broker.id}-send-thread"
    case Some(name) => s"$name:Controller-${config.brokerId}-to-broker-${broker.id}-send-thread"
  }

  val requestRateAndQueueTimeMetrics = newTimer(
    RequestRateAndQueueTimeMetricName, TimeUnit.MILLISECONDS, TimeUnit.SECONDS, brokerMetricTags(broker.id)
  )

  // 创建请求消费线程
  val requestThread = new RequestSendThread(config.brokerId, controllerContext, messageQueue, networkClient,
    brokerNode, config, time, requestRateAndQueueTimeMetrics, stateChangeLogger, threadName)
  requestThread.setDaemon(false)

  val queueSizeGauge = newGauge(QueueSizeMetricName, () => messageQueue.size, brokerMetricTags(broker.id))

  // 创建ControllerStateInfo
  brokerStateInfo.put(broker.id, ControllerBrokerStateInfo(networkClient, brokerNode, messageQueue,
    requestThread, queueSizeGauge, requestRateAndQueueTimeMetrics, reconfigurableChannelBuilder))
}

// 关闭ControllerChannelManager
def shutdown() = {
  brokerLock synchronized {
    brokerStateInfo.values.toList.foreach(removeExistingBroker)
  }
}

private def removeExistingBroker(brokerState: ControllerBrokerStateInfo): Unit = {
  try {
    brokerState.reconfigurableChannelBuilder.foreach(config.removeReconfigurable)
    // 关闭请求处理线程
    brokerState.requestSendThread.shutdown()
    // 关闭网络通信类
    brokerState.networkClient.close()
    // 清空请求队列
    brokerState.messageQueue.clear()
    removeMetric(QueueSizeMetricName, brokerMetricTags(brokerState.brokerNode.id))
    removeMetric(RequestRateAndQueueTimeMetricName, brokerMetricTags(brokerState.brokerNode.id))
    // 移除ControllerBrokerStateInfo
    brokerStateInfo.remove(brokerState.brokerNode.id)
  } catch {
    case e: Throwable => error("Error while removing broker by the controller", e)
  }
}

def sendRequest(brokerId: Int, request: AbstractControlRequest.Builder[_ <: AbstractControlRequest], callback: AbstractResponse => Unit = null): Unit = {
  brokerLock synchronized {
    // 获取ControllerBrokerStateInfo
    val stateInfoOpt = brokerStateInfo.get(brokerId)
    stateInfoOpt match {
      case Some(stateInfo) =>
      	// 将请求封装在QueueItem中，放入请求队列
        stateInfo.messageQueue.put(QueueItem(request.apiKey, request, callback, time.milliseconds()))
      case None =>
        warn(s"Not sending request $request to broker $brokerId, since it is offline.")
    }
  }
}

class RequestSendThread(
  val controllerId: Int,
  val controllerContext: ControllerContext,
  val queue: BlockingQueue[QueueItem], // 请求队列
  val networkClient: NetworkClient, // 网络通信类
  val brokerNode: Node, // 目标broker的连接信息
  val config: KafkaConfig,
  val time: Time,
  val requestRateAndQueueTimeMetrics: Timer,
  val stateChangeLogger: StateChangeLogger,
  name: String
) extends ShutdownableThread(name = name) {
  // ...
}

override def run(): Unit = {
  isStarted = true
  info("Starting")
  try {
    // 不断执行doWork
    while (isRunning)
      doWork()
  } catch {
    case e: FatalExitError =>
      shutdownInitiated.countDown()
      shutdownComplete.countDown()
      info("Stopped")
      Exit.exit(e.statusCode())
    case e: Throwable =>
      if (isRunning)
        error("Error due to", e)
  } finally {
     shutdownComplete.countDown()
  }
  info("Stopped")
}

override def doWork(): Unit = {

  def backoff(): Unit = pause(100, TimeUnit.MILLISECONDS)

  // 从请求队列取出
  val QueueItem(apiKey, requestBuilder, callback, enqueueTimeMs) = queue.take()
  requestRateAndQueueTimeMetrics.update(time.milliseconds() - enqueueTimeMs, TimeUnit.MILLISECONDS)

  var clientResponse: ClientResponse = null
  try {
    var isSendSuccessful = false
    while (isRunning && !isSendSuccessful) {
      try {
        // 如果broker还没连接上，就等一会再试试
        if (!brokerReady()) {
          isSendSuccessful = false
          backoff()
        }
        // 发送请求
        else {
          val clientRequest = networkClient.newClientRequest(brokerNode.idString, requestBuilder,
            time.milliseconds(), true)
          clientResponse = NetworkClientUtils.sendAndReceive(networkClient, clientRequest, time)
          isSendSuccessful = true
        }
      } catch {
        case e: Throwable => // if the send was not successful, reconnect to broker and resend the message
          warn(s"Controller $controllerId epoch ${controllerContext.epoch} fails to send request $requestBuilder " +
            s"to broker $brokerNode. Reconnecting to broker.", e)
          networkClient.close(brokerNode.idString)
          isSendSuccessful = false
          backoff()
      }
    }
    if (clientResponse != null) {
      val requestHeader = clientResponse.requestHeader
      val api = requestHeader.apiKey
      // 检查响应是否对应请求的类型
      if (api != ApiKeys.LEADER_AND_ISR && api != ApiKeys.STOP_REPLICA && api != ApiKeys.UPDATE_METADATA)
        throw new KafkaException(s"Unexpected apiKey received: $apiKey")

      val response = clientResponse.responseBody

      stateChangeLogger.withControllerEpoch(controllerContext.epoch).trace(s"Received response " +
        s"${response.toString(requestHeader.apiVersion)} for request $api with correlation id " +
        s"${requestHeader.correlationId} sent to broker $brokerNode")

      // 调用回调函数，进一步处理响应
      if (callback != null) {
        callback(response)
      }
    }
  } catch {
    case e: Throwable =>
      error(s"Controller $controllerId fails to send a request to broker $brokerNode", e)
      // If there is any socket error (eg, socket timeout), the connection is no longer usable and needs to be recreated.
      networkClient.close(brokerNode.idString)
  }
}