ActionScheduler implements Runnable { void cancelHostRoleCommands(Collection<HostRoleCommand> hostRoleCommands, String reason) { for (HostRoleCommand hostRoleCommand : hostRoleCommands) { if (!Role.AMBARI_SERVER_ACTION.equals(hostRoleCommand.getRole())) { if (hostRoleCommand.getStatus() == HostRoleStatus.QUEUED) { actionQueue.dequeue(hostRoleCommand.getHostName(), hostRoleCommand.getExecutionCommandWrapper(). getExecutionCommand().getCommandId()); } if (hostRoleCommand.getStatus() == HostRoleStatus.QUEUED || hostRoleCommand.getStatus() == HostRoleStatus.IN_PROGRESS) { CancelCommand cancelCommand = new CancelCommand(); cancelCommand.setTargetTaskId(hostRoleCommand.getTaskId()); cancelCommand.setReason(reason); actionQueue.enqueue(hostRoleCommand.getHostName(), cancelCommand); } } if (hostRoleCommand.getStatus().isHoldingState()) { db.abortHostRole(hostRoleCommand.getHostName(), hostRoleCommand.getRequestId(), hostRoleCommand.getStageId(), hostRoleCommand.getRole().name()); } if (hostRoleCommand.getRoleCommand().equals(RoleCommand.ACTIONEXECUTE)) { String clusterName = hostRoleCommand.getExecutionCommandWrapper().getExecutionCommand().getClusterName(); processActionDeath(clusterName, hostRoleCommand.getHostName(), hostRoleCommand.getRole().name()); } } } @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 testAbortHolding() { UnitOfWork unitOfWork = EasyMock.createMock(UnitOfWork.class); ActionDBAccessor db = EasyMock.createMock(ActionDBAccessor.class); ActionQueue aq = new ActionQueue(); Clusters fsm = EasyMock.createMock(Clusters.class); Configuration conf = new Configuration(new Properties()); HostEntity hostEntity1 = new HostEntity(); hostEntity1.setHostName("h1"); hostDAO.merge(hostEntity1); db.abortHostRole("h1", -1L, -1L, "AMBARI_SERVER_ACTION"); EasyMock.expectLastCall(); EasyMock.replay(db); ActionScheduler scheduler = new ActionScheduler(100, 50, db, aq, fsm, 3, new HostsMap((String) null), unitOfWork, null, conf, entityManagerProviderMock, (HostRoleCommandDAO)null, (HostRoleCommandFactory)null); HostRoleCommand hrc1 = hostRoleCommandFactory.create("h1", Role.NAMENODE, null, RoleCommand.EXECUTE); hrc1.setStatus(HostRoleStatus.COMPLETED); HostRoleCommand hrc3 = hostRoleCommandFactory.create("h1", Role.AMBARI_SERVER_ACTION, null, RoleCommand.CUSTOM_COMMAND); hrc3.setStatus(HostRoleStatus.HOLDING); HostRoleCommand hrc4 = hostRoleCommandFactory.create("h1", Role.FLUME_HANDLER, null, RoleCommand.EXECUTE); hrc4.setStatus(HostRoleStatus.PENDING); List<HostRoleCommand> hostRoleCommands = Arrays.asList(hrc1, hrc3, hrc4); scheduler.cancelHostRoleCommands(hostRoleCommands, "foo"); EasyMock.verify(db); }
@Test public void testAbortAmbariServerAction() { UnitOfWork unitOfWork = EasyMock.createMock(UnitOfWork.class); ActionDBAccessor db = EasyMock.createMock(ActionDBAccessor.class); ActionQueue aq = new ActionQueue(); Clusters fsm = EasyMock.createMock(Clusters.class); Configuration conf = new Configuration(new Properties()); HostEntity hostEntity1 = new HostEntity(); hostEntity1.setHostName("h1"); hostDAO.merge(hostEntity1); EasyMock.replay(db); ActionScheduler scheduler = new ActionScheduler(100, 50, db, aq, fsm, 3, new HostsMap((String) null), unitOfWork, null, conf, entityManagerProviderMock, (HostRoleCommandDAO)null, (HostRoleCommandFactory)null); HostRoleCommand hrc1 = hostRoleCommandFactory.create("h1", Role.NAMENODE, null, RoleCommand.EXECUTE); hrc1.setStatus(HostRoleStatus.COMPLETED); HostRoleCommand hrc3 = hostRoleCommandFactory.create(null, Role.AMBARI_SERVER_ACTION, null, RoleCommand.CUSTOM_COMMAND); hrc3.setStatus(HostRoleStatus.IN_PROGRESS); HostRoleCommand hrc4 = hostRoleCommandFactory.create("h1", Role.FLUME_HANDLER, null, RoleCommand.EXECUTE); hrc4.setStatus(HostRoleStatus.PENDING); List<HostRoleCommand> hostRoleCommands = Arrays.asList(hrc1, hrc3, hrc4); scheduler.cancelHostRoleCommands(hostRoleCommands, "foo"); EasyMock.verify(db); }