public static void forRunningTasks(String tableNameWithType, String taskType, ClusterInfoAccessor clusterInfoAccessor,
      Consumer<Map<String, String>> taskConfigConsumer) {
    Map<String, TaskState> taskStates = clusterInfoAccessor.getTaskStates(taskType);
    for (Map.Entry<String, TaskState> entry : taskStates.entrySet()) {
      if (TASK_FINAL_STATES.contains(entry.getValue())) {
        continue;
      }
      String taskName = entry.getKey();
      for (PinotTaskConfig pinotTaskConfig : clusterInfoAccessor.getTaskConfigs(taskName)) {
        Map<String, String> config = pinotTaskConfig.getConfigs();
        String tableNameFromTaskConfig = config.get(MinionConstants.TABLE_NAME_KEY);
        if (tableNameWithType.equals(tableNameFromTaskConfig)) {
          taskConfigConsumer.accept(config);
        }
      }
    }
  }
@Test
  public void testForRunningTasks() {
    String tableName = "mytable_OFFLINE";
    String taskType = "myTaskType";
    ClusterInfoAccessor mockClusterInfoAccessor = createMockClusterInfoAccessor();

    Map<String, TaskState> taskStatesMap = new HashMap<>();
    String taskID = System.currentTimeMillis() + "_0";
    when(mockClusterInfoAccessor.getTaskStates(taskType)).thenReturn(taskStatesMap);
    when(mockClusterInfoAccessor.getTaskConfigs(taskID))
        .thenReturn(Collections.singletonList(createTaskConfig(taskType, tableName, taskID)));

    int[] count = new int[1];
    TaskState[] nonFinalTaskStates = new TaskState[]{
        TaskState.NOT_STARTED, TaskState.IN_PROGRESS, TaskState.FAILING, TaskState.STOPPING, TaskState.STOPPED,
        TaskState.TIMING_OUT
    };
    for (TaskState taskState : nonFinalTaskStates) {
      taskStatesMap.put(taskID, taskState);
      TaskGeneratorUtils.forRunningTasks(tableName, taskType, mockClusterInfoAccessor, taskConfig -> {
        assertEquals(taskConfig.get(MinionConstants.TABLE_NAME_KEY), tableName);
        assertEquals(taskConfig.get("taskID"), taskID);
        count[0]++;
      });
    }
    assertEquals(count[0], nonFinalTaskStates.length);
    for (TaskState taskState : new TaskState[]{
        TaskState.COMPLETED, TaskState.FAILED, TaskState.ABORTED, TaskState.TIMED_OUT
    }) {
      taskStatesMap.put(taskID, taskState);
      TaskGeneratorUtils.forRunningTasks(tableName, taskType, mockClusterInfoAccessor, taskConfig -> {
        fail("Task should be in final state");
      });
    }
    TaskGeneratorUtils.forRunningTasks("fooTable", taskType, mockClusterInfoAccessor, taskConfig -> {
      fail("Different table name");
    });
    TaskGeneratorUtils.forRunningTasks(tableName, "fooTask", mockClusterInfoAccessor, taskConfig -> {
      fail("Different task type");
    });
  }