ActionDBAccessorImpl implements ActionDBAccessor { @Override @Transactional public void hostRoleScheduled(Stage s, String hostname, String roleStr) { HostRoleCommand hostRoleCommand = s.getHostRoleCommand(hostname, roleStr); HostRoleCommandEntity entity = hostRoleCommandDAO.findByPK(hostRoleCommand.getTaskId()); if (entity != null) { entity.setStartTime(hostRoleCommand.getStartTime()); if (entity.getOriginalStartTime() == null || entity.getOriginalStartTime() == -1) { entity.setOriginalStartTime(System.currentTimeMillis()); } entity.setLastAttemptTime(hostRoleCommand.getLastAttemptTime()); entity.setStatus(hostRoleCommand.getStatus()); entity.setAttemptCount(hostRoleCommand.getAttemptCount()); auditLog(entity, s.getRequestId()); hostRoleCommandDAO.merge(entity); } else { throw new RuntimeException("HostRoleCommand is not persisted, cannot update:\n" + hostRoleCommand); } } @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 testHostRoleScheduled() throws InterruptedException, AmbariException { populateActionDB(db, hostName, requestId, stageId); Stage stage = db.getStage(StageUtils.getActionId(requestId, stageId)); assertEquals(HostRoleStatus.PENDING, stage.getHostRoleStatus(hostName, Role.HBASE_MASTER.toString())); List<HostRoleCommandEntity> entities= hostRoleCommandDAO.findByHostRole(hostName, requestId, stageId, Role.HBASE_MASTER.toString()); assertEquals(HostRoleStatus.PENDING, entities.get(0).getStatus()); stage.setHostRoleStatus(hostName, Role.HBASE_MASTER.toString(), HostRoleStatus.QUEUED); entities = hostRoleCommandDAO.findByHostRole(hostName, requestId, stageId, Role.HBASE_MASTER.toString()); assertEquals(HostRoleStatus.QUEUED, stage.getHostRoleStatus(hostName, Role.HBASE_MASTER.toString())); assertEquals(HostRoleStatus.PENDING, entities.get(0).getStatus()); db.hostRoleScheduled(stage, hostName, Role.HBASE_MASTER.toString()); entities = hostRoleCommandDAO.findByHostRole(hostName, requestId, stageId, Role.HBASE_MASTER.toString()); assertEquals(HostRoleStatus.QUEUED, entities.get(0).getStatus()); Thread thread = new Thread(){ @Override public void run() { Stage stage1 = db.getStage("23-31"); stage1.setHostRoleStatus(hostName, Role.HBASE_MASTER.toString(), HostRoleStatus.COMPLETED); db.hostRoleScheduled(stage1, hostName, Role.HBASE_MASTER.toString()); injector.getInstance(EntityManager.class).clear(); } }; thread.start(); thread.join(); injector.getInstance(EntityManager.class).clear(); entities = hostRoleCommandDAO.findByHostRole(hostName, requestId, stageId, Role.HBASE_MASTER.toString()); assertEquals("Concurrent update failed", HostRoleStatus.COMPLETED, entities.get(0).getStatus()); }