@Override
  protected final void runTask(Properties periodicTaskProperties) {
    // Make it so that only one controller returns the metric for all the tasks.
    if (!_leadControllerManager.isLeaderForTable(TASK_NAME)) {
      return;
    }

    Map<String, Map<String, Long>> taskMetadataLastUpdateTime =
        _helixTaskResourceManager.getTaskMetadataLastUpdateTimeMs();
    taskMetadataLastUpdateTime.forEach((tableNameWithType, taskTypeLastUpdateTime) ->
        taskTypeLastUpdateTime.forEach((taskType, lastUpdateTimeMs) ->
            _controllerMetrics.setOrUpdateTableGauge(tableNameWithType, taskType,
                ControllerGauge.TIME_MS_SINCE_LAST_MINION_TASK_METADATA_UPDATE,
                () -> System.currentTimeMillis() - lastUpdateTimeMs)));

    // The call to get task types can take time if there are a lot of tasks.
    // Potential optimization is to call it every (say) 30m if we detect a barrage of
    // zk requests.
    Set<String> taskTypes = _helixTaskResourceManager.getTaskTypes();
    for (String taskType : taskTypes) {
      TaskCount taskTypeAccumulatedCount = new TaskCount();
      Map<String, TaskCount> tableAccumulatedCount = new HashMap<>();
      try {
        Set<String> tasksInProgress = _helixTaskResourceManager.getTasksInProgress(taskType);
        final int numRunningTasks = tasksInProgress.size();
        for (String task : tasksInProgress) {
          Map<String, TaskCount> tableTaskCount = _helixTaskResourceManager.getTableTaskCount(task);
          tableTaskCount.forEach((tableNameWithType, taskCount) -> {
            taskTypeAccumulatedCount.accumulate(taskCount);
            tableAccumulatedCount.compute(tableNameWithType, (name, count) -> {
              if (count == null) {
                count = new TaskCount();
              }
              count.accumulate(taskCount);
              return count;
            });
          });
        }
        // Emit metrics for taskType.
        _controllerMetrics.setValueOfGlobalGauge(ControllerGauge.NUM_MINION_TASKS_IN_PROGRESS, taskType,
            numRunningTasks);
        _controllerMetrics.setValueOfGlobalGauge(ControllerGauge.NUM_MINION_SUBTASKS_RUNNING, taskType,
            taskTypeAccumulatedCount.getRunning());
        _controllerMetrics.setValueOfGlobalGauge(ControllerGauge.NUM_MINION_SUBTASKS_WAITING, taskType,
            taskTypeAccumulatedCount.getWaiting());
        _controllerMetrics.setValueOfGlobalGauge(ControllerGauge.NUM_MINION_SUBTASKS_ERROR, taskType,
            taskTypeAccumulatedCount.getError());
        int total = taskTypeAccumulatedCount.getTotal();
        int percent = total != 0
            ? (taskTypeAccumulatedCount.getWaiting() + taskTypeAccumulatedCount.getRunning()) * 100 / total : 0;
        _controllerMetrics.setValueOfGlobalGauge(ControllerGauge.PERCENT_MINION_SUBTASKS_IN_QUEUE, taskType, percent);
        percent = total != 0 ? taskTypeAccumulatedCount.getError() * 100 / total : 0;
        _controllerMetrics.setValueOfGlobalGauge(ControllerGauge.PERCENT_MINION_SUBTASKS_IN_ERROR, taskType, percent);

        // Emit metrics for table taskType
        tableAccumulatedCount.forEach((tableNameWithType, taskCount) -> {
          _controllerMetrics.setOrUpdateTableGauge(tableNameWithType, taskType,
              ControllerGauge.NUM_MINION_SUBTASKS_RUNNING, () -> (long) taskCount.getRunning());
          _controllerMetrics.setOrUpdateTableGauge(tableNameWithType, taskType,
              ControllerGauge.NUM_MINION_SUBTASKS_WAITING, taskCount.getWaiting());
          _controllerMetrics.setOrUpdateTableGauge(tableNameWithType, taskType,
              ControllerGauge.NUM_MINION_SUBTASKS_ERROR, taskCount.getError());
          int tableTotal = taskCount.getTotal();
          int tablePercent = tableTotal != 0 ? (taskCount.getWaiting() + taskCount.getRunning()) * 100 / tableTotal : 0;
          _controllerMetrics.setOrUpdateTableGauge(tableNameWithType, taskType,
              ControllerGauge.PERCENT_MINION_SUBTASKS_IN_QUEUE, tablePercent);
          tablePercent = tableTotal != 0 ? taskCount.getError() * 100 / tableTotal : 0;
          _controllerMetrics.setOrUpdateTableGauge(tableNameWithType, taskType,
              ControllerGauge.PERCENT_MINION_SUBTASKS_IN_ERROR, tablePercent);
        });

        if (_preReportedTables.containsKey(taskType)) {
          Set<String> tableNameWithTypeSet = _preReportedTables.get(taskType);
          tableNameWithTypeSet.removeAll(tableAccumulatedCount.keySet());
          removeTableTaskTypeMetrics(tableNameWithTypeSet, taskType);
        }
        if (!tableAccumulatedCount.isEmpty()) {
          // need to make a copy of the set because we may want to chagne the set later
          Set<String> tableNameWithTypeSet = new HashSet<>(tableAccumulatedCount.keySet());
          _preReportedTables.put(taskType, tableNameWithTypeSet);
        } else {
          _preReportedTables.remove(taskType);
        }
      } catch (Exception e) {
        LOGGER.error("Caught exception while getting metrics for task type {}", taskType, e);
      }
    }

    // clean up metrics for task types that have already been removed
    _preReportedTaskTypes.removeAll(taskTypes);
    for (String taskType : _preReportedTaskTypes) {
      // remove task type level gauges
      _controllerMetrics.removeGlobalGauge(taskType, ControllerGauge.NUM_MINION_TASKS_IN_PROGRESS);
      _controllerMetrics.removeGlobalGauge(taskType, ControllerGauge.NUM_MINION_SUBTASKS_RUNNING);
      _controllerMetrics.removeGlobalGauge(taskType, ControllerGauge.NUM_MINION_SUBTASKS_WAITING);
      _controllerMetrics.removeGlobalGauge(taskType, ControllerGauge.NUM_MINION_SUBTASKS_ERROR);
      _controllerMetrics.removeGlobalGauge(taskType, ControllerGauge.PERCENT_MINION_SUBTASKS_IN_QUEUE);
      _controllerMetrics.removeGlobalGauge(taskType, ControllerGauge.PERCENT_MINION_SUBTASKS_IN_ERROR);
      // remove table task type level gauges
      if (_preReportedTables.containsKey(taskType)) {
        removeTableTaskTypeMetrics(_preReportedTables.get(taskType), taskType);
        _preReportedTables.remove(taskType);
      }
    }

    // update previously reported task types
    _preReportedTaskTypes.clear();
    _preReportedTaskTypes.addAll(taskTypes);

    // Emit metric to count the number of online minion instances.
    List<String> onlineInstances = _pinotHelixResourceManager.getOnlineInstanceList();
    int onlineMinionInstanceCount = 0;
    for (String onlineInstance : onlineInstances) {
      if (InstanceTypeUtils.isMinion(onlineInstance)) {
        onlineMinionInstanceCount++;
      }
    }
    _controllerMetrics.setValueOfGlobalGauge(ControllerGauge.ONLINE_MINION_INSTANCES, onlineMinionInstanceCount);
  }
@Test
  public void noTaskTypeMetrics() {
    PinotMetricsRegistry metricsRegistry = _controllerMetrics.getMetricsRegistry();
    Mockito.when(_pinotHelixTaskResourceManager.getTaskTypes()).thenReturn(ImmutableSet.of());
    _taskMetricsEmitter.runTask(null);
    Assert.assertEquals(metricsRegistry.allMetrics().size(), 1);
    Assert.assertTrue(metricsRegistry.allMetrics().containsKey(
        new YammerMetricName(ControllerMetrics.class, "pinot.controller.onlineMinionInstances")));
  }