@GET
  @Path("/tasks/subtask/progress")
  @Produces(MediaType.APPLICATION_JSON)
  @ApiOperation("Get finer grained task progress tracked in memory for the given subtasks")
  @ApiResponses(value = {
      @ApiResponse(code = 200, message = "Success"), @ApiResponse(code = 500, message = "Internal server error")
  })
  public String getSubtaskProgress(
      @ApiParam(value = "Sub task names separated by comma") @QueryParam("subtaskNames") String subtaskNames) {
    try {
      LOGGER.debug("Getting progress for subtasks: {}", subtaskNames);
      Map<String, Object> progress = new HashMap<>();
      for (String subtaskName : StringUtils.split(subtaskNames, CommonConstants.Minion.TASK_LIST_SEPARATOR)) {
        MinionEventObserver observer = MinionEventObservers.getInstance().getMinionEventObserver(subtaskName);
        if (observer != null) {
          progress.put(subtaskName, observer.getProgress());
        }
      }
      LOGGER.debug("Got subtasks progress: {}", progress);
      return JsonUtils.objectToString(progress);
    } catch (Exception e) {
      throw new WebApplicationException(Response.status(Response.Status.INTERNAL_SERVER_ERROR).entity(
          String.format("Failed to get task progress for subtasks: %s due to error: %s", subtaskNames, e.getMessage()))
          .build());
    }
  }
@Test
  public void testGetGivenSubtaskOrStateProgress()
      throws IOException {
    MinionEventObserver observer1 = new MinionProgressObserver();
    observer1.notifyTaskStart(null);
    MinionEventObservers.getInstance().addMinionEventObserver("t01", observer1);

    MinionEventObserver observer2 = new MinionProgressObserver();
    observer2.notifyProgress(null, "");
    MinionEventObservers.getInstance().addMinionEventObserver("t02", observer2);

    MinionEventObserver observer3 = new MinionProgressObserver();
    observer3.notifyTaskSuccess(null, "");
    MinionEventObservers.getInstance().addMinionEventObserver("t03", observer3);

    PinotTaskProgressResource pinotTaskProgressResource = new PinotTaskProgressResource();

    // get all sub task progress
    String allSubTaskProgress = pinotTaskProgressResource.getSubtaskProgress(null, null);
    Map<String, Object> subtaskProgressMap = JsonUtils.stringToObject(allSubTaskProgress, Map.class);
    assertEquals(subtaskProgressMap.size(), 3);

    // get subtasks with given state
    String subtaskWithInProgressState =
        pinotTaskProgressResource.getSubtaskProgress(null, MinionTaskState.IN_PROGRESS.toString());
    assertInProgressSubtasks(subtaskWithInProgressState);

    String subtaskWithUndefinedState =
        pinotTaskProgressResource.getSubtaskProgress(null, "Undefined");
    assertInProgressSubtasks(subtaskWithUndefinedState);

    String subtaskWithSucceededState =
        pinotTaskProgressResource.getSubtaskProgress(null, MinionTaskState.SUCCEEDED.toString());
    subtaskProgressMap = JsonUtils.stringToObject(subtaskWithSucceededState, Map.class);
    assertEquals(subtaskProgressMap.size(), 1);

    String subtaskWithUnknownState =
        pinotTaskProgressResource.getSubtaskProgress(null, MinionTaskState.UNKNOWN.toString());
    assertNoSubtaskWithTheGivenState(subtaskWithUnknownState);

    String subtaskWithCancelledState =
        pinotTaskProgressResource.getSubtaskProgress(null, MinionTaskState.CANCELLED.toString());
    assertNoSubtaskWithTheGivenState(subtaskWithCancelledState);

    String subtaskWithErrorState =
        pinotTaskProgressResource.getSubtaskProgress(null, MinionTaskState.ERROR.toString());
    assertNoSubtaskWithTheGivenState(subtaskWithErrorState);

    // get subtasks with given name
    String subTasksWithGivenNamesProgress = pinotTaskProgressResource.getSubtaskProgress(" t01 , t02 ", null);
    assertInProgressSubtasks(subTasksWithGivenNamesProgress);

    // get subtasks with given names and state
    assertThrows(WebApplicationException.class,
        () -> pinotTaskProgressResource.getSubtaskProgress(" t01 , t02 ", MinionTaskState.IN_PROGRESS.toString()));
  }