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 = 5000)
  public void testWarmup() throws Exception
  {
    ScheduledExecutorService warmupExecutorService = Executors.newSingleThreadScheduledExecutor();
    boolean isDarkWarmupEnabled = true;
    String warmupClusterName = "warmup" + _cluster;
    String expectedWarmupUriNodePath = "/d2/uris/" + warmupClusterName + "/ephemoral-0000000000";
    int warmupDuration = 2; //run warm-up for 2 seconds
    Map<Integer, PartitionData> partitions = Collections.singletonMap(0, new PartitionData(0.5));
    UriProperties warmupProperties = new UriProperties(warmupClusterName, Collections.singletonMap(URI.create(_uri), partitions));

    ZooKeeperAnnouncer announcer = getZooKeeperWarmupAnnouncer(_cluster, _uri, WEIGHT, isDarkWarmupEnabled, warmupClusterName, warmupDuration, warmupExecutorService);
    TestDataHelper.MockServiceDiscoveryEventEmitter eventEmitter = getMockServiceDiscoveryEventEmitter();
    announcer.setEventEmitter(eventEmitter);

    ZooKeeperConnectionManager manager = createManagerForWarmupTests(false, warmupDuration, announcer);
    ZooKeeperEphemeralStore<UriProperties> store = createAndStartUriStore(announcer);
    FutureCallback<None> managerStartCallback = new FutureCallback<>();
    manager.start(managerStartCallback);

    // Wait till warm up completes and announcer successfully marks up to the regular cluster
    managerStartCallback.get();

    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);

    // If warm up has happened, mark down for the warm up cluster should be successful
    UriProperties warmupPropertiesAfterMarkdown = store.get(warmupClusterName);
    assertNotNull(warmupPropertiesAfterMarkdown);
    assertEquals(warmupPropertiesAfterMarkdown.Uris().size(), 0);

    List<ServiceDiscoveryEventEmitter.StatusUpdateActionType> expectedActionTypes = Arrays.asList(
        ServiceDiscoveryEventEmitter.StatusUpdateActionType.MARK_READY,
        ServiceDiscoveryEventEmitter.StatusUpdateActionType.MARK_DOWN,
        ServiceDiscoveryEventEmitter.StatusUpdateActionType.MARK_READY);
    List<String> expectedUriNodePaths = Arrays.asList(expectedWarmupUriNodePath, expectedWarmupUriNodePath, _expectedUriNodePath);
    eventEmitter.verifySDStatusActiveUpdateIntentEvents(
        Arrays.asList(Collections.singletonList(warmupClusterName), Collections.singletonList(warmupClusterName), Collections.singletonList(_cluster)),
        expectedActionTypes,
        expectedUriNodePaths
    );
    List<String> expectedWriteClusters = Arrays.asList(warmupClusterName, warmupClusterName, _cluster);
    eventEmitter.verifySDStatusWriteEvents(expectedWriteClusters, Arrays.asList(_cluster, _cluster, _cluster),
        expectedActionTypes, expectedUriNodePaths, Arrays.asList(warmupProperties.toString(), warmupProperties.toString(), properties.toString()),
        Arrays.asList(0, 0, 0), expectedUriNodePaths, Arrays.asList(true, true, true));

    shutdownManager(manager);
  }