@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);
    // Create a few nodes to set the initial state
    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);
    // Client should reconnect to one of the two remaining
    curator.create().forPath(pathPrefix + "/stillalive", getData());
    // Wait for the notification to go through. This could take some time since there's
    // reconnects and all sorts of magic happening under the hood
    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));
    // Kill the 2nd server. The cluster won't have a quorum now
    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));
    // Ensure data notifications are propagated after a failure
    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));
    // ..and remove notifications are sent
    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();
}