public void shutdown(final Callback<None> callback)
  {
    _managerStarted = false;
    for (ZooKeeperAnnouncer server : _servers)
    {
      server.shutdown();
    }
    Callback<None> zkCloseCallback = new CallbackAdapter<None, None>(callback)
    {
      @Override
      protected None convertResponse(None none) throws Exception
      {
        _zkConnection.shutdown();
        return none;
      }
    };
    if (_store != null)
    {
      _store.shutdown(zkCloseCallback);
    }
    else
    {
      zkCloseCallback.onSuccess(None.none());
    }
  }
@Test
  public void testMarkDownAndUpDuringDisconnection()
    throws Exception
  {
    ZooKeeperAnnouncer announcer = getZooKeeperAnnouncer(_cluster, _uri, WEIGHT);

    ZooKeeperConnectionManager manager = createManager(true, announcer);

    ZooKeeperEphemeralStore<UriProperties> store = createAndStartUriStore();
    UriProperties properties = store.get(_cluster);
    assertNotNull(properties);
    assertEquals(properties.getPartitionDataMap(URI.create(_uri)).get(DefaultPartitionAccessor.DEFAULT_PARTITION_ID).getWeight(), WEIGHT);
    assertEquals(properties.Uris().size(), 1);

    _zkServer.shutdown(false);

    FutureCallback<None> markDownCallback = new FutureCallback<>();
    announcer.markDown(markDownCallback);
    FutureCallback<None> markUpCallback = new FutureCallback<>();
    announcer.markUp(markUpCallback);

    // ugly, but we need to wait for a while just so that Disconnect event is propagated
    // to the caller before we restart zk sever.
    Thread.sleep(1000);
    _zkServer.restart();
    markUpCallback.get(10, TimeUnit.SECONDS);
    try
    {
      markDownCallback.get();
      Assert.fail("mark down should have thrown CancellationException.");
    }
    catch (ExecutionException e)
    {
      Assert.assertTrue(e.getCause() instanceof CancellationException);
    }

    properties = store.get(_cluster);
    assertNotNull(properties);
    assertEquals(properties.getPartitionDataMap(URI.create(_uri)).get(DefaultPartitionAccessor.DEFAULT_PARTITION_ID).getWeight(), WEIGHT);
    assertEquals(properties.Uris().size(), 1);

    shutdownManager(manager);
  }