@Override
  public void tryBecomeLeader() {
    String currentPath = zkUtils.registerProcessorAndGetId(new ProcessorData(hostName, processorIdStr));

    List<String> children = zkUtils.getSortedActiveProcessorsZnodes();
    LOG.debug(zLog("Current active processors - " + children));
    int index = children.indexOf(ZkKeyBuilder.parseIdFromPath(currentPath));

    LOG.info("tryBecomeLeader: index = " + index + " for path=" + currentPath + " out of " + Arrays.toString(children.toArray()));
    if (children.size() == 0 || index == -1) {
      throw new SamzaException("Looks like we are no longer connected to Zk. Need to reconnect!");
    }

    if (index == 0) {
      isLeader.getAndSet(true);
      LOG.info(zLog("Eligible to become the leader!"));
      if (leaderElectorListener != null) {
        leaderElectorListener.onBecomingLeader();
      }
      return;
    }

    isLeader.getAndSet(false);
    LOG.info("Index = " + index + " Not eligible to be a leader yet!");
    String predecessor = children.get(index - 1);
    if (!predecessor.equals(currentSubscription)) {
      if (currentSubscription != null) {
        LOG.debug(zLog("Unsubscribing data change for " + currentSubscription));
        zkUtils.unsubscribeDataChanges(keyBuilder.getProcessorsPath() + "/" + currentSubscription,
            previousProcessorChangeListener);
        previousProcessorChangeListener = new PreviousProcessorChangeListener(zkUtils);
      }
      currentSubscription = predecessor;
      LOG.info(zLog("Subscribing data change for " + predecessor));
      zkUtils.subscribeDataChanges(keyBuilder.getProcessorsPath() + "/" + currentSubscription,
          previousProcessorChangeListener);
    }
    /**
     * Verify that the predecessor still exists. This step is needed because the ZkClient subscribes for data changes
     * on the path, even if the path doesn't exist. Since we are using Ephemeral Sequential nodes, if the path doesn't
     * exist during subscription, it is not going to get created in the future.
     */
    boolean predecessorExists = zkUtils.exists(keyBuilder.getProcessorsPath() + "/" + currentSubscription);
    if (predecessorExists) {
      LOG.info(zLog("Predecessor still exists. Current subscription is valid. Continuing as non-leader."));
    } else {
      try {
        Thread.sleep(random.nextInt(1000));
      } catch (InterruptedException e) {
        Thread.interrupted();
      }
      LOG.info(zLog("Predecessor doesn't exist anymore. Trying to become leader again..."));
      tryBecomeLeader();
    }
  }
Async method that helps the caller participate in leader election.
samza-core/src/main/java/org/apache/samza/zk/ZkLeaderElector.java