public LeadControllerManager(String helixControllerInstanceId, HelixManager helixManager,
      ControllerMetrics controllerMetrics) {
    _helixControllerInstanceId = helixControllerInstanceId;
    _helixManager = helixManager;
    _controllerMetrics = controllerMetrics;
    _leadForPartitions = ConcurrentHashMap.newKeySet();

    // Create a thread to periodically fetch controller leadership as a work-around of Helix callback delay
    _controllerLeadershipFetchingThread = new Thread("ControllerLeadershipFetchingThread") {
      @Override
      public void run() {
        while (true) {
          try {
            synchronized (LeadControllerManager.this) {
              if (_isShuttingDown) {
                return;
              }
              if (isHelixLeader()) {
                if (!_amIHelixLeader) {
                  _amIHelixLeader = true;
                  LOGGER.warn("Becoming leader without getting Helix change callback");
                  _controllerMetrics
                      .addMeteredGlobalValue(ControllerMeter.CONTROLLER_LEADERSHIP_CHANGE_WITHOUT_CALLBACK, 1L);
                }
                _controllerMetrics.setValueOfGlobalGauge(ControllerGauge.PINOT_CONTROLLER_LEADER, 1L);
              } else {
                if (_amIHelixLeader) {
                  _amIHelixLeader = false;
                  LOGGER.warn("Losing leadership without getting Helix change callback");
                  _controllerMetrics
                      .addMeteredGlobalValue(ControllerMeter.CONTROLLER_LEADERSHIP_CHANGE_WITHOUT_CALLBACK, 1L);
                }
                _controllerMetrics.setValueOfGlobalGauge(ControllerGauge.PINOT_CONTROLLER_LEADER, 0L);
              }
              LeadControllerManager.this.wait(CONTROLLER_LEADERSHIP_FETCH_INTERVAL_MS);
            }
          } catch (Exception e) {
            // Ignore all exceptions. The thread keeps running until LeadControllerManager.stop() is invoked.
            LOGGER.error("Caught exception within controller leadership fetching thread", e);
          }
        }
      }
    };
  }
@Test
  public void testLeadControllerManager() {
    LeadControllerManager leadControllerManager =
        new LeadControllerManager(HELIX_CONTROLLER_INSTANCE_ID, _helixManager, _controllerMetrics);
    String tableName = "leadControllerTestTable";
    int expectedPartitionIndex = LeadControllerUtils.getPartitionIdForTable(tableName);
    String partitionName = LeadControllerUtils.generatePartitionName(expectedPartitionIndex);

    becomeHelixLeader(false);
    leadControllerManager.onHelixControllerChange();

    // When there's no resource config change nor helix controller change, leadControllerManager should return false.
    Assert.assertFalse(leadControllerManager.isLeaderForTable(tableName));

    enableResourceConfig(true);
    leadControllerManager.onResourceConfigChange();

    // Even resource config is enabled, leadControllerManager should return false because no index is cached yet.
    Assert.assertFalse(leadControllerManager.isLeaderForTable(tableName));
    Assert.assertTrue(LeadControllerUtils.isLeadControllerResourceEnabled(_helixManager));

    // After the target partition index is cached, leadControllerManager should return true.
    leadControllerManager.addPartitionLeader(partitionName);
    Assert.assertTrue(leadControllerManager.isLeaderForTable(tableName));

    // When the target partition index is removed, leadControllerManager should return false.
    leadControllerManager.removePartitionLeader(partitionName);
    Assert.assertFalse(leadControllerManager.isLeaderForTable(tableName));

    // When resource config is set to false, the cache should be disabled, even if the target partition index is in
    // the cache.
    // The leader depends on whether the current controller is helix leader.
    enableResourceConfig(false);
    leadControllerManager.onResourceConfigChange();

    Assert.assertFalse(LeadControllerUtils.isLeadControllerResourceEnabled(_helixManager));
    Assert.assertFalse(leadControllerManager.isLeaderForTable(tableName));
    leadControllerManager.addPartitionLeader(partitionName);
    Assert.assertFalse(leadControllerManager.isLeaderForTable(tableName));

    // When the current controller becomes helix leader and resource is disabled, leadControllerManager should return
    // true.
    becomeHelixLeader(true);
    leadControllerManager.onHelixControllerChange();
    Assert.assertTrue(leadControllerManager.isLeaderForTable(tableName));
  }