class TaskResourceProvider extends AbstractControllerResourceProvider {

  @Override
  public Set<Resource> getResources(Request request, Predicate predicate)
      throws SystemException, UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException {

    Set<Resource> results = new LinkedHashSet<>();
    Set<String> requestedIds = getRequestPropertyIds(request, predicate);

    List<HostRoleCommandEntity> entities = s_dao.findAll(request, predicate);

    // !!! getting the cluster name out of the request property maps is a little
    // hacky since there could be a different request per cluster name; however
    // nobody is making task requests across clusters. Overall, this loop could
    // be called multiple times when using predicates like
    // tasks/Tasks/status.in(FAILED,ABORTED,TIMEDOUT) which would unnecessarily
    // make the same call to the API over and over
    String clusterName = null;
    Long requestId = null;
    for (Map<String, Object> propertyMap : getPropertyMaps(predicate)) {
      clusterName = (String) propertyMap.get(TASK_CLUSTER_NAME_PROPERTY_ID);
      String requestIdStr = (String) propertyMap.get(TASK_REQUEST_ID_PROPERTY_ID);
      requestId = Long.parseLong(requestIdStr);
    }

    Collection<HostRoleCommand> commands = new ArrayList<>(100);

    if (!entities.isEmpty()) {
      for (HostRoleCommandEntity entity : entities) {
        commands.add(s_hostRoleCommandFactory.createExisting(entity));
      }
    } else {
      // if query has no results, look up in TopologyManager as the request might be a TopologyLogicalRequest
      // which is not directly linked to tasks
      if (requestId != null) {
        commands.addAll(s_topologyManager.getTasks(requestId));
      }
    }

    LOG.debug("Retrieved {} commands for request {}", commands.size(), request);

    // convert each entity into a response
    for (HostRoleCommand hostRoleCommand : commands) {
      Resource resource = new ResourceImpl(Resource.Type.Task);

      // !!! shocked this isn't broken.  the key can be null for non-cluster tasks
      if (null != clusterName) {
        setResourceProperty(resource, TASK_CLUSTER_NAME_PROPERTY_ID, clusterName, requestedIds);
      }

      setResourceProperty(resource, TASK_REQUEST_ID_PROPERTY_ID, hostRoleCommand.getRequestId(), requestedIds);
      setResourceProperty(resource, TASK_ID_PROPERTY_ID, hostRoleCommand.getTaskId(), requestedIds);
      setResourceProperty(resource, TASK_STAGE_ID_PROPERTY_ID, hostRoleCommand.getStageId(), requestedIds);
      setResourceProperty(resource, TASK_HOST_NAME_PROPERTY_ID, ensureHostname(hostRoleCommand.getHostName()), requestedIds);
      setResourceProperty(resource, TASK_ROLE_PROPERTY_ID, hostRoleCommand.getRole().toString(), requestedIds);
      setResourceProperty(resource, TASK_COMMAND_PROPERTY_ID, hostRoleCommand.getRoleCommand(), requestedIds);
      setResourceProperty(resource, TASK_STATUS_PROPERTY_ID, hostRoleCommand.getStatus(), requestedIds);
      setResourceProperty(resource, TASK_EXIT_CODE_PROPERTY_ID, hostRoleCommand.getExitCode(), requestedIds);
      setResourceProperty(resource, TASK_STDERR_PROPERTY_ID, hostRoleCommand.getStderr(), requestedIds);
      setResourceProperty(resource, TASK_STOUT_PROPERTY_ID, hostRoleCommand.getStdout(), requestedIds);
      setResourceProperty(resource, TASK_OUTPUTLOG_PROPERTY_ID, hostRoleCommand.getOutputLog(), requestedIds);
      setResourceProperty(resource, TASK_ERRORLOG_PROPERTY_ID, hostRoleCommand.getErrorLog(), requestedIds);
      setResourceProperty(resource, TASK_STRUCT_OUT_PROPERTY_ID, parseStructuredOutput(hostRoleCommand.getStructuredOut()), requestedIds);
      setResourceProperty(resource, TASK_START_TIME_PROPERTY_ID, hostRoleCommand.getStartTime(), requestedIds);
      setResourceProperty(resource, TASK_END_TIME_PROPERTY_ID, hostRoleCommand.getEndTime(), requestedIds);
      setResourceProperty(resource, TASK_ATTEMPT_CNT_PROPERTY_ID, hostRoleCommand.getAttemptCount(), requestedIds);

      if (hostRoleCommand.getCustomCommandName() != null) {
        setResourceProperty(resource, TASK_CUST_CMD_NAME_PROPERTY_ID, hostRoleCommand.getCustomCommandName(), requestedIds);
      }

      if (hostRoleCommand.getCommandDetail() == null) {
        setResourceProperty(resource, TASK_COMMAND_DET_PROPERTY_ID,
            String.format("%s %s", hostRoleCommand.getRole().toString(), hostRoleCommand.getRoleCommand()), requestedIds);
      } else {
        setResourceProperty(resource, TASK_COMMAND_DET_PROPERTY_ID, hostRoleCommand.getCommandDetail(), requestedIds);
      }

      setResourceProperty(resource, TASK_COMMAND_OPS_DISPLAY_NAME, hostRoleCommand.getOpsDisplayName(), requestedIds);
      results.add(resource);
    }

    return results;
  }

    TaskResourceProvider(AmbariManagementController managementController);

}