ZookeeperClient implements Watcher { String createIfNeeded(String path, byte[] data, CreateMode createMode) throws ZookeeperClientFailedException, InterruptedException { assert !createMode.isSequential() : createMode; if (data == null) data = EMPTY_BYTES; for (;;) { long connStartTime = this.connStartTime; try { return zk.create(path, data, ZK_ACL, createMode); } catch (KeeperException.NodeExistsException e) { if (log.isDebugEnabled()) log.debug("Node already exists: " + path); return path; } catch (Exception e) { onZookeeperError(connStartTime, e); } } } ZookeeperClient(IgniteLogger log, String connectString, int sesTimeout, IgniteRunnable connLostC);   ZookeeperClient(String igniteInstanceName,
        IgniteLogger log,
        String connectString,
        int sesTimeout,
        IgniteRunnable connLostC); @Override void process(WatchedEvent evt); void close();  }
@Test public void testConnectionLoss1() throws Exception { ZookeeperClient client = new ZookeeperClient(log, "localhost:2200", 3000, null); try { client.createIfNeeded("/apacheIgnite", null, CreateMode.PERSISTENT); fail(); } catch (ZookeeperClientFailedException e) { info("Expected error: " + e); } }
@Test public void testConnectionLoss2() throws Exception { startZK(1); ZookeeperClient client = createClient(3000); client.createIfNeeded("/apacheIgnite1", null, CreateMode.PERSISTENT); closeZK(); try { client.createIfNeeded("/apacheIgnite2", null, CreateMode.PERSISTENT); fail(); } catch (ZookeeperClientFailedException e) { info("Expected error: " + e); } }
@Test public void testReconnect1() throws Exception { startZK(1); ZookeeperClient client = createClient(SES_TIMEOUT); client.createIfNeeded("/apacheIgnite1", null, CreateMode.PERSISTENT); zkCluster.getServers().get(0).stop(); IgniteInternalFuture fut = GridTestUtils.runAsync(new Callable<Void>() { @Override public Void call() throws Exception { U.sleep(2000); info("Restart zookeeper server"); zkCluster.getServers().get(0).restart(); info("Zookeeper server restarted"); return null; } }, "start-zk"); client.createIfNeeded("/apacheIgnite2", null, CreateMode.PERSISTENT); fut.get(); }
@Test public void testReconnect2() throws Exception { startZK(1); ZookeeperClient client = createClient(SES_TIMEOUT); client.createIfNeeded("/apacheIgnite1", null, CreateMode.PERSISTENT); zkCluster.getServers().get(0).restart(); client.createIfNeeded("/apacheIgnite2", null, CreateMode.PERSISTENT); }
@Test public void testReconnect3() throws Exception { startZK(3); ZookeeperClient client = createClient(SES_TIMEOUT); ThreadLocalRandom rnd = ThreadLocalRandom.current(); for (int i = 0; i < 30; i++) { info("Iteration: " + i); int idx = rnd.nextInt(3); zkCluster.getServers().get(idx).restart(); doSleep(rnd.nextLong(100) + 1); client.createIfNeeded("/apacheIgnite" + i, null, CreateMode.PERSISTENT); } }
@Test public void testReconnect4() throws Exception { startZK(3); ZookeeperClient client = new ZookeeperClient(log, zkCluster.getServers().get(2).getInstanceSpec().getConnectString(), 60_000, null); client.createIfNeeded("/apacheIgnite1", null, CreateMode.PERSISTENT); zkCluster.getServers().get(0).stop(); zkCluster.getServers().get(1).stop(); IgniteInternalFuture fut = GridTestUtils.runAsync(new Callable<Void>() { @Override public Void call() throws Exception { U.sleep(2000); info("Restart zookeeper server"); zkCluster.getServers().get(0).restart(); info("Zookeeper server restarted"); return null; } }, "start-zk"); client.createIfNeeded("/apacheIgnite2", null, CreateMode.PERSISTENT); fut.get(); }