ActionScheduler implements Runnable { public void setTaskTimeoutAdjustment(boolean val) { taskTimeoutAdjustment = val; } @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 testActionSchedule() throws Exception { Type type = new TypeToken<Map<String, Set<String>>>() {}.getType(); Map<String, List<String>> clusterHostInfo = StageUtils.getGson().fromJson(CLUSTER_HOST_INFO, type); ActionQueue aq = new ActionQueue(); Properties properties = new Properties(); Configuration conf = new Configuration(properties); 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(fsm.getClusterById(anyLong())).thenReturn(oneClusterMock); when(oneClusterMock.getService(anyString())).thenReturn(serviceObj); when(oneClusterMock.getClusterId()).thenReturn(Long.valueOf(1L)); when(serviceObj.getServiceComponent(anyString())).thenReturn(scomp); when(scomp.getServiceComponentHost(anyString())).thenReturn(sch); when(serviceObj.getCluster()).thenReturn(oneClusterMock); Host host = mock(Host.class); HashMap<String, ServiceComponentHost> hosts = new HashMap<String, ServiceComponentHost>(); hosts.put(hostname, sch); HostEntity hostEntity = new HostEntity(); hostEntity.setHostName(hostname); hostDAO.merge(hostEntity); when(scomp.getServiceComponentHosts()).thenReturn(hosts); when(fsm.getHost(anyString())).thenReturn(host); when(host.getState()).thenReturn(HostState.HEALTHY); when(host.getHostName()).thenReturn(hostname); ActionDBAccessor db = mock(ActionDBAccessorImpl.class); HostRoleCommandDAO hostRoleCommandDAOMock = mock(HostRoleCommandDAO.class); Mockito.doNothing().when(hostRoleCommandDAOMock).publishTaskCreateEvent(anyListOf(HostRoleCommand.class)); List<Stage> stages = new ArrayList<Stage>(); Stage s = StageUtils.getATestStage(1, 977, hostname, CLUSTER_HOST_INFO, "{\"host_param\":\"param_value\"}", "{\"stage_param\":\"param_value\"}"); stages.add(s); when(db.getCommandsInProgressCount()).thenReturn(stages.size()); when(db.getStagesInProgress()).thenReturn(stages); RequestEntity request = mock(RequestEntity.class); when(request.isExclusive()).thenReturn(false); when(db.getRequestEntity(anyLong())).thenReturn(request); ActionScheduler scheduler = new ActionScheduler(100, 5, db, aq, fsm, 10000, new HostsMap((String) null), unitOfWork, null, conf, entityManagerProviderMock, hostRoleCommandDAOMock, null); scheduler.setTaskTimeoutAdjustment(false); List<AgentCommand> ac = waitForQueueSize(hostname, aq, 1, scheduler); AgentCommand scheduledCommand = ac.get(0); assertTrue(scheduledCommand instanceof ExecutionCommand); assertEquals("1-977", ((ExecutionCommand) scheduledCommand).getCommandId()); assertEquals(clusterHostInfo, ((ExecutionCommand) scheduledCommand).getClusterHostInfo()); ac = waitForQueueSize(hostname, aq, 2, scheduler); scheduledCommand = ac.get(1); assertTrue(scheduledCommand instanceof ExecutionCommand); assertEquals("1-977", ((ExecutionCommand) scheduledCommand).getCommandId()); assertEquals(clusterHostInfo, ((ExecutionCommand) scheduledCommand).getClusterHostInfo()); s.setHostRoleStatus(hostname, "NAMENODE", HostRoleStatus.COMPLETED); ac = aq.dequeueAll(hostname); ac = waitForQueueSize(hostname, aq, 0, scheduler); EasyMock.verify(entityManagerProviderMock); }
@Test public void testClusterHostInfoCache() throws Exception { Type type = new TypeToken<Map<String, Set<String>>>() {}.getType(); Map<String, Set<String>> clusterHostInfo1 = StageUtils.getGson().fromJson(CLUSTER_HOST_INFO, type); Map<String, Set<String>> clusterHostInfo2 = StageUtils.getGson().fromJson(CLUSTER_HOST_INFO_UPDATED, type); int stageId = 1; int requestId1 = 1; int requestId2 = 2; ActionQueue aq = new ActionQueue(); Properties properties = new Properties(); Configuration conf = new Configuration(properties); 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); Host host = mock(Host.class); HashMap<String, ServiceComponentHost> hosts = new HashMap<String, ServiceComponentHost>(); hosts.put(hostname, sch); when(scomp.getServiceComponentHosts()).thenReturn(hosts); when(fsm.getHost(anyString())).thenReturn(host); when(host.getState()).thenReturn(HostState.HEALTHY); when(host.getHostName()).thenReturn(hostname); ActionDBAccessor db = mock(ActionDBAccessorImpl.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); Stage s1 = StageUtils.getATestStage(requestId1, stageId, hostname, CLUSTER_HOST_INFO, "{\"host_param\":\"param_value\"}", "{\"stage_param\":\"param_value\"}"); Stage s2 = StageUtils.getATestStage(requestId2, stageId, hostname, CLUSTER_HOST_INFO_UPDATED, "{\"host_param\":\"param_value\"}", "{\"stage_param\":\"param_value\"}"); when(db.getCommandsInProgressCount()).thenReturn(1); when(db.getStagesInProgress()).thenReturn(Collections.singletonList(s1)); ActionScheduler scheduler = new ActionScheduler(100, 100, db, aq, fsm, 10000, new HostsMap((String) null), unitOfWork, null, conf, entityManagerProviderMock, hostRoleCommandDAOMock, (HostRoleCommandFactory)null); scheduler.setTaskTimeoutAdjustment(false); List<AgentCommand> ac = waitForQueueSize(hostname, aq, 1, scheduler); assertTrue(ac.get(0) instanceof ExecutionCommand); assertEquals(String.valueOf(requestId1) + "-" + stageId, ((ExecutionCommand) (ac.get(0))).getCommandId()); assertEquals(clusterHostInfo1, ((ExecutionCommand) (ac.get(0))).getClusterHostInfo()); when(db.getCommandsInProgressCount()).thenReturn(1); when(db.getStagesInProgress()).thenReturn(Collections.singletonList(s2)); ac = waitForQueueSize(hostname, aq, 1, scheduler); assertTrue(ac.get(0) instanceof ExecutionCommand); assertEquals(String.valueOf(requestId2) + "-" + stageId, ((ExecutionCommand) (ac.get(0))).getCommandId()); assertEquals(clusterHostInfo2, ((ExecutionCommand) (ac.get(0))).getClusterHostInfo()); }