10分钟带你玩转Kafka基于Controller的领导选举！

${KAFKA_PATH}/bin/kafka-run-class.sh kafka.tools.JmxTool --jmx-url service:jmx:rmi:///jndi/rmi://"${BrokerIP}":"${JMXPort}"/jmxrmi --object-name kafka.controller:type=KafkaController,name=ActiveControllerCount --date-format "YYYY-MM-dd_HH:mm" --reporting-interval -1 | grep -v type

function inter_controller_history(){  #第一次检测集群Controller  if [ ! -f "${clusterID}"_controller_history ]; then    awk '/,1$/ {print $0}' "${clusterID}"_controller >> "${clusterID}"_controller_history  #记录Controller变更历史  else     nowController=$(awk '/,1$/ {print $0}' "${clusterID}"_controller | awk -F ',' '{print $1}')     LastTimeController=$(tail -n 1 "${clusterID}"_controller_history | awk '/,1$/ {print $0}'  | awk -F ',' '{print $1}')
     if [ "${nowController}_X" != "${LastTimeController}_X" ];then        awk '/,1$/ {print $0}' "${clusterID}"_controller >> "${clusterID}"_controller_history        msg="${msg_tmp} clusterID:${clusterID} ${ClusterNameCN}  Controller From ${LastTimeController} to ${nowController}"       echo "$msg" >> $log_file_name       send_warning
     fi  fi}

${KAFKA_PATH}/bin/kafka-run-class.sh kafka.tools.JmxTool --jmx-url service:jmx:rmi:///jndi/rmi://${BrokerIP}:"${JMXPort}"/jmxrmi --object-name kafka.controller:type=KafkaController,name=OfflinePartitionsCount  --date-format "YYYY-MM-dd_HH:mm" --reporting-interval -1

${KAFKA_PATH}/bin/kafka-run-class.sh kafka.tools.JmxTool --jmx-url service:jmx:rmi:///jndi/rmi://"${BrokerIP}":"${JMXPort}"/jmxrmi --object-name kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec --date-format "YYYY-MM-dd HH:mm" --attributes Count  --reporting-interval -1

/** * KafkaController#startup中为每一个server都会启动一个eventManager * * 集群首次启动时，Controller 尚未被选举出来。 * 于是，Broker 启动后，首先将 Startup 这 个 ControllerEvent 写入到事件队列中，然后启动对应的事件处理线程和 ControllerChangeHandler ZooKeeper 监听器， * 最后依赖事件处理线程进行 Controller 的选举。 * 在源码中，KafkaController 类的 startup 方法就是做这些事情的。 * 当Broker 启动时，它 会调用这个方法启动 ControllerEventThread 线程。值得注意的是，每个 Broker 都需要 做这些事情， * 不是说只有 Controller 所在的 Broker 才需要执行这些逻辑。 * * 首先，startup 方法会注册 ZooKeeper 状态变更监听器，用于监听 Broker 与 ZooKeeper 之间的会话是否过期。 * 接着，写入 Startup 事件到事件队列，然后启动 ControllerEventThread 线程，开始处理事件队列中的 Startup 事件。 */def startup() = {  /**   * controller组件启动监听器的方式是在zk上面注册一个stateChangeHandler   * 在KafkaController.startup()方法中首先通过zk注册监听事件，监听StateChangeHandler   * registerStateChangeHandler用于session过期后触发重新选举   *   * 第1步：注册Zookeeper状态变更监听器，它是用于监听Zookeeper和broker会话过期的,   */  zkClient.registerStateChangeHandler(new StateChangeHandler {    override val name: String = StateChangeHandlers.ControllerHandler    override def afterInitializingSession(): Unit = {      eventManager.put(RegisterBrokerAndReelect)    }    override def beforeInitializingSession(): Unit = {      val expireEvent = new Expire      eventManager.clearAndPut(expireEvent)
      // Block initialization of the new session until the expiration event is being handled,      // which ensures that all pending events have been processed before creating the new session      /**       * 阻塞等待时间被处理结束，session过期触发重新选举，必须等待选举这个时间完成Controller才能正常工作       */      expireEvent.waitUntilProcessingStarted()    }  })
  /** Startup是一个ControllerEvent，ControllerEventThread会执行它的process方法   * Startup 类型的 ControllerEvent 被放入到 eventmanager中,被 KafkaController#process 方法调用,在下面有定义   * 在Startup的回调方法process()中，首先在zk中监听/controller路径。并且调用elect()进行选举过程。   *   *   case object Startup extends ControllerEvent {   *    def state = ControllerState.ControllerChange   *      override def process(): Unit = {   *          zkClient.registerZNodeChangeHandlerAndCheckExistence(controllerChangeHandler)   *          // elect就是尝试竞选controller   *          elect()   *          }   *    }   *   * 调用elect()，进行选举，在onControllerFailover（）中   * 放入Startup并启动eventManager后台线程开始选举, Startup 是个事件   *   *  put 方法 在 core/src/main/scala/kafka/controller/ControllerEventManager.scala#def put(event: ControllerEvent): Unit = inLock(putLock) {   * 第2步：写入Startup事件到事件队列   */  eventManager.put(Startup)
  /** 启动了ControllerEventManager   * KafkaController#startup 中为每一个 server 都会启动一个 eventManage   * 启动eventManager来处理event queue中的任务   *   * ControllerEventThread在   * core/src/main/scala/kafka/controller/ControllerEventManager.scala中定义   *   * 第3步：内部启动ControllerEventThread线程，开始处理事件队列中的ControllerEvent   */  eventManager.start()}

  /** * elect方法是关于Controller选举的核心方法 * elect就是尝试竞选controller,如果我们当前节点真的被选为controller(onControllerFailover()–故障转移) * */
private def elect(): Unit = {  /** 获取zk /controller节点中的brokerId，没有返回-1   * 第1步：获取当前Controller所在Broker的序列号，如果Controller不存在，显式标记为-1   */  activeControllerId = zkClient.getControllerId.getOrElse(-1)  /**   * 开始创建临时节点前检查，如果/controller节点已经存在，说明已经有broker成为controller,   * 因此本broker直接退出controller选举   * 第2步：如果当前Controller已经选出来了，直接返回即可   */  if (activeControllerId != -1) {    /** 在初始化的时候可能会走到这里，如果当前controller不空，则退出选举     * [2020-11-06 13:27:50,200] INFO [ControllerEventThread controllerId=19] Starting (kafka.controller.ControllerEventManager$ControllerEventThread)     * [2020-11-06 13:27:50,208] DEBUG [Controller id=19(当前机器的Broker ID)] Broker 13 has been elected as the controller, so stopping the election process. (kafka.controller.KafkaController)     */    debug(s"Broker $activeControllerId has been elected as the controller, so stopping the election process.")    return  }
  try {    /**     * //所谓选举，就是抢占zk上面一个节点，如果抛异常说明未能选举上     * registerControllerAndIncrementControllerEpoch在zk/KafkaZkClient.scala     * 创建临时节点，声明本broker成为controller     * 主要是创建/controller节点     * 尝试去创建/controller节点，如果创建失败了(已存在)，会在catch里处理     *     * registerControllerAndIncrementControllerEpoch 方法在 zk/KafkaZkClient.scala     */    val (epoch, epochZkVersion) = zkClient.registerControllerAndIncrementControllerEpoch(config.brokerId)    controllerContext.epoch = epoch    controllerContext.epochZkVersion = epochZkVersion    /** ControllerId就是当前主机的 brokerId */    activeControllerId = config.brokerId
    /** 未抛异常说明写入创建成功，本broker荣升为controller */    info(s"${config.brokerId} successfully elected as the controller. Epoch incremented to ${controllerContext.epoch} " +      s"and epoch zk version is now ${controllerContext.epochZkVersion}")
    /** 成为Controller后，主要做的工作     * 第4步：执行当选Controller的后续逻辑     */    onControllerFailover()  } catch {    case e: ControllerMovedException =>      maybeResign()
      /** 如果/controller已存在， brokerid就不会是-1      * {"version":1,"brokerid":0,"timestamp":"1582610063256"}      */      if (activeControllerId != -1)        debug(s"Broker $activeControllerId was elected as controller instead of broker ${config.brokerId}", e)      else      /** 上一届controller刚下台，节点还没删除的情况 */        warn("A controller has been elected but just resigned, this will result in another round of election", e)
    case t: Throwable =>      error(s"Error while electing or becoming controller on broker ${config.brokerId}. " +        s"Trigger controller movement immediately", t)      /** 遇到不可知错误，取消zk相关节点的监听注册，并调用删除/controller的zk的node */      triggerControllerMove()  }}

/**   * 真正复杂的是broker在成为Controller之后，在onControllerFailover方法中进行的一系列初始化动作   *   * .启动controller的channel manager用于接收请求   * 3.启动replica的状态机，监测replica是OnlineReplica还是OfflineReplica的状态。这里的offline是指该replica的broker已经挂掉。   * 4.启动partition的状态机，监测partition是OnlinePartition还是OfflinePartition。这里的offline是指该partion的leader已经挂掉。   * 5.启动自动的leader分配rebalance（如果启动设置）   */  private def onControllerFailover() {    info("Registering handlers")
    // before reading source of truth from zookeeper, register the listeners to get broker/topic callbacks    /**     * 在读取zk之前注册brokerchange, topicchange, topicdeletion, logDirEventNoti,isrChange事件，     * 从这行及下面事件注册也看出control在集群中的作用     * 注册一组childrenChangeHandler，在NodeChildrenChange事件触发后，会分发给这些handler     * handler   监听的zk节点   事件    ControllerEvent   功能*--------------------------------------------------*brokerChangeHandler /brokers/ids    childChange   BrokerChange*topicChangeHandler  /brokers/topics childChange TopicChange*topicDeletionHandler  /admin/delete_topics    childChange   TopicDeletion*logDirEventNotificationHandler  /log_dir_event_notification   childChange   LogDirEventNotification*isrChangeNotificationHandler  /isr_change_notification    childChange   IsrChangeNotification*partitionReassignmentHandler  /admin/reassign_partitions  create  PartitionReassignment 执行副本重分配*preferredReplicaElectionHandler /admin/preferred_replica_election create  PreferredReplicaLeaderElection  Preferred leader副本选举     *     * 其中brokerChangeHandler（new BrokerChangeHandler（this，eventManager）为Broker数据量zookeeper监听器 ）     */      /** 注册各种监听器 */    val childChangeHandlers = Seq(brokerChangeHandler, topicChangeHandler, topicDeletionHandler, logDirEventNotificationHandler,      isrChangeNotificationHandler)    /** 注册对broker数据量监听器 */    childChangeHandlers.foreach(zkClient.registerZNodeChildChangeHandler)
    /**     * 注册/admin/preferred_replica_election, /admin/reassign_partitions节点事件处理     * 也是注册，不过要检查节点是否存在(这里不对是否存在做处理，只是保证没有异常)     */    val nodeChangeHandlers = Seq(preferredReplicaElectionHandler, partitionReassignmentHandler)    nodeChangeHandlers.foreach(zkClient.registerZNodeChangeHandlerAndCheckExistence)
    info("Deleting log dir event notifications")
    /**     * 删除节点：/log_dir_event_notification/log_dir_event_xxx，/isr_change_notification/isr_change_xxx节点     */    /**     * 删除log_dir_event_notification这个目录下面的子节点     * 清理已存在的LogDirEventNotifications在zk上的记录     */    zkClient.deleteLogDirEventNotifications(controllerContext.epochZkVersion)    info("Deleting isr change notifications")
    /**     * 删除isr_change_notification这个目录下面的子节点     * 清理已存在的IsrChangeNotifications在zk上的记录     */    zkClient.deleteIsrChangeNotifications(controllerContext.epochZkVersion)    info("Initializing controller context")
    /**     *  初始化controller的上下文    * 在initializeControllerContext()中:    * 1. 注册监听所有topic的partitionModification事件    * 2. 从zk中获取所有topic的副本分配信息    * 3. 在zk中监听所有broker的更新情况    * 4. 从zk中读取topicPartition的leadership,更新本地缓存    * 5. 初始化ControllerChannelManager，为每个broker生成一个后台通信线程用于和broker通信，并启动后台线程    * 6. 从zk的/admin/reassign_partitions路径下读取partition的reassigned信息放入缓存用于后续处理     *     * 初始化 controller 相关的变量信息:包括 alive broker 列表、partition 的详细信息等     */    /** 初始化集群元数据，元数据对象ControllerContext */    initializeControllerContext()  /** 初始化 controller 相关的变量信息 */    info("Fetching topic deletions in progress")
    /**     * 获取所有待删除的topic     * 要删除的topics和删除失败的topics     */    val (topicsToBeDeleted, topicsIneligibleForDeletion) = fetchTopicDeletionsInProgress()    info("Initializing topic deletion manager")
    /**     * 初始化ControllerContext之后，接下来是topicDeletionManager——topic删除管理器的初始化     * 注：topic删除只会在delete.topic.enable为true时才能进行，而且分阶段进行删除     *  初始化通过topicDeletionManager，如果isDeleteTopicEnabled则在zk中直接删除topicsToBeDeleted     *     *  初始化topic删除管理器     */    topicDeletionManager.init(topicsToBeDeleted, topicsIneligibleForDeletion)
    info("Sending update metadata request")
    /**     * 同步一下live的broker列表     * controller context 初始化结束之后发送请求更新metadata,这是因为需要在brokers能处理LeaderAndIsrRequests前获取哪些brokers是live的,     *     * 在 controller contest 初始化之后,我们需要发送 UpdateMetadata 请求在状态机启动之前,这是因为 broker 需要从 UpdateMetadata 请求     * 获取当前存活的 broker list, 因为它们需要处理来自副本状态机或分区状态机启动发送的 LeaderAndIsr 请求     *     * 在处理LeaderAndIsrRequest请求之前，先更新所有broker以及所有partition的元数据     * 是为后面的副本状态机，分区状态机的启动做准备，将元数据同步给其它broker，让它们可以处理LeaderAndIsrRequest请求     */    sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq, Set.empty)
    /**     * 启动副本状态机，初始化zk中所有副本的状态     * 如果是online的副本则标记为OnlineReplica状态，否则标记为ReplicaDeletionIneligible     * 生成LeaderAndIsrRequest请求并发送到对应brokerId     *     * 初始化 replica 的状态信息: replica 是存活状态时是 OnlineReplica, 否则是 ReplicaDeletionIneligible     */    replicaStateMachine.startup()    /**     * 启动分区状态机，初始化分区状态至OnlinePartition，对于那些NewPartition,OfflinePartition状态的分区进行选举并在zk中更新updateLeaderAndIsr     *     * 初始化 partition 的状态信息:如果 leader 所在 broker 是 alive 的,那么状态为 OnlinePartition,否则为 OfflinePartition     * 并状态为 OfflinePartition 的 topic 选举 leader     */    partitionStateMachine.startup()
    info(s"Ready to serve as the new controller with epoch $epoch")    /**     * 判断是否需要重新分配partition     * 检查是否topic的partition的副本需要重新分配(reassign),     * 如果partitionsBeingReassigned缓存中的分配信息和controllerContext缓存中不一致，则需要触发重新分配     *     * 分区副本重分配的方法入口是maybeTriggerPartitionReassignment方法，     * 该方法会在Controller初始化和PartitionReassignment事件处理器中调用     */    maybeTriggerPartitionReassignment(controllerContext.partitionsBeingReassigned.keySet)    topicDeletionManager.tryTopicDeletion()    /** Preferred leader副本选举     * 首先是通过fetchPendingPreferredReplicaElections获取要进行Preferred leader副本选举的分区     * */    val pendingPreferredReplicaElections = fetchPendingPreferredReplicaElections()
    /**     * onPreferredReplicaElection方法通过分区状态机，将分区转换为OnlinePartition状态，     * 并根据PreferredReplicaPartitionLeaderElectionStrategy选举leader，     */    onPreferredReplicaElection(pendingPreferredReplicaElections, ZkTriggered)    info("Starting the controller scheduler")    /** 定时任务 */    kafkaScheduler.startup()    if (config.autoLeaderRebalanceEnable) {      scheduleAutoLeaderRebalanceTask(delay = 5, unit = TimeUnit.SECONDS)    }
    if (config.tokenAuthEnabled) {      info("starting the token expiry check scheduler")      tokenCleanScheduler.startup()      tokenCleanScheduler.schedule(name = "delete-expired-tokens",        fun = () => tokenManager.expireTokens,        period = config.delegationTokenExpiryCheckIntervalMs,        unit = TimeUnit.MILLISECONDS)    }  }

/** config: Kafka配置信息  * zkClient:Zk客户端，Controller与zookeeper交互使用该属性  * Time,  时间戳工具类  * initialBrokerInfo: Broker 节点信息，hostname，port等  * initialBrokerEpoch: Controller所在Broker的Epoch值 *//** * KafkaController#startup之前，需要说明下KafkaController中有很多成员变量，主要分为 * zk事件处理器(ZNodeChangeHandler，ZNodeChildChangeHandler) * StateMachine(有限状态机): 副本的状态机，分区的状态机，主要负责状态的维护及转换时的处理 * ControllerContext：broker，topic，partition，replica相关的数据缓存 * ControllerEventManager: zk事件管理器， */class KafkaController(val config: KafkaConfig, zkClient: KafkaZkClient, time: Time, metrics: Metrics,                      initialBrokerInfo: BrokerInfo, initialBrokerEpoch: Long, tokenManager: DelegationTokenManager,                      threadNamePrefix: Option[String] = None) extends Logging with KafkaMetricsGroup {
  this.logIdent = s"[Controller id=${config.brokerId}] "
  @volatile private var brokerInfo = initialBrokerInfo  @volatile private var _brokerEpoch = initialBrokerEpoch
  private val stateChangeLogger = new StateChangeLogger(config.brokerId, inControllerContext = true, None)  /**   * 实例化上下文   * broker，topic，partition，replica相关的数据缓存   * 维护上下文信息，缓存 ZK 中记录的整个集群的元数据信息   */  val controllerContext = new ControllerContext
  // have a separate scheduler for the controller to be able to start and stop independently of the kafka server  // visible for testing  private[controller] val kafkaScheduler = new KafkaScheduler(1)
  // visible for testing ,  /** ControllerEventManager: zk事件管理器   * ControllerEventManager 在core/src/main/scala/kafka/controller/ControllerEventManager.scala   *  初始化eventManager，对ControllerEvent事件进行管理   *   *  KafkaController重要属性   * activeControllerId：获取controller的主broker的id   * eventManager: ControllerEventManager实例，负责处理事件   *   * */  private[controller] val eventManager = new ControllerEventManager(config.brokerId,    controllerContext.stats.rateAndTimeMetrics, _ => updateMetrics(), () => maybeResign())
  /** topic 删除管理*/  val topicDeletionManager = new TopicDeletionManager(this, eventManager, zkClient)
  /** Controller 给 其他broker 发送批量请求 , 这里有一个比较重要的对象，就是 ControllerBrokerRequestBatch 对象,   * 在ControllerChannelManager.scala中定义   * */  private val brokerRequestBatch = new ControllerBrokerRequestBatch(this, stateChangeLogger)  /** StateMachine(有限状态机): 副本的状态机，分区的状态机，主要负责状态的维护及转换时的处理 */  /** 管理集群中所有副本状态的状态机 */  val replicaStateMachine = new ReplicaStateMachine(config, stateChangeLogger, controllerContext, topicDeletionManager, zkClient, mutable.Map.empty, new ControllerBrokerRequestBatch(this, stateChangeLogger))  /** 管理集群中所有分区状态的状态机 */ff  val partitionStateMachine = new PartitionStateMachine(config, stateChangeLogger, controllerContext, zkClient, mutable.Map.empty, new ControllerBrokerRequestBatch(this, stateChangeLogger))  partitionStateMachine.setTopicDeletionManager(topicDeletionManager)
  /** Controller 控制的 事件 */    /**  Controller 节点 zookeeper 监听器*/  private val controllerChangeHandler = new ControllerChangeHandler(this, eventManager)  /** Broker  zookeeper 监听器*/  private val brokerChangeHandler = new BrokerChangeHandler(this, eventManager)  /** Broker 节点信息变更 zookeeper 监听器 */  private val brokerModificationsHandlers: mutable.Map[Int, BrokerModificationsHandler] = mutable.Map.empty  /** 主题数量 zookeeper 监听器 */  private val topicChangeHandler = new TopicChangeHandler(this, eventManager)  /** 主题删除 zookeeper 监听器 */  private val topicDeletionHandler = new TopicDeletionHandler(this, eventManager)  /** 主题分区变更 zookeeper 监听器 */  private val partitionModificationsHandlers: mutable.Map[String, PartitionModificationsHandler] = mutable.Map.empty  /** 主题分区重分配 zookeeper 监听器   * 分区副本重分配主要由/admin/reassign_partitions节点的create事件触发，该事件的处理器为partitionReassignmentHandler,   * */  private val partitionReassignmentHandler = new PartitionReassignmentHandler(this, eventManager)  /** Preferred Leader 选举zookeeper 监听器*/  private val preferredReplicaElectionHandler = new PreferredReplicaElectionHandler(this, eventManager)  /** ISR副本集群变更 zookeeper 监听器 */  private val isrChangeNotificationHandler = new IsrChangeNotificationHandler(this, eventManager)  /** 日志路径变更 zookeeper 监听器*/  private val logDirEventNotificationHandler = new LogDirEventNotificationHandler(this, eventManager)