ActionScheduler implements Runnable { public void doWork() throws AmbariException { try { unitOfWork.begin(); threadEntityManager = entityManagerProvider.get(); processCancelledRequestsList(); if (db.getCommandsInProgressCount() == 0) { if (LOG.isDebugEnabled()) { LOG.debug("There are no stages currently in progress."); } actionQueue.updateListOfHostsWithPendingTask(null); return; } Set<Long> runningRequestIds = new HashSet<Long>(); List<Stage> stages = db.getStagesInProgress(); if (LOG.isDebugEnabled()) { LOG.debug("Scheduler wakes up"); LOG.debug("Processing {} in progress stages ", stages.size()); } publishInProgressTasks(stages); if (stages.isEmpty()) { if (LOG.isDebugEnabled()) { LOG.debug("There are no stages currently in progress."); } actionQueue.updateListOfHostsWithPendingTask(null); return; } int i_stage = 0; HashSet<String> hostsWithTasks = getListOfHostsWithPendingTask(stages); actionQueue.updateListOfHostsWithPendingTask(hostsWithTasks); stages = filterParallelPerHostStages(stages); boolean exclusiveRequestIsGoing = false; for (Stage stage : stages) { i_stage++; long requestId = stage.getRequestId(); LOG.debug("==> STAGE_i = " + i_stage + "(requestId=" + requestId + ",StageId=" + stage.getStageId() + ")"); RequestEntity request = db.getRequestEntity(requestId); if (request.isExclusive()) { if (runningRequestIds.size() > 0) { LOG.debug("Stage requires exclusive execution, but other requests are already executing. Stopping for now"); break; } exclusiveRequestIsGoing = true; } if (runningRequestIds.contains(requestId)) { LOG.debug("==> We don't want to process different stages from the same request in parallel"); continue; } else { runningRequestIds.add(requestId); if (!requestsInProgress.contains(requestId)) { requestsInProgress.add(requestId); db.startRequest(requestId); } } List<ExecutionCommand> commandsToSchedule = new ArrayList<>(); Multimap<String, AgentCommand> commandsToEnqueue = ArrayListMultimap.create(); Map<String, RoleStats> roleStats = processInProgressStage(stage, commandsToSchedule, commandsToEnqueue); boolean failed = false; for (Map.Entry<String, RoleStats> entry : roleStats.entrySet()) { String role = entry.getKey(); RoleStats stats = entry.getValue(); if (LOG.isDebugEnabled()) { LOG.debug("Stats for role: {}, stats={}", role, stats); } if (stats.isRoleFailed() && !stage.isSkippable()) { LOG.warn("{} failed, request {} will be aborted", role, request.getRequestId()); failed = true; break; } } if (!failed) { failed = hasPreviousStageFailed(stage); } if (failed) { LOG.error("Operation completely failed, aborting request id: {}", stage.getRequestId()); cancelHostRoleCommands(stage.getOrderedHostRoleCommands(), FAILED_TASK_ABORT_REASONING); abortOperationsForStage(stage); return; } List<ExecutionCommand> commandsToStart = new ArrayList<ExecutionCommand>(); List<ExecutionCommand> commandsToUpdate = new ArrayList<ExecutionCommand>(); for (ExecutionCommand cmd : commandsToSchedule) { ConfigHelper.processHiddenAttribute(cmd.getConfigurations(), cmd.getConfigurationAttributes(), cmd.getRole(), false); processHostRole(stage, cmd, commandsToStart, commandsToUpdate); } LOG.debug("==> Commands to start: {}", commandsToStart.size()); LOG.debug("==> Commands to update: {}", commandsToUpdate.size()); ListMultimap<String, ServiceComponentHostEvent> eventMap = formEventMap(stage, commandsToStart); Map<ExecutionCommand, String> commandsToAbort = new HashMap<ExecutionCommand, String>(); if (!eventMap.isEmpty()) { LOG.debug("==> processing {} serviceComponentHostEvents...", eventMap.size()); Cluster cluster = clusters.getCluster(stage.getClusterName()); if (cluster != null) { Map<ServiceComponentHostEvent, String> failedEvents = cluster.processServiceComponentHostEvents(eventMap); if (failedEvents.size() > 0) { LOG.error("==> {} events failed.", failedEvents.size()); } for (Iterator<ExecutionCommand> iterator = commandsToUpdate.iterator(); iterator.hasNext(); ) { ExecutionCommand cmd = iterator.next(); for (ServiceComponentHostEvent event : failedEvents.keySet()) { if (StringUtils.equals(event.getHostName(), cmd.getHostname()) && StringUtils.equals(event.getServiceComponentName(), cmd.getRole())) { iterator.remove(); commandsToAbort.put(cmd, failedEvents.get(event)); break; } } } } else { LOG.warn("There was events to process but cluster {} not found", stage.getClusterName()); } } LOG.debug("==> Scheduling {} tasks...", commandsToUpdate.size()); db.bulkHostRoleScheduled(stage, commandsToUpdate); if (commandsToAbort.size() > 0) { LOG.debug("==> Aborting {} tasks...", commandsToAbort.size()); List<Long> taskIds = new ArrayList<Long>(); for (ExecutionCommand command : commandsToAbort.keySet()) { taskIds.add(command.getTaskId()); } Collection<HostRoleCommand> hostRoleCommands = db.getTasks(taskIds); cancelHostRoleCommands(hostRoleCommands, FAILED_TASK_ABORT_REASONING); db.bulkAbortHostRole(stage, commandsToAbort); } LOG.debug("==> Adding {} tasks to queue...", commandsToUpdate.size()); for (ExecutionCommand cmd : commandsToUpdate) { if (Role.AMBARI_SERVER_ACTION.name().equals(cmd.getRole())) { serverActionExecutor.awake(); } else { commandsToEnqueue.put(cmd.getHostname(), cmd); } } actionQueue.enqueueAll(commandsToEnqueue.asMap()); LOG.debug("==> Finished."); if (!configuration.getParallelStageExecution()) { return; } if (exclusiveRequestIsGoing) { LOG.debug("Stage requires exclusive execution, skipping all executing any further stages"); break; } } requestsInProgress.retainAll(runningRequestIds); } finally { LOG.debug("Scheduler finished work."); unitOfWork.end(); } } @Inject  ActionScheduler(@Named("schedulerSleeptime") long sleepTime,
                         @Named("actionTimeout") long actionTimeout, ActionDBAccessor db,
                         JPAEventPublisher jpaPublisher); protected  ActionScheduler(long sleepTimeMilliSec, long actionTimeoutMilliSec, ActionDBAccessor db,
                            ActionQueue actionQueue, Clusters fsmObject, int maxAttempts, HostsMap hostsMap,
                            UnitOfWork unitOfWork, AmbariEventPublisher ambariEventPublisher,
                            Configuration configuration, Provider<EntityManager> entityManagerProvider,
                            HostRoleCommandDAO hostRoleCommandDAO, HostRoleCommandFactory hostRoleCommandFactory); void start(); void stop(); void awake(); @Override void run(); void doWork(); void scheduleCancellingRequest(long requestId, String reason); void setTaskTimeoutAdjustment(boolean val); @Subscribe void onEvent(EntityManagerCacheInvalidationEvent event); static final String FAILED_TASK_ABORT_REASONING; }
@Test public void testServerActionInMultipleRequests() throws Exception { ActionQueue aq = new ActionQueue(); Clusters fsm = mock(Clusters.class); Cluster oneClusterMock = mock(Cluster.class); Service serviceObj = mock(Service.class); ServiceComponent scomp = mock(ServiceComponent.class); ServiceComponentHost sch = mock(ServiceComponentHost.class); UnitOfWork unitOfWork = mock(UnitOfWork.class); when(fsm.getCluster(anyString())).thenReturn(oneClusterMock); when(oneClusterMock.getService(anyString())).thenReturn(serviceObj); when(serviceObj.getServiceComponent(anyString())).thenReturn(scomp); when(scomp.getServiceComponentHost(anyString())).thenReturn(sch); when(serviceObj.getCluster()).thenReturn(oneClusterMock); String clusterName = "cluster1"; String hostname1 = "ahost.ambari.apache.org"; String hostname2 = "bhost.ambari.apache.org"; HashMap<String, ServiceComponentHost> hosts = new HashMap<String, ServiceComponentHost>(); hosts.put(hostname1, sch); hosts.put(hostname2, sch); hosts.put(Stage.INTERNAL_HOSTNAME, sch); when(scomp.getServiceComponentHosts()).thenReturn(hosts); List<Stage> stages = new ArrayList<Stage>(); Stage stage01 = createStage(clusterName, 0, 1); addTask(stage01, Stage.INTERNAL_HOSTNAME, clusterName, Role.AMBARI_SERVER_ACTION, RoleCommand.ACTIONEXECUTE, "AMBARI", 1); Stage stage11 = createStage("cluster1", 1, 1); addTask(stage11, hostname1, clusterName, Role.KERBEROS_CLIENT, RoleCommand.CUSTOM_COMMAND, "KERBEROS", 2); Stage stage02 = createStage("cluster1", 0, 2); addTask(stage02, Stage.INTERNAL_HOSTNAME, clusterName, Role.AMBARI_SERVER_ACTION, RoleCommand.ACTIONEXECUTE, "AMBARI", 3); Stage stage12 = createStage("cluster1", 1, 2); addTask(stage12, hostname2, clusterName, Role.KERBEROS_CLIENT, RoleCommand.CUSTOM_COMMAND, "KERBEROS", 4); stages.add(stage01); stages.add(stage11); stages.add(stage02); stages.add(stage12); ActionDBAccessor db = mock(ActionDBAccessor.class); HostRoleCommandDAO hostRoleCommandDAOMock = mock(HostRoleCommandDAO.class); Mockito.doNothing().when(hostRoleCommandDAOMock).publishTaskCreateEvent(anyListOf(HostRoleCommand.class)); RequestEntity request = mock(RequestEntity.class); when(request.isExclusive()).thenReturn(false); when(db.getRequestEntity(anyLong())).thenReturn(request); when(db.getCommandsInProgressCount()).thenReturn(stages.size()); when(db.getStagesInProgress()).thenReturn(stages); Properties properties = new Properties(); properties.put(Configuration.PARALLEL_STAGE_EXECUTION.getKey(), "true"); Configuration conf = new Configuration(properties); ActionScheduler scheduler = new ActionScheduler(100, 50, db, aq, fsm, 3, new HostsMap((String) null), unitOfWork, EasyMock.createNiceMock(AmbariEventPublisher.class), conf, entityManagerProviderMock, hostRoleCommandDAOMock, (HostRoleCommandFactory)null); scheduler.doWork(); Assert.assertEquals(HostRoleStatus.QUEUED, stages.get(0).getHostRoleStatus(Stage.INTERNAL_HOSTNAME, Role.AMBARI_SERVER_ACTION.name())); Assert.assertEquals(HostRoleStatus.PENDING, stages.get(1).getHostRoleStatus(hostname1, Role.KERBEROS_CLIENT.name())); Assert.assertEquals(HostRoleStatus.QUEUED, stages.get(2).getHostRoleStatus(Stage.INTERNAL_HOSTNAME, Role.AMBARI_SERVER_ACTION.name())); Assert.assertEquals(HostRoleStatus.PENDING, stages.get(3).getHostRoleStatus(hostname2, Role.KERBEROS_CLIENT.name())); }
@Test public void testRequestAbortsOnlyWhenNoQueuedTaskAndSuccessFactorUnmet() throws Exception { ActionQueue aq = new ActionQueue(); Clusters fsm = mock(Clusters.class); Cluster oneClusterMock = mock(Cluster.class); Service serviceObj = mock(Service.class); ServiceComponent scomp = mock(ServiceComponent.class); ServiceComponentHost sch = mock(ServiceComponentHost.class); UnitOfWork unitOfWork = mock(UnitOfWork.class); when(fsm.getCluster(anyString())).thenReturn(oneClusterMock); when(oneClusterMock.getService(anyString())).thenReturn(serviceObj); when(serviceObj.getServiceComponent(anyString())).thenReturn(scomp); when(scomp.getServiceComponentHost(anyString())).thenReturn(sch); when(serviceObj.getCluster()).thenReturn(oneClusterMock); String host1 = "host1"; String host2 = "host2"; Host host = mock(Host.class); HashMap<String, ServiceComponentHost> hosts = new HashMap<String, ServiceComponentHost>(); hosts.put(host1, sch); hosts.put(host2, sch); when(scomp.getServiceComponentHosts()).thenReturn(hosts); when(fsm.getHost(anyString())).thenReturn(host); when(host.getState()).thenReturn(HostState.HEALTHY); when(host.getHostName()).thenReturn(host1); HostEntity hostEntity1 = new HostEntity(); HostEntity hostEntity2 = new HostEntity(); hostEntity1.setHostName(host1); hostEntity2.setHostName(host2); hostDAO.create(hostEntity1); hostDAO.create(hostEntity2); final List<Stage> stages = new ArrayList<Stage>(); long now = System.currentTimeMillis(); Stage stage = stageFactory.createNew(1, "/tmp", "cluster1", 1L, "testRequestFailureBasedOnSuccessFactor", CLUSTER_HOST_INFO, "", ""); stage.setStageId(1); addHostRoleExecutionCommand(now, stage, Role.SQOOP, Service.Type.SQOOP, RoleCommand.INSTALL, host1, "cluster1"); addHostRoleExecutionCommand(now, stage, Role.OOZIE_CLIENT, Service.Type.OOZIE, RoleCommand.INSTALL, host1, "cluster1"); addHostRoleExecutionCommand(now, stage, Role.MAPREDUCE_CLIENT, Service.Type.MAPREDUCE, RoleCommand.INSTALL, host1, "cluster1"); addHostRoleExecutionCommand(now, stage, Role.HBASE_CLIENT, Service.Type.HBASE, RoleCommand.INSTALL, host1, "cluster1"); addHostRoleExecutionCommand(now, stage, Role.GANGLIA_MONITOR, Service.Type.GANGLIA, RoleCommand.INSTALL, host1, "cluster1"); addHostRoleExecutionCommand(now, stage, Role.HBASE_CLIENT, Service.Type.HBASE, RoleCommand.INSTALL, host2, "cluster1"); addHostRoleExecutionCommand(now, stage, Role.GANGLIA_MONITOR, Service.Type.GANGLIA, RoleCommand.INSTALL, host2, "cluster1"); stages.add(stage); HostRoleStatus[] statusesAtIterOne = {HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.FAILED, HostRoleStatus.FAILED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED}; for (int index = 0; index < stage.getOrderedHostRoleCommands().size(); index++) { stage.getOrderedHostRoleCommands().get(index).setTaskId(index + 1); stage.getOrderedHostRoleCommands().get(index).setStatus(statusesAtIterOne[index]); } stage.setLastAttemptTime(host1, Role.SQOOP.toString(), now); stage.setLastAttemptTime(host1, Role.MAPREDUCE_CLIENT.toString(), now); stage.setLastAttemptTime(host1, Role.OOZIE_CLIENT.toString(), now); stage.setLastAttemptTime(host1, Role.GANGLIA_MONITOR.toString(), now); stage.setLastAttemptTime(host1, Role.HBASE_CLIENT.toString(), now); stage.setLastAttemptTime(host2, Role.GANGLIA_MONITOR.toString(), now); stage.setLastAttemptTime(host2, Role.HBASE_CLIENT.toString(), now); ActionDBAccessor db = mock(ActionDBAccessor.class); HostRoleCommandDAO hostRoleCommandDAOMock = mock(HostRoleCommandDAO.class); Mockito.doNothing().when(hostRoleCommandDAOMock).publishTaskCreateEvent(anyListOf(HostRoleCommand.class)); RequestEntity request = mock(RequestEntity.class); when(request.isExclusive()).thenReturn(false); when(db.getRequestEntity(anyLong())).thenReturn(request); when(db.getCommandsInProgressCount()).thenReturn(stages.size()); when(db.getStagesInProgress()).thenReturn(stages); doAnswer(new Answer<Void>() { @Override public Void answer(InvocationOnMock invocation) throws Throwable { String host = (String) invocation.getArguments()[0]; Long requestId = (Long) invocation.getArguments()[1]; Long stageId = (Long) invocation.getArguments()[2]; String role = (String) invocation.getArguments()[3]; CommandReport commandReport = (CommandReport) invocation.getArguments()[4]; for (Stage stage : stages) { if (requestId.equals(stage.getRequestId()) && stageId.equals(stage.getStageId())) { HostRoleCommand command = stage.getHostRoleCommand(host, role); command.setStatus(HostRoleStatus.valueOf(commandReport.getStatus())); } } return null; } }).when(db).updateHostRoleState(anyString(), anyLong(), anyLong(), anyString(), any(CommandReport.class)); when(db.getTask(anyLong())).thenAnswer(new Answer<Object>() { @Override public Object answer(InvocationOnMock invocation) throws Throwable { Long taskId = (Long) invocation.getArguments()[0]; for (Stage stage : stages) { for (HostRoleCommand command : stage.getOrderedHostRoleCommands()) { if (taskId.equals(command.getTaskId())) { return command; } } } return null; } }); doAnswer(new Answer<Collection<HostRoleCommandEntity>>() { @Override public Collection<HostRoleCommandEntity> answer(InvocationOnMock invocation) throws Throwable { Long requestId = (Long) invocation.getArguments()[0]; List<HostRoleCommandEntity> abortedCommands = Lists.newArrayList(); for (Stage stage : stages) { if (requestId.equals(stage.getRequestId())) { for (HostRoleCommand command : stage.getOrderedHostRoleCommands()) { if (command.getStatus() == HostRoleStatus.QUEUED || command.getStatus() == HostRoleStatus.IN_PROGRESS || command.getStatus() == HostRoleStatus.PENDING) { command.setStatus(HostRoleStatus.ABORTED); HostRoleCommandEntity hostRoleCommandEntity = command.constructNewPersistenceEntity(); hostRoleCommandEntity.setStage(stage.constructNewPersistenceEntity()); abortedCommands.add(hostRoleCommandEntity); } } } } return abortedCommands; } }).when(db).abortOperation(anyLong()); Properties properties = new Properties(); Configuration conf = new Configuration(properties); ActionScheduler scheduler = new ActionScheduler(100, 10000, db, aq, fsm, 3, new HostsMap((String) null), unitOfWork, null, conf, entityManagerProviderMock, hostRoleCommandDAOMock, (HostRoleCommandFactory)null); scheduler.doWork(); HostRoleStatus[] expectedStatusesAtIterOne = {HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.FAILED, HostRoleStatus.FAILED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED}; for (int index = 0; index < stage.getOrderedHostRoleCommands().size(); index++) { log.info(stage.getOrderedHostRoleCommands().get(index).toString()); Assert.assertEquals(expectedStatusesAtIterOne[index], stage.getOrderedHostRoleCommands().get(index).getStatus()); } HostRoleStatus[] statusesAtIterTwo = {HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.FAILED, HostRoleStatus.FAILED, HostRoleStatus.QUEUED, HostRoleStatus.COMPLETED}; for (int index = 0; index < stage.getOrderedHostRoleCommands().size(); index++) { stage.getOrderedHostRoleCommands().get(index).setStatus(statusesAtIterTwo[index]); } scheduler.doWork(); HostRoleStatus[] expectedStatusesAtIterTwo = {HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.FAILED, HostRoleStatus.FAILED, HostRoleStatus.QUEUED, HostRoleStatus.COMPLETED}; for (int index = 0; index < stage.getOrderedHostRoleCommands().size(); index++) { log.info(stage.getOrderedHostRoleCommands().get(index).toString()); Assert.assertEquals(expectedStatusesAtIterTwo[index], stage.getOrderedHostRoleCommands().get(index).getStatus()); } HostRoleStatus[] statusesAtIterThree = {HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.FAILED, HostRoleStatus.FAILED, HostRoleStatus.FAILED, HostRoleStatus.COMPLETED}; for (int index = 0; index < stage.getOrderedHostRoleCommands().size(); index++) { stage.getOrderedHostRoleCommands().get(index).setStatus(statusesAtIterThree[index]); } scheduler.doWork(); HostRoleStatus[] expectedStatusesAtIterThree = {HostRoleStatus.ABORTED, HostRoleStatus.ABORTED, HostRoleStatus.ABORTED, HostRoleStatus.ABORTED, HostRoleStatus.FAILED, HostRoleStatus.FAILED, HostRoleStatus.FAILED, HostRoleStatus.COMPLETED}; for (int index = 0; index < stage.getOrderedHostRoleCommands().size(); index++) { log.info(stage.getOrderedHostRoleCommands().get(index).toString()); Assert.assertEquals(expectedStatusesAtIterThree[index], stage.getOrderedHostRoleCommands().get(index).getStatus()); } }
@Test public void testRequestFailureBasedOnSuccessFactor() throws Exception { ActionQueue aq = new ActionQueue(); Clusters fsm = mock(Clusters.class); Cluster oneClusterMock = mock(Cluster.class); Service serviceObj = mock(Service.class); ServiceComponent scomp = mock(ServiceComponent.class); ServiceComponentHost sch = mock(ServiceComponentHost.class); UnitOfWork unitOfWork = mock(UnitOfWork.class); RequestFactory requestFactory = mock(RequestFactory.class); when(fsm.getCluster(anyString())).thenReturn(oneClusterMock); when(oneClusterMock.getService(anyString())).thenReturn(serviceObj); when(serviceObj.getServiceComponent(anyString())).thenReturn(scomp); when(scomp.getServiceComponentHost(anyString())).thenReturn(sch); when(serviceObj.getCluster()).thenReturn(oneClusterMock); final List<Stage> stages = new ArrayList<Stage>(); long now = System.currentTimeMillis(); Stage stage = stageFactory.createNew(1, "/tmp", "cluster1", 1L, "testRequestFailureBasedOnSuccessFactor", CLUSTER_HOST_INFO, "", ""); stage.setStageId(1); stage.addHostRoleExecutionCommand("host1", Role.DATANODE, RoleCommand.UPGRADE, new ServiceComponentHostUpgradeEvent(Role.DATANODE.toString(), "host1", now, "HDP-0.2"), "cluster1", Service.Type.HDFS.toString(), false, false); stage.getExecutionCommandWrapper("host1", Role.DATANODE.toString()).getExecutionCommand(); stage.addHostRoleExecutionCommand("host2", Role.DATANODE, RoleCommand.UPGRADE, new ServiceComponentHostUpgradeEvent(Role.DATANODE.toString(), "host2", now, "HDP-0.2"), "cluster1", Service.Type.HDFS.toString(), false, false); stage.getExecutionCommandWrapper("host2", Role.DATANODE.toString()).getExecutionCommand(); stage.addHostRoleExecutionCommand("host3", Role.DATANODE, RoleCommand.UPGRADE, new ServiceComponentHostUpgradeEvent(Role.DATANODE.toString(), "host3", now, "HDP-0.2"), "cluster1", Service.Type.HDFS.toString(), false, false); stage.getExecutionCommandWrapper("host3", Role.DATANODE.toString()).getExecutionCommand(); stages.add(stage); stage.getOrderedHostRoleCommands().get(0).setTaskId(1); stage.getOrderedHostRoleCommands().get(1).setTaskId(2); stage.getOrderedHostRoleCommands().get(2).setTaskId(3); stages.add( getStageWithSingleTask( "host1", "cluster1", Role.HDFS_CLIENT, RoleCommand.UPGRADE, Service.Type.HDFS, 4, 2, 1)); ActionDBAccessor db = mock(ActionDBAccessor.class); HostRoleCommandDAO hostRoleCommandDAOMock = mock(HostRoleCommandDAO.class); Mockito.doNothing().when(hostRoleCommandDAOMock).publishTaskCreateEvent(anyListOf(HostRoleCommand.class)); RequestEntity request = mock(RequestEntity.class); when(request.isExclusive()).thenReturn(false); when(db.getRequestEntity(anyLong())).thenReturn(request); when(db.getCommandsInProgressCount()).thenReturn(stages.size()); when(db.getStagesInProgress()).thenReturn(stages); doAnswer(new Answer<Void>() { @Override public Void answer(InvocationOnMock invocation) throws Throwable { List<CommandReport> reports = (List<CommandReport>) invocation.getArguments()[0]; for (CommandReport report : reports) { String actionId = report.getActionId(); long[] requestStageIds = StageUtils.getRequestStage(actionId); Long requestId = requestStageIds[0]; Long stageId = requestStageIds[1]; Long id = report.getTaskId(); for (Stage stage : stages) { if (requestId.equals(stage.getRequestId()) && stageId.equals(stage.getStageId())) { for (HostRoleCommand hostRoleCommand : stage.getOrderedHostRoleCommands()) { if (hostRoleCommand.getTaskId() == id) { hostRoleCommand.setStatus(HostRoleStatus.valueOf(report.getStatus())); } } } } } return null; } }).when(db).updateHostRoleStates(anyCollectionOf(CommandReport.class)); when(db.getTask(anyLong())).thenAnswer(new Answer<Object>() { @Override public Object answer(InvocationOnMock invocation) throws Throwable { Long taskId = (Long) invocation.getArguments()[0]; for (Stage stage : stages) { for (HostRoleCommand command : stage.getOrderedHostRoleCommands()) { if (taskId.equals(command.getTaskId())) { return command; } } } return null; } }); doAnswer(new Answer<Collection<HostRoleCommandEntity>>() { @Override public Collection<HostRoleCommandEntity> answer(InvocationOnMock invocation) throws Throwable { Long requestId = (Long) invocation.getArguments()[0]; List<HostRoleCommandEntity> abortedCommands = Lists.newArrayList(); for (Stage stage : stages) { if (requestId.equals(stage.getRequestId())) { for (HostRoleCommand command : stage.getOrderedHostRoleCommands()) { if (command.getStatus() == HostRoleStatus.QUEUED || command.getStatus() == HostRoleStatus.IN_PROGRESS || command.getStatus() == HostRoleStatus.PENDING) { command.setStatus(HostRoleStatus.ABORTED); HostRoleCommandEntity hostRoleCommandEntity = command.constructNewPersistenceEntity(); hostRoleCommandEntity.setStage(stage.constructNewPersistenceEntity()); abortedCommands.add(hostRoleCommandEntity); } } } } return abortedCommands; } }).when(db).abortOperation(anyLong()); Properties properties = new Properties(); Configuration conf = new Configuration(properties); ActionScheduler scheduler = new ActionScheduler(100, 50, db, aq, fsm, 3, new HostsMap((String) null), unitOfWork, null, conf, entityManagerProviderMock, hostRoleCommandDAOMock, (HostRoleCommandFactory)null); ActionManager am = new ActionManager(db, requestFactory, scheduler); scheduler.doWork(); List<CommandReport> reports = new ArrayList<CommandReport>(); reports.add(getCommandReport(HostRoleStatus.FAILED, Role.DATANODE, Service.Type.HDFS, "1-1", 1)); am.processTaskResponse("host1", reports, CommandUtils.convertToTaskIdCommandMap(stage.getOrderedHostRoleCommands())); reports.clear(); reports.add(getCommandReport(HostRoleStatus.FAILED, Role.DATANODE, Service.Type.HDFS, "1-1", 2)); am.processTaskResponse("host2", reports, CommandUtils.convertToTaskIdCommandMap(stage.getOrderedHostRoleCommands())); reports.clear(); reports.add(getCommandReport(HostRoleStatus.COMPLETED, Role.DATANODE, Service.Type.HDFS, "1-1", 3)); am.processTaskResponse("host3", reports, CommandUtils.convertToTaskIdCommandMap(stage.getOrderedHostRoleCommands())); scheduler.doWork(); Assert.assertEquals(HostRoleStatus.ABORTED, stages.get(1).getHostRoleStatus("host1", "HDFS_CLIENT")); }