public void start(Callback<None> callback)
  {
    _managerStarted = true;
    if (!_startupCallback.compareAndSet(null, callback))
    {
      throw new IllegalStateException("Already starting");
    }
    try
    {
      _zkConnection.start();
      //Trying to start store here. If the connection is not ready, will return immediately.
      //The connection event will trigger the actual store startup
      tryStartStore();
      LOG.info("Started ZooKeeper connection to {}", _zkConnectString);
    }
    catch (Exception e)
    {
      _startupCallback.set(null);
      callback.onError(e);
    }
  }
@Test (invocationCount = 1, timeOut = 15000)
  public void testWarmupDuringSessionExpiration() throws Exception
  {
    ScheduledExecutorService warmupExecutorService = Executors.newSingleThreadScheduledExecutor();
    boolean isDarkWarmupEnabled = true;
    String warmupClusterName = "warmup" + _cluster;
    int warmupDuration = 5; //run warm up for 5 sec
    final double newWeight = 1.5d;

    ZooKeeperAnnouncer announcer = getZooKeeperWarmupAnnouncer(_cluster, _uri, WEIGHT, isDarkWarmupEnabled, warmupClusterName, warmupDuration, warmupExecutorService);
    ZKPersistentConnection zkPersistentConnection = getZkPersistentConnection();
    ZooKeeperConnectionManager manager = createManagerForWarmupTests(false, zkPersistentConnection, warmupDuration, announcer);
    ZooKeeperEphemeralStore<UriProperties> store = createAndStartUriStore();
    FutureCallback<None> managerStartCallback = new FutureCallback<>();
    manager.start(managerStartCallback);

    // Ensure warm-up has begin before expiring the session
    try
    {
      managerStartCallback.get(1, TimeUnit.SECONDS);
    }
    catch (TimeoutException e)
    {
      // We are expecting TimeoutException here because the warmup is set to run for 5 seconds,
      // but we are getting the result from the callback after 1 sec, so the warm up should not have completed
    }
    UriProperties properties = store.get(warmupClusterName);
    assertNotNull(properties);
    assertEquals(properties.getPartitionDataMap(URI.create(_uri)).get(DefaultPartitionAccessor.DEFAULT_PARTITION_ID).getWeight(), WEIGHT);
    assertEquals(properties.Uris().size(), 1);

    // the new WEIGHT will be picked up only if the connection is re-established
    announcer.setWeight(newWeight);

    // expiring the session
    long oldSessionId = zkPersistentConnection.getZooKeeper().getSessionId();
    ZKTestUtil.expireSession("localhost:" + PORT, zkPersistentConnection.getZooKeeper(), 10, TimeUnit.SECONDS);

    // making sure that a new session has been established.
    ZKTestUtil.waitForNewSessionEstablished(oldSessionId, zkPersistentConnection, 10, TimeUnit.SECONDS);

    // Validate the after new session creation, mark up has completed
    // Warm up will run again in this case as part of mark up for the new session
    AssertionMethods.assertWithTimeout((warmupDuration+1)*1000, () -> {
      UriProperties newProperties = store.get(_cluster);
      assertNotNull(newProperties);
      if (newProperties.getPartitionDataMap(URI.create(_uri)) == null)
      {
        Assert.fail("Supposed to have the uri present in ZK");
      }
      assertEquals(newProperties.getPartitionDataMap(URI.create(_uri)).get(DefaultPartitionAccessor.DEFAULT_PARTITION_ID).getWeight(), newWeight);
      assertEquals(newProperties.Uris().size(), 1);

      newProperties = store.get(warmupClusterName);
      assertNotNull(newProperties);
      assertEquals(newProperties.Uris().size(), 0);
    });

    shutdownManager(manager);
  }