@Override
  public void shutdown(Callback<None> shutdown)
  {
    MultiCallback multiCallback = new MultiCallback(shutdown, 2);
    _fsStore.shutdown(multiCallback);
    _zkAwareStore.shutdown(multiCallback);
  }
@Test
  public void testLastSeenLifeCycle()
      throws InterruptedException, ExecutionException, TimeoutException, IOException, PropertyStoreException
  {
    createZKServer();
    // Fill the store with data
    File dataPath = ZKTestUtil.createTempDir("randomFileDataPath");
    LastSeenZKStore<String> store = ZkStoreTestOnlyUtil.getLastSeenZKStore(dataPath.getPath(), PORT);

    ZooKeeperEphemeralStore<String> storeWriter = ZkStoreTestOnlyUtil.getZooKeeperEphemeralStore(PORT);
    storeWriter.put(TEST_ZK_PROP_NAME, "randomData");

    PropertyEventBusImpl<String>  propertyEventBus = new PropertyEventBusImpl<>(Executors.newSingleThreadExecutor());
    propertyEventBus.setPublisher(store);

    CountDownLatch initializedLatch = new CountDownLatch(1);
    propertyEventBus.register(Collections.singleton(TEST_ZK_PROP_NAME), new LatchSubscriber(initializedLatch, null));
    initializedLatch.await(5, TimeUnit.SECONDS);
    if (initializedLatch.getCount() != 0)
    {
      fail("Initialized not received");
    }

    // stopping ZK without removing data. This make ZK unreachable

    _zkServer.shutdown(false);

    // create new last seen, without ZK Connection, and see if it fetches from the server

    store = ZkStoreTestOnlyUtil.getLastSeenZKStore(dataPath.getPath(), PORT);

    propertyEventBus = new PropertyEventBusImpl<>(Executors.newSingleThreadExecutor());
    propertyEventBus.setPublisher(store);

    CountDownLatch initializedLatch2 = new CountDownLatch(1);
    CountDownLatch addLatch2 = new CountDownLatch(1);
    propertyEventBus.register(Collections.singleton(TEST_ZK_PROP_NAME),
        new LatchSubscriber(initializedLatch2, addLatch2));

    initializedLatch2.await(5, TimeUnit.SECONDS);
    if (initializedLatch2.getCount() != 0)
    {
      fail("Initialized not received");
    }

    if (addLatch2.getCount() != 1)
    {
      fail("The add latch should have not been invoked yet");
    }

    // restart ZK and see if it reads the most updated value, the most updated value in this case is identical
    _zkServer.restart();

    addLatch2.await(50, TimeUnit.SECONDS);
    if (addLatch2.getCount() != 0)
    {
      fail("When ZK restarted we didn't read the most updated value from ZK");
    }

    // shutting everything down
    final FutureCallback<None> shutdownCallback = new FutureCallback<>();
    store.shutdown(shutdownCallback);
    shutdownCallback.get(5, TimeUnit.SECONDS);

    final FutureCallback<None> shutdownCallback2 = new FutureCallback<>();
    storeWriter.shutdown(shutdownCallback2);
    shutdownCallback2.get(5, TimeUnit.SECONDS);

    _zkServer.shutdown();
  }