NodeCollectionWatcher { public void shutdown() { shuttingDown.set(true); } NodeCollectionWatcher(
            final ZooKeeper zk, final String pathToWatch, final NodeWatcherListener listener); void shutdown();  }
@Test public void sequentialNotifications() throws Exception { final int maxPropagationTime = 4; final String pathPrefix = "/foo/slow"; curator.create().creatingParentsIfNeeded().forPath(pathPrefix); final ListenerCounter listener = new ListenerCounter(1, 1, 1); final NodeCollectionWatcher nodeCollectionWatcher = new NodeCollectionWatcher(zooKeeper, pathPrefix, listener); curator.create().forPath(pathPrefix + "/node1", getData()); assertTrue(listener.createLatch.await(maxPropagationTime, TimeUnit.MILLISECONDS)); assertThat(listener.createCount.get(), is(1)); assertThat(listener.dataCount.get(), is(0)); assertThat(listener.removeCount.get(), is(0)); curator.setData().forPath(pathPrefix + "/node1", getData()); assertTrue(listener.dataLatch.await(maxPropagationTime, TimeUnit.MILLISECONDS)); assertThat(listener.createCount.get(), is(1)); assertThat(listener.dataCount.get(), is(1)); assertThat(listener.removeCount.get(), is(0)); curator.delete().forPath(pathPrefix + "/node1"); assertTrue(listener.removeLatch.await(maxPropagationTime, TimeUnit.MILLISECONDS)); assertThat(listener.createCount.get(), is(1)); assertThat(listener.dataCount.get(), is(1)); assertThat(listener.removeCount.get(), is(1)); nodeCollectionWatcher.shutdown(); curator.create().forPath(pathPrefix + "node_9", getData()); Thread.sleep(maxPropagationTime); assertThat(listener.createCount.get(), is(1)); assertThat(listener.dataCount.get(), is(1)); assertThat(listener.removeCount.get(), is(1)); curator.setData().forPath(pathPrefix + "node_9", getData()); Thread.sleep(maxPropagationTime); assertThat(listener.createCount.get(), is(1)); assertThat(listener.dataCount.get(), is(1)); assertThat(listener.removeCount.get(), is(1)); curator.delete().forPath(pathPrefix + "node_9"); Thread.sleep(maxPropagationTime); assertThat(listener.createCount.get(), is(1)); assertThat(listener.dataCount.get(), is(1)); assertThat(listener.removeCount.get(), is(1)); }
@Test public void rapidChanges() throws Exception { final int maxPropagationTime = 100; final String pathPrefix = "/foo/rapido"; curator.create().creatingParentsIfNeeded().forPath(pathPrefix); final int numNodes = 50; final ListenerCounter listener = new ListenerCounter(numNodes, 0, numNodes); final NodeCollectionWatcher nodeCollectionWatcher = new NodeCollectionWatcher(zooKeeper, pathPrefix, listener); for (int i = 0; i < numNodes; i++) { curator.create().forPath(pathPrefix + "/node" + i, getData()); } assertTrue(listener.createLatch.await(maxPropagationTime, TimeUnit.MILLISECONDS)); assertThat(listener.createCount.get(), is(numNodes)); assertThat(listener.dataCount.get(), is(0)); assertThat(listener.removeCount.get(), is(0)); int total = 0; for (int j = 0; j < 5; j++) { listener.dataLatch = new CountDownLatch(numNodes); for (int i = 0; i < numNodes; i++) { curator.setData().forPath(pathPrefix + "/node" + i, getData()); } total += numNodes; assertTrue(listener.dataLatch.await(maxPropagationTime, TimeUnit.MILLISECONDS)); assertThat(listener.createCount.get(), is(numNodes)); assertThat(listener.dataCount.get(), is(total)); assertThat(listener.removeCount.get(), is(0)); } for (int i = 0; i < numNodes; i++) { curator.delete().forPath(pathPrefix + "/node" + i); } assertTrue(listener.removeLatch.await(maxPropagationTime, TimeUnit.MILLISECONDS)); assertThat(listener.createCount.get(), is(numNodes)); assertThat(listener.dataCount.get(), is(total)); assertThat(listener.removeCount.get(), is(numNodes)); nodeCollectionWatcher.shutdown(); }
@Test public void networkPartitionTest() throws Exception { assumeThat(System.getProperty("NodeCollectionWatcher.SlowTests"), is("ok")); final int maxPropagationTime = 10; final String pathPrefix = "/foo/partition"; curator.create().creatingParentsIfNeeded().forPath(pathPrefix); final int nodeCount = 10; final ListenerCounter listener = new ListenerCounter(nodeCount, nodeCount, nodeCount); final NodeCollectionWatcher nodeCollectionWatcher = new NodeCollectionWatcher(zooKeeper, pathPrefix, listener); for (int i = 0; i < nodeCount; i++) { curator.create().forPath(pathPrefix + "/node" + i, getData()); } assertTrue(listener.createLatch.await(maxPropagationTime, TimeUnit.MILLISECONDS)); assertThat(listener.createCount.get(), is(nodeCount)); assertThat(listener.removeCount.get(), is(0)); assertThat(listener.dataCount.get(), is(0)); final InstanceSpec firstInstance = zkServer.findConnectionInstance(zooKeeper); zkServer.killServer(firstInstance); listener.createLatch = new CountDownLatch(1); curator.create().forPath(pathPrefix + "/stillalive", getData()); assertTrue(listener.createLatch.await(10, TimeUnit.SECONDS)); assertThat(listener.createCount.get(), is(nodeCount + 1)); assertThat(listener.removeCount.get(), is(0)); assertThat(listener.dataCount.get(), is(0)); final InstanceSpec secondInstance = zkServer.findConnectionInstance(zooKeeper); assertThat(firstInstance, is(not(secondInstance))); zkServer.killServer(secondInstance); boolean retry; do { System.out.println("Checking node with Curator... This might take a while..."); try { final Stat stat = curator.checkExists().forPath(pathPrefix); retry = false; assertThat(stat, is(notNullValue())); } catch (CuratorConnectionLossException ex) { System.out.println("Missing connection. Retrying"); retry = true; } } while (retry); zkServer.restartServer(firstInstance); zkServer.restartServer(secondInstance); listener.createLatch = new CountDownLatch(1); System.out.println("Creating node via Curator... This might take a while..."); curator.create().forPath(pathPrefix + "/imback", getData()); assertTrue(listener.createLatch.await(maxPropagationTime, TimeUnit.MILLISECONDS)); assertThat(listener.createCount.get(), is(nodeCount + 2)); assertThat(listener.removeCount.get(), is(0)); assertThat(listener.dataCount.get(), is(0)); for (int i = 0; i < nodeCount; i++) { final Stat stat = curator.setData().forPath(pathPrefix + "/node" + i, getData()); assertThat(stat, is(notNullValue())); } assertTrue(listener.dataLatch.await(maxPropagationTime, TimeUnit.MILLISECONDS)); assertThat(listener.createCount.get(), is(nodeCount + 2)); assertThat(listener.removeCount.get(), is(0)); assertThat(listener.dataCount.get(), is(nodeCount)); for (int i = 0; i < nodeCount; i++) { curator.delete().forPath(pathPrefix + "/node" + i); } assertTrue(listener.removeLatch.await(maxPropagationTime, TimeUnit.MILLISECONDS)); assertThat(listener.createCount.get(), is(nodeCount + 2)); assertThat(listener.removeCount.get(), is(nodeCount)); assertThat(listener.dataCount.get(), is(nodeCount)); nodeCollectionWatcher.shutdown(); }
@Test public void misbehavingClient() throws Exception { final int propagationTime = 5; final AtomicBoolean triggerExceptions = new AtomicBoolean(false); final CountDownLatch createLatch = new CountDownLatch(1); final CountDownLatch dataLatch = new CountDownLatch(1); final CountDownLatch removeLatch = new CountDownLatch(1); final NodeWatcherListener listener = new NodeWatcherListener() { @Override public void nodeCreated(String zkPath, String data) { if (triggerExceptions.get()) { throw new RuntimeException("boo!"); } createLatch.countDown(); } @Override public void dataChanged(String zkPath, String data) { if (triggerExceptions.get()) { throw new RuntimeException("boo!"); } dataLatch.countDown(); } @Override public void nodeRemoved(String zkPath) { if (triggerExceptions.get()) { throw new RuntimeException("boo!"); } removeLatch.countDown(); } }; final String pathPrefix = "/foo/misbehaving"; curator.create().creatingParentsIfNeeded().forPath(pathPrefix); final NodeCollectionWatcher nodeCollectionWatcher = new NodeCollectionWatcher(zooKeeper, pathPrefix, listener); triggerExceptions.set(true); curator.create().forPath(pathPrefix + "/first", getData()); Thread.sleep(propagationTime); curator.setData().forPath(pathPrefix + "/first", getData()); Thread.sleep(propagationTime); curator.delete().forPath(pathPrefix + "/first"); Thread.sleep(propagationTime); triggerExceptions.set(false); curator.create().forPath(pathPrefix + "/second"); assertTrue(createLatch.await(propagationTime, TimeUnit.MILLISECONDS)); curator.setData().forPath(pathPrefix + "/second", getData()); assertTrue(dataLatch.await(propagationTime, TimeUnit.MILLISECONDS)); curator.delete().forPath(pathPrefix + "/second"); assertTrue(removeLatch.await(propagationTime, TimeUnit.MILLISECONDS)); nodeCollectionWatcher.shutdown(); }