selectLeader

gistfile1.scala

  if (clientEnabled) {
    startThread("leaderSelection") {
      while (electingLeader) {
        selectLeader()
        electLoop.acquire()
        if (electingLeader) log.info("Lost connection to Zookeeper - reselecting leader")
      }
    }
    log.info("Waiting for leader to be selected")
    leaderSelected.acquire()
    log.info("Leader selected")
  }

  def selectLeader() {
    val leadershipTaken = new Semaphore(0)

    val listener: LeaderSelectorListener = new LeaderSelectorListener {
      def takeLeadership(client: CuratorFramework) {
        log.info("nodeId=" + nodeId + " *** Cluster leadership taken ***")
        leadershipTaken.release()
        leadershipHold.acquire()
        log.info("nodeId=" + nodeId + " *** Releasing cluster leadership ***")
      }

      def stateChanged(client: CuratorFramework, newState: ConnectionState) {
        val currentLeaderSelector = leaderSelector
        log.info("nodeId=" + nodeId + " State changed.  new state=" + newState)
        if (newState == ConnectionState.LOST) {
          if (leadershipHold.availablePermits() == 0)
            leadershipHold.release()
          if (electLoop.availablePermits() == 0)
            electLoop.release()
          currentLeaderSelector.close()
        }
      }
    }

    leaderSelector = new LeaderSelector(client, leaderPath, listener)
    leaderSelector.setId(nodeId)
    leaderSelector.start()

    // Wait until someone has been selected as leader
    val timedOut = !leadershipTaken.tryAcquire(waitingSec, TimeUnit.SECONDS)
    log.debug("nodeId=" + nodeId + " leadershipTaken.tryAcquire timedOut=" + timedOut)

    val participants = leaderSelector.getParticipants
    log.info("nodeId=" + nodeId + " Cluster participants: " + participants.asScala.mkString(", "))

    if (leaderSelected.availablePermits() == 0)
      leaderSelected.release()
  }