class BatchRequestJob extends AbstractLinearExecutionJob {

  @Override
  protected void doWork(Map<String, Object> properties) throws AmbariException {

    Long executionId = properties.get(BATCH_REQUEST_EXECUTION_ID_KEY) != null ?
      (Long) properties.get(BATCH_REQUEST_EXECUTION_ID_KEY) : null;
    Long batchId = properties.get(BATCH_REQUEST_BATCH_ID_KEY) != null ?
      (Long) properties.get(BATCH_REQUEST_BATCH_ID_KEY) : null;
    String clusterName = (String) properties.get(BATCH_REQUEST_CLUSTER_NAME_KEY);


    if (executionId == null || batchId == null) {
      throw new AmbariException("Unable to retrieve persisted batch request"
        + ", execution_id = " + executionId
        + ", batch_id = " + batchId);
    }

    // Aggregate tasks counts stored in the DataMap
    Map<String, Integer> taskCounts = getTaskCountProperties(properties);

    Long requestId = executionScheduleManager.executeBatchRequest
      (executionId, batchId, clusterName);

    if (requestId != null) {
      HostRoleStatus status;
      BatchRequestResponse batchRequestResponse;
      do {
        batchRequestResponse = executionScheduleManager
          .getBatchRequestResponse(requestId, clusterName);

        status = HostRoleStatus.valueOf(batchRequestResponse.getStatus());

        executionScheduleManager.updateBatchRequest(executionId, batchId,
          clusterName, batchRequestResponse, true);

        try {
          Thread.sleep(statusCheckInterval);
        } catch (InterruptedException e) {
          String message = "Job Thread interrupted";
          LOG.error(message, e);
          throw new AmbariException(message, e);
        }
      } while (!status.isCompletedState());

      // Store aggregated task status counts in the DataMap
      Map<String, Integer> aggregateCounts = addTaskCountToProperties
        (properties, taskCounts, batchRequestResponse);

      if (executionScheduleManager.hasToleranceThresholdExceeded
          (executionId, clusterName, aggregateCounts)) {

        throw new AmbariException("Task failure tolerance limit exceeded"
            + ", execution_id = " + executionId
            + ", processed batch_id = " + batchId
            + ", failed tasks = " + aggregateCounts.get(BATCH_REQUEST_FAILED_TASKS_KEY)
            + ", total tasks completed = " + aggregateCounts.get(BATCH_REQUEST_TOTAL_TASKS_KEY));
      }
    }
  }

  @Inject public  BatchRequestJob(ExecutionScheduleManager executionScheduleManager,
                         @Named("statusCheckInterval") long statusCheckInterval);

}