ZkBarrierForVersionUpgrade { public ZkBarrierForVersionUpgrade(String barrierRoot, ZkUtils zkUtils, ZkBarrierListener barrierListener, ScheduleAfterDebounceTime debounceTimer) { if (zkUtils == null) { throw new RuntimeException("Cannot operate ZkBarrierForVersionUpgrade without ZkUtils."); } this.zkUtils = zkUtils; this.keyBuilder = new BarrierKeyBuilder(barrierRoot); this.barrierListenerOptional = Optional.ofNullable(barrierListener); this.debounceTimer = debounceTimer; } ZkBarrierForVersionUpgrade(String barrierRoot, ZkUtils zkUtils, ZkBarrierListener barrierListener, ScheduleAfterDebounceTime debounceTimer); void create(final String version, List<String> participants); void join(String version, String participantId); void expire(String version); static int getVersion(String barrierPath);  }
@Test public void testZkBarrierForVersionUpgrade() { String barrierId = String.format("%s/%s", zkUtils1.getKeyBuilder().getRootPath(), RandomStringUtils.randomAlphabetic(4)); List<String> processors = ImmutableList.of("p1", "p2"); CountDownLatch latch = new CountDownLatch(2); TestZkBarrierListener listener = new TestZkBarrierListener(latch, State.DONE); ZkBarrierForVersionUpgrade processor1Barrier = new ZkBarrierForVersionUpgrade(barrierId, zkUtils, listener, debounceTimer); ZkBarrierForVersionUpgrade processor2Barrier = new ZkBarrierForVersionUpgrade(barrierId, zkUtils1, listener, debounceTimer); processor1Barrier.create(BARRIER_VERSION, processors); String barrierState = zkUtils.getZkClient().readData(barrierId + "/barrier_1/barrier_state"); assertEquals(State.NEW, State.valueOf(barrierState)); processor1Barrier.join(BARRIER_VERSION, "p1"); processor2Barrier.join(BARRIER_VERSION, "p2"); boolean result = false; try { result = latch.await(10000, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { e.printStackTrace(); } assertTrue("Barrier failed to complete within test timeout.", result); List<String> children = zkUtils.getZkClient().getChildren(barrierId + "/barrier_1/barrier_participants"); barrierState = zkUtils.getZkClient().readData(barrierId + "/barrier_1/barrier_state"); assertEquals(State.DONE, State.valueOf(barrierState)); assertNotNull(children); assertEquals("Unexpected barrier state. Didn't find two processors.", 2, children.size()); assertEquals("Unexpected barrier state. Didn't find the expected members.", processors, children); }