@GET
  @Path("/tasks/subtask/workers/progress")
  @Authorize(targetType = TargetType.CLUSTER, action = Actions.Cluster.GET_TASK)
  @Produces(MediaType.APPLICATION_JSON)
  @ApiOperation("Get progress of all subtasks with specified state tracked by minion worker in memory")
  @ApiResponses(value = {
      @ApiResponse(code = 200, message = "Success"), @ApiResponse(code = 500, message = "Internal server error")
  })
  public String getSubtaskOnWorkerProgress(@Context HttpHeaders httpHeaders,
      @ApiParam(value = "Subtask state (UNKNOWN,IN_PROGRESS,SUCCEEDED,CANCELLED,ERROR)", required = true)
      @QueryParam("subTaskState") String subTaskState,
      @ApiParam(value = "Minion worker IDs separated by comma") @QueryParam("minionWorkerIds") @Nullable
          String minionWorkerIds) {
    Set<String> selectedMinionWorkers = new HashSet<>();
    if (StringUtils.isNotEmpty(minionWorkerIds)) {
      selectedMinionWorkers.addAll(
          Arrays.stream(StringUtils.split(minionWorkerIds, ',')).map(String::trim).collect(Collectors.toList()));
    }
    // Relying on original schema that was used to query the controller
    String scheme = _uriInfo.getRequestUri().getScheme();
    List<InstanceConfig> allMinionWorkerInstanceConfigs = _pinotHelixResourceManager.getAllMinionInstanceConfigs();
    Map<String, String> selectedMinionWorkerEndpoints = new HashMap<>();
    for (InstanceConfig worker : allMinionWorkerInstanceConfigs) {
      if (selectedMinionWorkers.isEmpty() || selectedMinionWorkers.contains(worker.getId())) {
        selectedMinionWorkerEndpoints.put(worker.getId(),
            String.format("%s://%s:%d", scheme, worker.getHostName(), Integer.parseInt(worker.getPort())));
      }
    }
    Map<String, String> requestHeaders = new HashMap<>();
    httpHeaders.getRequestHeaders().keySet().forEach(header ->
        requestHeaders.put(header, httpHeaders.getHeaderString(header)));
    int timeoutMs = _controllerConf.getMinionAdminRequestTimeoutSeconds() * 1000;
    try {
      Map<String, Object> minionWorkerIdSubtaskProgressMap =
          _pinotHelixTaskResourceManager.getSubtaskOnWorkerProgress(subTaskState, _executor, _connectionManager,
              selectedMinionWorkerEndpoints, requestHeaders, timeoutMs);
      return JsonUtils.objectToString(minionWorkerIdSubtaskProgressMap);
    } catch (Exception e) {
      throw new ControllerApplicationException(LOGGER,
          String.format("Failed to get minion worker side progress for subtasks with state %s due to error: %s",
              subTaskState, ExceptionUtils.getStackTrace(e)), Response.Status.INTERNAL_SERVER_ERROR, e);
    }
  }
@Test
  public void testGetSubtaskWithGivenStateProgressWithException()
      throws JsonProcessingException {
    when(_pinotHelixResourceManager.getAllMinionInstanceConfigs()).thenReturn(Collections.emptyList());
    HttpHeaders httpHeaders = Mockito.mock(HttpHeaders.class);
    when(httpHeaders.getRequestHeaders()).thenReturn(new MultivaluedHashMap<>());
    when(_pinotHelixTaskResourceManager
        .getSubtaskOnWorkerProgress(anyString(), any(), any(), anyMap(), anyMap(), anyInt()))
        .thenThrow(new RuntimeException());
    assertThrows(ControllerApplicationException.class,
        () -> _pinotTaskRestletResource.getSubtaskOnWorkerProgress(httpHeaders, "IN_PROGRESS", null));
  }