本文所講的Leader是指叢集中的Controller，而不是各個Partition的Leader。

為什麼要有Leader？

在Kafka早期版本，對於分割槽和副本的狀態的管理依賴於zookeeper的Watcher和佇列：每一個broker都會在zookeeper註冊Watcher，所以zookeeper就會出現大量的Watcher, 如果宕機的broker上的partition很多比較多，會造成多個Watcher觸發，造成叢集內大規模調整；每一個replica都要去再次zookeeper上註冊監視器，當叢集規模很大的時候，zookeeper負擔很重。這種設計很容易出現腦裂和羊群效應以及zookeeper叢集過載。

新的版本中該變了這種設計，使用KafkaController，只有KafkaController，Leader會向zookeeper上註冊Watcher，其他broker幾乎不用監聽zookeeper的狀態變化。

Kafka叢集中多個broker，有一個會被選舉為controller leader，負責管理整個叢集中分割槽和副本的狀態，比如partition的leader 副本故障，由controller 負責為該partition重新選舉新的leader 副本；當檢測到ISR列表發生變化，有controller通知叢集中所有broker更新其MetadataCache資訊；或者增加某個topic分割槽的時候也會由controller管理分割槽的重新分配工作

Kafka叢集Leader選舉原理

我們知道Zookeeper叢集中也有選舉機制，是通過Paxos演算法，通過不同節點向其他節點發送資訊來投票選舉出leader，但是Kafka的leader的選舉就沒有這麼複雜了。
Kafka的Leader選舉是通過在zookeeper上建立/controller臨時節點來實現leader選舉，並在該節點中寫入當前broker的資訊
{“version”:1,”brokerid”:1,”timestamp”:”1512018424988”}
利用Zookeeper的強一致性特性，一個節點只能被一個客戶端建立成功，建立成功的broker即為leader，即先到先得原則，leader也就是叢集中的controller，負責叢集中所有大小事務。
當leader和zookeeper失去連線時，臨時節點會刪除，而其他broker會監聽該節點的變化，當節點刪除時，其他broker會收到事件通知，重新發起leader選舉。

KafkaController

KafkaController初始化ZookeeperLeaderElector物件，為ZookeeperLeaderElector設定兩個回撥方法，onControllerFailover和onControllerResignation
onControllerFailover在選舉leader成功後會回撥，在onControllerFailover中進行leader依賴的模組初始化，包括向zookeeper上/controller_epoch節點上記錄leader的選舉次數，這個epoch數值在處理分散式腦裂的場景中很有用。
而onControllerResignation在當前broker不再成為leader（即當前leader退位後）時會回撥。
KafkaController在啟動後註冊zookeeper的會話超時監聽器，並嘗試選舉leader。

class KafkaController {
  private val controllerElector = new ZookeeperLeaderElector(controllerContext, ZkUtils.ControllerPath, onControllerFailover, onControllerResignation, config.brokerId)

  def startup() = {
    inLock(controllerContext.controllerLock) {
      info("Controller starting up")
      //註冊Session過期監聽器
      registerSessionExpirationListener()
      isRunning = true
      //每次啟動時，嘗試選舉leader
      controllerElector.startup
      info("Controller startup complete")
    }
  }

  private def registerSessionExpirationListener() = {
    zkUtils.zkClient.subscribeStateChanges(new SessionExpirationListener())
  }
}

SessionExpirationListener

當broker和zookeeper重新建立連線後，SessionExpirationListener中的handleNewSession會被呼叫，這時先關閉之前的leader相關模組，然後重新嘗試選舉成為leader。

  class SessionExpirationListener() extends IZkStateListener with Logging {
    this.logIdent = "[SessionExpirationListener on " + config.brokerId + "], "
    @throws(classOf[Exception])
    def handleStateChanged(state: KeeperState) {
      // do nothing, since zkclient will do reconnect for us.
    }

    /**
     * Called after the zookeeper session has expired and a new session has been created. You would have to re-create
     * any ephemeral nodes here.
     *
     * @throws Exception
     *             On any error.
     */
    @throws(classOf[Exception])
    def handleNewSession() {
      info("ZK expired; shut down all controller components and try to re-elect")
      //和Zookeeper重新建立連線後，此方法會被呼叫
      inLock(controllerContext.controllerLock) {
        //先登出一些已經註冊的監聽器，關閉資源
        onControllerResignation()
        //重新嘗試選舉成controller
        controllerElector.elect
      }
    }

    override def handleSessionEstablishmentError(error: Throwable): Unit = {
      //no-op handleSessionEstablishmentError in KafkaHealthCheck should handle this error in its handleSessionEstablishmentError
    }
  }

ZookeeperLeaderElector

ZookeeperLeaderElector類實現leader選舉的功能，但是它並不負責處理broker和zookeeper的會話超時（連線超時）的情況，而是認為呼叫者應該在會話恢復（連線重新建立）時進行重新選舉。

class ZookeeperLeaderElector(controllerContext: ControllerContext,
                             electionPath: String,
                             onBecomingLeader: () => Unit,
                             onResigningAsLeader: () => Unit,
                             brokerId: Int)
  extends LeaderElector with Logging {
  var leaderId = -1
  // create the election path in ZK, if one does not exist
  val index = electionPath.lastIndexOf("/")
  if (index > 0)
    controllerContext.zkUtils.makeSurePersistentPathExists(electionPath.substring(0, index))
  val leaderChangeListener = new LeaderChangeListener

  def startup {
    inLock(controllerContext.controllerLock) {
      // 新增/controller節點的IZkDataListener監聽器
      controllerContext.zkUtils.zkClient.subscribeDataChanges(electionPath, leaderChangeListener)
      // 選舉
      elect
    }
  }
}

ZookeeperLeaderElector的startup方法中呼叫elect方法選舉leader

有下面幾種情況會呼叫elect方法
1. broker啟動時，第一次呼叫
2. 上一次建立節點成功，但是可能在等Zookeeper響應的時候，連線中斷，resign方法中刪除/controller節點後，觸發了leaderChangeListener的handleDataDeleted
3. 上一次建立節點未成功，但是可能在等Zookeeper響應的時候，連線中斷，而再次進入elect方法時，已有別的broker建立controller節點成功，成為了leader
4. 上一次建立節點成功，但是onBecomingLeader丟擲了異常，而再次進入
所以elect方法中先獲取/controller節點資訊，判斷是否已經存在，然後再嘗試選舉leader

  private def getControllerID(): Int = {
    controllerContext.zkUtils.readDataMaybeNull(electionPath)._1 match {
       case Some(controller) => KafkaController.parseControllerId(controller)
       case None => -1
    }
  }

  def elect: Boolean = {
    val timestamp = SystemTime.milliseconds.toString
    val electString = Json.encode(Map("version" -> 1, "brokerid" -> brokerId, "timestamp" -> timestamp))

    //先嚐試獲取/controller節點資訊
   leaderId = getControllerID 
    /* 
     * We can get here during the initial startup and the handleDeleted ZK callback. Because of the potential race condition, 
     * it's possible that the controller has already been elected when we get here. This check will prevent the following 
     * createEphemeralPath method from getting into an infinite loop if this broker is already the controller.
     */
    // 有下面幾種情況會呼叫elect方法
    // 1.broker啟動時，第一次呼叫
    // 2.上一次建立節點成功，但是可能在等Zookeeper響應的時候，連線中斷，resign方法中刪除/controller節點後，觸發了leaderChangeListener的handleDataDeleted
    // 3.上一次建立節點未成功，但是可能在等Zookeeper響應的時候，連線中斷，而再次進入elect方法時，已有別的broker建立controller節點成功，成為了leader
    // 4.上一次建立節點成功，但是onBecomingLeader丟擲了異常，而再次進入
    // 所以先獲取節點資訊，判斷是否已經存在
    if(leaderId != -1) {
       debug("Broker %d has been elected as leader, so stopping the election process.".format(leaderId))
       return amILeader
    }

    try {
      val zkCheckedEphemeral = new ZKCheckedEphemeral(electionPath,
                                                      electString,
                                                      controllerContext.zkUtils.zkConnection.getZookeeper,
                                                      JaasUtils.isZkSecurityEnabled())
      //建立/controller節點，並寫入controller資訊，brokerid, version, timestamp
      zkCheckedEphemeral.create()
      info(brokerId + " successfully elected as leader")
      leaderId = brokerId
      //寫入成功，成為Leader，回撥
      onBecomingLeader()
    } catch {
      case e: ZkNodeExistsException =>
        // If someone else has written the path, then
        leaderId = getControllerID 
        //寫入失敗，節點已經存在，說明已有其他broker建立成功
        if (leaderId != -1)
          debug("Broker %d was elected as leader instead of broker %d".format(leaderId, brokerId))
        else
          warn("A leader has been elected but just resigned, this will result in another round of election")

      case e2: Throwable =>
        error("Error while electing or becoming leader on broker %d".format(brokerId), e2)
        //這裡有可能是建立節點時，和zookeeper斷開了連線，也有可能是onBecomingLeader的回撥方法裡出了異常
        //onBecomingLeader方法裡，一般是初始化leader的相關的模組，如果初始化失敗，則呼叫resign方法先刪除/controller節點
        //當/controller節點被刪除時，會觸發leaderChangeListener的handleDataDeleted，會重新嘗試選舉成Leader，更重要的是也讓其他broker有機會成為leader，避免某一個broker的onBecomingLeader一直失敗造成整個叢集一直處於“群龍無首”的尷尬局面
        resign()
    }
    amILeader
  }

  def close = {
    leaderId = -1
  }

  def amILeader : Boolean = leaderId == brokerId

  def resign() = {
    leaderId = -1
    // 刪除/controller節點
    controllerContext.zkUtils.deletePath(electionPath)
  }

在建立/controller節點時，若收到的異常是ZkNodeExistsException，則說明其他broker已經成為了leader。
而若是onBecomingLeader的回撥方法裡出了異常，一般是初始化leader的相關的模組出了問題，如果初始化失敗，則呼叫resign方法先刪除/controller節點。
當/controller節點被刪除時，會觸發leaderChangeListener的handleDataDeleted，會重新嘗試選舉成Leader。
更重要的是也讓其他broker有機會成為leader，避免某一個broker的onBecomingLeader一直失敗造成整個叢集一直處於“群龍無首”的尷尬局面。

LeaderChangeListener

在startup方法中，註冊了/controller節點的IZkDataListener監聽器即LeaderChangeListener。
若節點資料有變化時，則有可能別的broker成為了leader，則呼叫onResigningAsLeader方法。
若節點被刪除，則是leader已經出了故障下線了，如果當前broker之前是leader，則呼叫onResigningAsLeader方法，然後重新嘗試選舉成為leader。

  class LeaderChangeListener extends IZkDataListener with Logging {
    /**
     * Called when the leader information stored in zookeeper has changed. Record the new leader in memory
     * @throws Exception On any error.
     */
    @throws(classOf[Exception])
    def handleDataChange(dataPath: String, data: Object) {
      inLock(controllerContext.controllerLock) {
        val amILeaderBeforeDataChange = amILeader
        leaderId = KafkaController.parseControllerId(data.toString)
        info("New leader is %d".format(leaderId))
        // The old leader needs to resign leadership if it is no longer the leader
        if (amILeaderBeforeDataChange && !amILeader)
          //如果之前是Leader，而現在不是Leader
          onResigningAsLeader()
      }
    }

    /**
     * Called when the leader information stored in zookeeper has been delete. Try to elect as the leader
     * @throws Exception
     *             On any error.
     */
    @throws(classOf[Exception])
    def handleDataDeleted(dataPath: String) {
      inLock(controllerContext.controllerLock) {
        debug("%s leader change listener fired for path %s to handle data deleted: trying to elect as a leader"
          .format(brokerId, dataPath))
        if(amILeader)
          //如果之前是Leader
          onResigningAsLeader()
        //重新嘗試選舉成Leader
        elect
      }
    }
  }

onBecomingLeader方法對應KafkaController裡的onControllerFailover方法，當成為新的leader後，要初始化leader所依賴的功能模組
onResigningAsLeader方法對應KafkaController裡的onControllerResignation方法，當leader退位後，要關閉leader所依賴的功能模組

Leader選舉流程圖

整個leader選舉的過程的流程圖為