@Test
  public void testLeaderElection() {
    BooleanResult isLeader1 = new BooleanResult();
    BooleanResult isLeader2 = new BooleanResult();
    BooleanResult isLeader3 = new BooleanResult();


    // Processor-1
    ZkUtils zkUtils1 = getZkUtilsWithNewClient();
    ZkLeaderElector leaderElector1 = new ZkLeaderElector("1", zkUtils1);
    leaderElector1.setLeaderElectorListener(() -> isLeader1.res = true);

    // Processor-2
    ZkUtils zkUtils2 = getZkUtilsWithNewClient();
    ZkLeaderElector leaderElector2 = new ZkLeaderElector("2", zkUtils2);
    leaderElector2.setLeaderElectorListener(() -> isLeader2.res = true);

    // Processor-3
    ZkUtils zkUtils3 = getZkUtilsWithNewClient();
    ZkLeaderElector leaderElector3 = new ZkLeaderElector("3", zkUtils3);
    leaderElector3.setLeaderElectorListener(() -> isLeader3.res = true);

    Assert.assertEquals(0, testZkUtils.getSortedActiveProcessorsZnodes().size());

    leaderElector1.tryBecomeLeader();
    leaderElector2.tryBecomeLeader();
    leaderElector3.tryBecomeLeader();

    Assert.assertTrue(TestZkUtils.testWithDelayBackOff(() -> isLeader1.res, 2, 100));
    Assert.assertFalse(TestZkUtils.testWithDelayBackOff(() -> isLeader2.res, 2, 100));
    Assert.assertFalse(TestZkUtils.testWithDelayBackOff(() -> isLeader3.res, 2, 100));

    Assert.assertEquals(3, testZkUtils.getSortedActiveProcessorsZnodes().size());

    // Clean up
    zkUtils1.close();
    zkUtils2.close();
    zkUtils3.close();

    Assert.assertEquals(new ArrayList<String>(), testZkUtils.getSortedActiveProcessorsZnodes());

  }
Test starts 3 processors and verifies the state of the Zk tree after all processors participate in LeaderElection
samza-core/src/test/java/org/apache/samza/zk/TestZkLeaderElector.java