public synchronized Map<String, Object> getSubtaskOnWorkerProgress(String subtaskState, Executor executor,
      HttpClientConnectionManager connMgr, Map<String, String> selectedMinionWorkerEndpoints,
      Map<String, String> requestHeaders, int timeoutMs)
      throws JsonProcessingException {
    return getSubtaskOnWorkerProgress(subtaskState, new CompletionServiceHelper(executor, connMgr, HashBiMap.create(0)),
        selectedMinionWorkerEndpoints, requestHeaders, timeoutMs);
  }
@Test
  public void testGetSubtaskWithGivenStateProgress()
      throws IOException {
    CompletionServiceHelper httpHelper = mock(CompletionServiceHelper.class);
    CompletionServiceHelper.CompletionServiceResponse httpResp =
        new CompletionServiceHelper.CompletionServiceResponse();
    String taskIdPrefix = "Task_SegmentGenerationAndPushTask_someone";
    String workerIdPrefix = "worker";
    String[] subtaskIds = new String[6];
    String[] workerIds = new String[3];
    Map<String, String> selectedMinionWorkerEndpoints = new HashMap<>();
    for (int i = 0; i < 3; i++) {
      workerIds[i] = workerIdPrefix + i;
      String workerEndpoint = "http://" + workerIds[i] + ":9000";
      selectedMinionWorkerEndpoints.put(workerIds[i], workerEndpoint);

      subtaskIds[2 * i] = taskIdPrefix + "_" + (2 * i);
      subtaskIds[2 * i + 1] = taskIdPrefix + "_" + (2 * i + 1);
      // Notice that for testing purpose, we map subtask names to empty strings. In reality, subtask names will be
      // mapped to jsonized org.apache.pinot.minion.event.MinionEventObserver
      httpResp._httpResponses.put(
          String.format("%s/tasks/subtask/state/progress?subTaskState=IN_PROGRESS", workerEndpoint),
          JsonUtils.objectToString(ImmutableMap.of(subtaskIds[2 * i], "", subtaskIds[2 * i + 1], "")));
    }
    httpResp._failedResponseCount = 1;
    ArgumentCaptor<List<String>> workerEndpointCaptor = ArgumentCaptor.forClass(List.class);
    when(httpHelper.doMultiGetRequest(workerEndpointCaptor.capture(), any(), anyBoolean(), any(), anyInt())).thenReturn(
        httpResp);

    PinotHelixTaskResourceManager mgr =
        new PinotHelixTaskResourceManager(mock(PinotHelixResourceManager.class), mock(TaskDriver.class));
    Map<String, Object> progress =
        mgr.getSubtaskOnWorkerProgress("IN_PROGRESS", httpHelper, selectedMinionWorkerEndpoints, Collections.emptyMap(),
            1000);
    List<String> value = workerEndpointCaptor.getValue();
    Set<String> expectedWorkerUrls = selectedMinionWorkerEndpoints.values().stream().map(
            workerEndpoint -> String.format("%s/tasks/subtask/state/progress?subTaskState=IN_PROGRESS", workerEndpoint))
        .collect(Collectors.toSet());
    assertEquals(new HashSet<>(value), expectedWorkerUrls);
    assertEquals(progress.size(), 3);
    for (int i = 0; i < 3; i++) {
      Object responseFromMinionWorker = progress.get(workerIds[i]);
      Map<String, Object> subtaskProgressMap = (Map<String, Object>) responseFromMinionWorker;
      assertEquals(subtaskProgressMap.size(), 2);
      assertTrue(subtaskProgressMap.containsKey(subtaskIds[2 * i]));
      assertTrue(subtaskProgressMap.containsKey(subtaskIds[2 * i + 1]));
    }
  }