ActionDBAccessorImpl implements ActionDBAccessor { @Override @Experimental(feature = ExperimentalFeature.PARALLEL_PROCESSING) public List<Stage> getStagesInProgress() { List<StageEntity> stageEntities = stageDAO.findByStatuses( HostRoleStatus.IN_PROGRESS_STATUSES); return getStagesForEntities(stageEntities); } @Inject  ActionDBAccessorImpl(@Named("executionCommandCacheSize") long cacheLimit,
                              AmbariEventPublisher eventPublisher); @Override Stage getStage(String actionId); @Override List<Stage> getAllStages(long requestId); @Override RequestEntity getRequestEntity(long requestId); @Override Request getRequest(long requestId); @Override Collection<HostRoleCommandEntity> abortOperation(long requestId); @Override void timeoutHostRole(String host, long requestId, long stageId,
                              String role); @Override void timeoutHostRole(String host, long requestId, long stageId, String role,
                              boolean skipSupported, boolean hostUnknownState); @Override @Experimental(feature = ExperimentalFeature.PARALLEL_PROCESSING) List<Stage> getStagesInProgressForRequest(Long requestId); @Override @Experimental(feature = ExperimentalFeature.PARALLEL_PROCESSING) List<Stage> getStagesInProgress(); @Override int getCommandsInProgressCount(); @Override @Transactional @TransactionalLock(lockArea = LockArea.HRC_STATUS_CACHE, lockType = LockType.WRITE) void persistActions(Request request); @Override @Transactional void startRequest(long requestId); @Override @Transactional void endRequest(long requestId); void endRequestIfCompleted(long requestId); @Override @Transactional void setSourceScheduleForRequest(long requestId, long scheduleId); @Override void updateHostRoleStates(Collection<CommandReport> reports); @Override void updateHostRoleState(String hostname, long requestId,
                                  long stageId, String role, CommandReport report); @Override void abortHostRole(String host, long requestId, long stageId, String role); @Override void abortHostRole(String host, long requestId, long stageId, String role, String reason); @Override long getLastPersistedRequestIdWhenInitialized(); @Override @Transactional void bulkHostRoleScheduled(Stage s, List<ExecutionCommand> commands); @Override @Transactional void bulkAbortHostRole(Stage s, Map<ExecutionCommand, String> commands); @Override @Transactional void hostRoleScheduled(Stage s, String hostname, String roleStr); @Override List<HostRoleCommand> getRequestTasks(long requestId); @Override List<HostRoleCommand> getAllTasksByRequestIds(Collection<Long> requestIds); @Override List<HostRoleCommand> getTasks(Collection<Long> taskIds); @Override List<HostRoleCommand> getTasksByHostRoleAndStatus(String hostname, String role, HostRoleStatus status); @Override List<HostRoleCommand> getTasksByRoleAndStatus(String role, HostRoleStatus status); @Override HostRoleCommand getTask(long taskId); @Override List<Long> getRequestsByStatus(RequestStatus status, int maxResults,
    boolean ascOrder); @Override Map<Long, String> getRequestContext(List<Long> requestIds); @Override String getRequestContext(long requestId); @Override List<Request> getRequests(Collection<Long> requestIds); @Override void resubmitTasks(List<Long> taskIds); @Subscribe void invalidateCommandCacheOnHostRemove(HostsRemovedEvent event);  }
@Test public void testGetStagesInProgress() throws AmbariException { List<Stage> stages = new ArrayList<Stage>(); stages.add(createStubStage(hostName, requestId, stageId)); stages.add(createStubStage(hostName, requestId, stageId + 1)); Request request = new Request(stages, clusters); db.persistActions(request); assertEquals(2, stages.size()); }