@Test
  public void testNonLeaderFailure() {
    /**
     * electionLatch and count together verify that:
     * 1. the registered listeners are actually invoked by the ZkClient on the correct path
     * 2. for a single participant failure, at-most 1 other participant is notified
     */
    final CountDownLatch electionLatch = new CountDownLatch(1);
    final AtomicInteger count = new AtomicInteger(0);

    BooleanResult isLeader1 = new BooleanResult();
    BooleanResult isLeader2 = new BooleanResult();
    BooleanResult isLeader3 = new BooleanResult();

    // Processor-1
    ZkUtils zkUtils1 = getZkUtilsWithNewClient();
    zkUtils1.registerProcessorAndGetId(new ProcessorData("processor1", "1"));
    ZkLeaderElector leaderElector1 = new ZkLeaderElector("processor1", zkUtils1, new IZkDataListener() {
        @Override
        public void handleDataChange(String dataPath, Object data) throws Exception { }

        @Override
        public void handleDataDeleted(String dataPath) throws Exception {
          count.incrementAndGet();
        }
      });
    leaderElector1.setLeaderElectorListener(() -> isLeader1.res = true);

    // Processor-2
    ZkUtils zkUtils2 = getZkUtilsWithNewClient();
    zkUtils2.registerProcessorAndGetId(new ProcessorData("processor2", "2"));
    ZkLeaderElector leaderElector2 = new ZkLeaderElector("processor2", zkUtils2, new IZkDataListener() {
      @Override
      public void handleDataChange(String dataPath, Object data) throws Exception { }

      @Override
      public void handleDataDeleted(String dataPath) throws Exception {
        count.incrementAndGet();
      }
    });
    leaderElector2.setLeaderElectorListener(() -> isLeader2.res = true);

    // Processor-3
    ZkUtils zkUtils3  = getZkUtilsWithNewClient();
    final String path3 = zkUtils3.registerProcessorAndGetId(new ProcessorData("processor3", "3"));
    ZkLeaderElector leaderElector3 = new ZkLeaderElector("processor3", zkUtils3, new IZkDataListener() {
      @Override
      public void handleDataChange(String dataPath, Object data) throws Exception { }

      @Override
      public void handleDataDeleted(String dataPath)
          throws Exception {
        String registeredIdStr = ZkKeyBuilder.parseIdFromPath(path3);
        Assert.assertNotNull(registeredIdStr);

        String predecessorIdStr = ZkKeyBuilder.parseIdFromPath(dataPath);
        Assert.assertNotNull(predecessorIdStr);

        try {
          int selfId = Integer.parseInt(registeredIdStr);
          int predecessorId = Integer.parseInt(predecessorIdStr);
          Assert.assertEquals(1, selfId - predecessorId);
        } catch (Exception e) {
          Assert.fail("Exception in LeaderElectionListener!");
        }
        count.incrementAndGet();
        electionLatch.countDown();
      }
    });
    leaderElector3.setLeaderElectorListener(() -> isLeader3.res = true);

    // Join Leader Election
    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));

    List<String> currentActiveProcessors = zkUtils1.getSortedActiveProcessorsZnodes();
    Assert.assertEquals(3, currentActiveProcessors.size());

    zkUtils2.close();
    currentActiveProcessors.remove(1);

    try {
      Assert.assertTrue(electionLatch.await(5, TimeUnit.SECONDS));
    } catch (InterruptedException e) {
      Assert.fail("Interrupted while waiting for leaderElection listener callback to complete!");
    }

    Assert.assertEquals(1, count.get());
    Assert.assertEquals(currentActiveProcessors, zkUtils1.getSortedActiveProcessorsZnodes());

    // Clean up
    zkUtils1.close();
    zkUtils3.close();
  }
Tests that a non-leader failure updates the Zk tree and participants' state correctly
samza-core/src/test/java/org/apache/samza/zk/TestZkLeaderElector.java