ZkProcessorLatch implements Latch { @Override public void await(long timeout, TimeUnit timeUnit) throws TimeoutException { boolean targetPathExists = zkUtils.getZkClient().waitUntilExists(targetPath, timeUnit, timeout); if (!targetPathExists) { throw new TimeoutException("Timed out waiting for the targetPath"); } } ZkProcessorLatch(int size, String latchId, String participantId, ZkUtils zkUtils); @Override void await(long timeout, TimeUnit timeUnit); @Override void countDown(); final static String LATCH_PATH; }
@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); 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()); } }