public synchronized Map<String, Object> getSubtaskProgress(String taskName, @Nullable String subtaskNames,
      Executor executor, HttpClientConnectionManager connMgr, Map<String, String> workerEndpoints,
      Map<String, String> requestHeaders, int timeoutMs)
      throws Exception {
    return getSubtaskProgress(taskName, subtaskNames,
        new CompletionServiceHelper(executor, connMgr, HashBiMap.create(0)), workerEndpoints, requestHeaders,
        timeoutMs);
  }
@Test
  public void testGetSubtaskProgressPending()
      throws Exception {
    TaskDriver taskDriver = mock(TaskDriver.class);
    JobConfig jobConfig = mock(JobConfig.class);
    when(taskDriver.getJobConfig(anyString())).thenReturn(jobConfig);
    JobContext jobContext = mock(JobContext.class);
    when(taskDriver.getJobContext(anyString())).thenReturn(jobContext);
    PinotHelixTaskResourceManager mgr =
        new PinotHelixTaskResourceManager(mock(PinotHelixResourceManager.class), taskDriver);
    CompletionServiceHelper httpHelper = mock(CompletionServiceHelper.class);
    CompletionServiceHelper.CompletionServiceResponse httpResp =
        new CompletionServiceHelper.CompletionServiceResponse();
    when(httpHelper.doMultiGetRequest(any(), any(), anyBoolean(), any(), anyInt())).thenReturn(httpResp);
    String[] workers = new String[]{"worker0", "worker1", "worker2"};
    Map<String, String> workerEndpoints = new HashMap<>();
    for (String worker : workers) {
      workerEndpoints.put(worker, "http://" + worker + ":9000");
    }
    String taskName = "Task_SegmentGenerationAndPushTask_someone";
    String[] subtaskNames = new String[3];
    Map<String, Integer> taskIdPartitionMap = new HashMap<>();
    for (int i = 0; i < 3; i++) {
      String subtaskName = taskName + "_" + i;
      subtaskNames[i] = subtaskName;
      taskIdPartitionMap.put(subtaskName, i);
    }
    Map<String, TaskConfig> taskConfigMap = new HashMap<>();
    for (String subtaskName : subtaskNames) {
      taskConfigMap.put(subtaskName, mock(TaskConfig.class));
    }
    when(jobConfig.getTaskConfigMap()).thenReturn(taskConfigMap);
    // Some subtasks are pending to be run
    httpResp._httpResponses.put(workers[0],
        JsonUtils.objectToString(Collections.singletonMap(subtaskNames[0], "running on worker: 0")));
    when(jobContext.getTaskIdPartitionMap()).thenReturn(taskIdPartitionMap);
    when(jobContext.getAssignedParticipant(0)).thenReturn(workers[0]);
    when(jobContext.getPartitionState(0)).thenReturn(TaskPartitionState.RUNNING);
    Map<String, Object> progress =
        mgr.getSubtaskProgress(taskName, StringUtils.join(subtaskNames, ','), httpHelper, workerEndpoints,
            Collections.emptyMap(), 1000);
    String taskProgress = (String) progress.get(subtaskNames[0]);
    assertEquals(taskProgress, "running on worker: 0");
    taskProgress = (String) progress.get(subtaskNames[1]);
    assertEquals(taskProgress, "No worker has run this subtask");
    taskProgress = (String) progress.get(subtaskNames[2]);
    assertEquals(taskProgress, "No worker has run this subtask");
  }