@Test
  public void testLatchSizeOneWithTwoParticipants() {
    final int latchSize = 1;
    final String latchId = "testLatchSizeOneWithTwoParticipants";

    ExecutorService pool = Executors.newFixedThreadPool(3);
    Future f1 = pool.submit(
      () -> {
        String participant1 = "participant1";
        ZkUtils zkUtils = getZkUtilsWithNewClient(participant1);
        zkUtils.connect();
        Latch latch = new ZkProcessorLatch(latchSize, latchId, participant1, zkUtils);
        //latch.countDown(); only one thread counts down
        try {
          latch.await(30, TimeUnit.SECONDS);
        } catch (TimeoutException e) {
          Assert.fail(String.format("await timed out from  %s - %s", participant1, e.getLocalizedMessage()));
        } finally {
          zkUtils.close();
        }
      });

    Future f2 = pool.submit(getParticipantRunnable(latchSize, latchId, "participant2"));

    try {
      f1.get(30, TimeUnit.SECONDS);
      f2.get(30, TimeUnit.SECONDS);
    } catch (Exception e) {
      Assert.fail("failed to get future." + e.getLocalizedMessage());
    } finally {
      pool.shutdownNow();
    }
    try {
      List<String> latchParticipants =
          testZkUtils.getZkClient().getChildren(
              String.format("%s/%s_%s", KEY_BUILDER.getRootPath(), ZkProcessorLatch.LATCH_PATH, latchId));
      Assert.assertNotNull(latchParticipants);
      Assert.assertEquals(1, latchParticipants.size());
      Assert.assertEquals("0000000000", latchParticipants.get(0));
    } catch (Exception e) {
      Assert.fail("Failed to read the latch status from ZK directly" + e.getLocalizedMessage());
    }
  }
The ZkProcessorLatch uses a shared Znode as a latch. Each participant await existence of a target znode under the
shared latch, which is a persistent, sequential target znode with value (latchSize - 1). latchSize is the minimum
number of participants that need to join the latch.
samza-core/src/test/java/org/apache/samza/zk/TestZkProcessorLatch.java