ActionDBAccessorImpl implements ActionDBAccessor { @Override @Transactional @TransactionalLock(lockArea = LockArea.HRC_STATUS_CACHE, lockType = LockType.WRITE) public void persistActions(Request request) throws AmbariException { RequestEntity requestEntity = request.constructNewPersistenceEntity(); Long clusterId = -1L; Long requestId = requestEntity.getRequestId(); ClusterEntity clusterEntity = clusterDAO.findById(request.getClusterId()); if (clusterEntity != null) { clusterId = clusterEntity.getClusterId(); } requestEntity.setClusterId(clusterId); requestDAO.create(requestEntity); List<StageEntity> stageEntities = new ArrayList<StageEntity>(request.getStages().size()); addRequestToAuditlogCache(request); List<HostRoleCommand> hostRoleCommands = new ArrayList<>(); for (Stage stage : request.getStages()) { StageEntity stageEntity = stage.constructNewPersistenceEntity(); Long stageId = stageEntity.getStageId(); stageEntities.add(stageEntity); stageEntity.setClusterId(clusterId); stageEntity.setRequest(requestEntity); stageDAO.create(stageEntity); List<HostRoleCommand> orderedHostRoleCommands = stage.getOrderedHostRoleCommands(); for (HostRoleCommand hostRoleCommand : orderedHostRoleCommands) { hostRoleCommand.setRequestId(requestId); hostRoleCommand.setStageId(stageId); HostRoleCommandEntity hostRoleCommandEntity = hostRoleCommand.constructNewPersistenceEntity(); hostRoleCommandEntity.setStage(stageEntity); hostRoleCommandDAO.create(hostRoleCommandEntity); assert hostRoleCommandEntity.getTaskId() != null; hostRoleCommand.setTaskId(hostRoleCommandEntity.getTaskId()); String prefix = ""; String output = "output-" + hostRoleCommandEntity.getTaskId() + ".txt"; String error = "errors-" + hostRoleCommandEntity.getTaskId() + ".txt"; HostEntity hostEntity = null; if (null != hostRoleCommandEntity.getHostId()) { hostEntity = hostDAO.findById(hostRoleCommandEntity.getHostId()); if (hostEntity == null) { String msg = String.format("Host %s doesn't exist in database", hostRoleCommandEntity.getHostName()); LOG.error(msg); throw new AmbariException(msg); } hostRoleCommandEntity.setHostEntity(hostEntity); try { Host hostObject = clusters.getHost(hostEntity.getHostName()); if (!StringUtils.isBlank(hostObject.getPrefix())) { prefix = hostObject.getPrefix(); if (!prefix.endsWith("/")) { prefix = prefix + "/"; } } } catch (AmbariException e) { LOG.warn("Exception in getting prefix for host and setting output and error log files. Using no prefix"); } } hostRoleCommand.setOutputLog(prefix + output); hostRoleCommand.setErrorLog(prefix + error); hostRoleCommandEntity.setOutputLog(hostRoleCommand.getOutputLog()); hostRoleCommandEntity.setErrorLog(hostRoleCommand.getErrorLog()); ExecutionCommandEntity executionCommandEntity = hostRoleCommand.constructExecutionCommandEntity(); executionCommandEntity.setHostRoleCommand(hostRoleCommandEntity); executionCommandEntity.setTaskId(hostRoleCommandEntity.getTaskId()); hostRoleCommandEntity.setExecutionCommand(executionCommandEntity); executionCommandDAO.create(hostRoleCommandEntity.getExecutionCommand()); hostRoleCommandEntity = hostRoleCommandDAO.mergeWithoutPublishEvent(hostRoleCommandEntity); if (null != hostEntity) { hostEntity = hostDAO.merge(hostEntity); } hostRoleCommands.add(hostRoleCommand); } for (RoleSuccessCriteriaEntity roleSuccessCriteriaEntity : stageEntity.getRoleSuccessCriterias()) { roleSuccessCriteriaDAO.create(roleSuccessCriteriaEntity); } stageEntity = stageDAO.merge(stageEntity); } requestEntity.setStages(stageEntities); requestDAO.merge(requestEntity); TaskCreateEvent taskCreateEvent = new TaskCreateEvent(hostRoleCommands); taskEventPublisher.publish(taskCreateEvent); } @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 testPersistActions() throws AmbariException { populateActionDB(db, hostName, requestId, stageId); for (Stage stage : db.getAllStages(requestId)) { log.info("taskId={}" + stage.getExecutionCommands(hostName).get(0). getExecutionCommand().getTaskId()); assertTrue(stage.getExecutionCommands(hostName).get(0). getExecutionCommand().getTaskId() > 0); assertTrue(executionCommandDAO.findByPK(stage.getExecutionCommands(hostName). get(0).getExecutionCommand().getTaskId()) != null); } }
@Test public void testEntitiesCreatedWithIDs() throws Exception { List<Stage> stages = new ArrayList<Stage>(); Stage stage = createStubStage(hostName, requestId, stageId); stages.add(stage); Request request = new Request(stages, clusters); db.persistActions(request); List<HostRoleCommandEntity> commandEntities = hostRoleCommandDAO.findByRequest(requestId); Assert.assertEquals(2, commandEntities.size()); for (HostRoleCommandEntity entity : commandEntities) { Assert.assertEquals(Long.valueOf(requestId), entity.getRequestId()); Assert.assertEquals(Long.valueOf(stageId), entity.getStageId()); } }
@Test public void testGet1000TasksFromOracleDB() throws Exception { Stage s = stageFactory.createNew(requestId, "/a/b", "cluster1", 1L, "action db accessor test", "clusterHostInfo", "commandParamsStage", "hostParamsStage"); s.setStageId(stageId); for (int i = 1000; i < 2002; i++) { String host = "host" + i; clusters.addHost(host); s.addHostRoleExecutionCommand("host" + i, Role.HBASE_MASTER, RoleCommand.START, null, "cluster1", "HBASE", false, false); } List<Stage> stages = new ArrayList<Stage>(); stages.add(s); Request request = new Request(stages, clusters); db.persistActions(request); List<HostRoleCommandEntity> entities = hostRoleCommandDAO.findByRequest(request.getRequestId()); assertEquals(1002, entities.size()); List<Long> taskIds = new ArrayList<Long>(); for (HostRoleCommandEntity entity : entities) { taskIds.add(entity.getTaskId()); } TestDBAccessorImpl testDBAccessorImpl = (TestDBAccessorImpl) injector.getInstance(DBAccessor.class); testDBAccessorImpl.setDbTypeOverride(ORACLE); assertEquals(ORACLE, injector.getInstance(DBAccessor.class).getDbType()); entities = hostRoleCommandDAO.findByPKs(taskIds); assertEquals("Tasks returned from DB match the ones created", taskIds.size(), entities.size()); }