@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);
    // Create should trigger create notification (and no other notification)
    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));
    // Data change should trigger the data notification (and no other notification)
    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));
    // Delete should trigger the remove notification (and no other notification)
    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();
    // Ensure that there are no notifications when the watcher shuts down
    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));
}