ZooKeeperConnection implements Watcher { public synchronized ZooKeeper getZooKeeper() throws IOException, InterruptedException { if (zk == null || zk.getState() != ZooKeeper.States.CONNECTED) { latch = new CountDownLatch(1); zk = new ZooKeeper(connection, sessionTimeoutMs, this); boolean done = latch.await(connectionTimeoutMs, TimeUnit.MILLISECONDS); if (!done) { throw new InterruptedException("Did not connect to ZooKeeper within " + connectionTimeoutMs + " seconds"); } } return zk; } ZooKeeperConnection(String connection); void setSessionTimeoutMs(int sessionTimeoutMs); void setConnectionTimeoutMs(int connectionTimeoutMs); @Override void process(WatchedEvent event); synchronized ZooKeeper getZooKeeper(); static final String CONNECTION_CONFIG; }
@Test public void testGoodConnection() throws Exception { new ZooKeeperConnection(String.format("127.0.0.1:%d", zk.getPort())).getZooKeeper(); }