@Test public void testJobShutdown() { WorkerRegistryV2 workerRegistryV2 = new WorkerRegistryV2(); LifecycleEventPublisher eventPublisher = new LifecycleEventPublisherImpl(new AuditEventSubscriberLoggingImpl(), new StatusEventSubscriberLoggingImpl(), new DummyWorkerEventSubscriberImpl(workerRegistryV2)); Map<StageScalingPolicy.ScalingReason, StageScalingPolicy.Strategy> smap = new HashMap<>(); smap.put(StageScalingPolicy.ScalingReason.CPU, new StageScalingPolicy.Strategy(StageScalingPolicy.ScalingReason.CPU, 0.5, 0.75, null)); smap.put(StageScalingPolicy.ScalingReason.DataDrop, new StageScalingPolicy.Strategy(StageScalingPolicy.ScalingReason.DataDrop, 0.0, 2.0, null)); SchedulingInfo sInfo = new SchedulingInfo.Builder() .numberOfStages(1) .multiWorkerScalableStageWithConstraints(1, new MachineDefinition(1.0,1.0,1.0,3), Lists.newArrayList(), Lists.newArrayList(), new StageScalingPolicy(1, 0, 10, 1, 1, 0, smap)) .build(); String clusterName = "testJobShutdown"; MantisScheduler schedulerMock = mock(MantisScheduler.class); MantisJobStore jobStoreMock = mock(MantisJobStore.class); try { ActorRef jobActor = JobTestHelper.submitSingleStageScalableJob(system,probe, clusterName, sInfo, schedulerMock, jobStoreMock, eventPublisher); assertEquals(2, workerRegistryV2.getNumRunningWorkers()); jobActor.tell(new JobClusterProto.KillJobRequest( new JobId(clusterName,1), "test reason", JobCompletedReason.Normal, "nj", probe.getRef()), probe.getRef()); probe.expectMsgClass(JobClusterProto.KillJobResponse.class); Thread.sleep(1000); int cnt = 0; for(int i=0; i<100; i++) { cnt++; if(workerRegistryV2.getNumRunningWorkers() == 0) { break; } } assertTrue(cnt < 100); } catch (Exception e) { e.printStackTrace(); } }
@Override public int getNumRunningWorkers() { if(logger.isDebugEnabled()) { logger.debug("In getNumRunningWorkers"); } int cnt = jobToWorkerInfoMap.values().stream() .map(workerList -> workerList.stream() .filter(wm -> WorkerState.isRunningState(wm.getState())) .collect(Collectors.toList()) .size() ) .reduce(0,(a, b) -> a + b); if(logger.isDebugEnabled()) { logger.debug("Returning {} from getNumRunningWorkers", cnt); } return cnt; }
WorkerRegistryV2 implements WorkerRegistry, WorkerEventSubscriber { @Override public int getNumRunningWorkers() { if(logger.isDebugEnabled()) { logger.debug("In getNumRunningWorkers"); } int cnt = jobToWorkerInfoMap.values().stream() .map(workerList -> workerList.stream() .filter(wm -> WorkerState.isRunningState(wm.getState())) .collect(Collectors.toList()) .size() ) .reduce(0,(a, b) -> a + b); if(logger.isDebugEnabled()) { logger.debug("Returning {} from getNumRunningWorkers", cnt); } return cnt; } }
WorkerRegistryV2 implements WorkerRegistry, WorkerEventSubscriber { @Override public int getNumRunningWorkers() { if(logger.isDebugEnabled()) { logger.debug("In getNumRunningWorkers"); } int cnt = jobToWorkerInfoMap.values().stream() .map(workerList -> workerList.stream() .filter(wm -> WorkerState.isRunningState(wm.getState())) .collect(Collectors.toList()) .size() ) .reduce(0,(a, b) -> a + b); if(logger.isDebugEnabled()) { logger.debug("Returning {} from getNumRunningWorkers", cnt); } return cnt; } WorkerRegistryV2(); }
WorkerRegistryV2 implements WorkerRegistry, WorkerEventSubscriber { @Override public int getNumRunningWorkers() { if(logger.isDebugEnabled()) { logger.debug("In getNumRunningWorkers"); } int cnt = jobToWorkerInfoMap.values().stream() .map(workerList -> workerList.stream() .filter(wm -> WorkerState.isRunningState(wm.getState())) .collect(Collectors.toList()) .size() ) .reduce(0,(a, b) -> a + b); if(logger.isDebugEnabled()) { logger.debug("Returning {} from getNumRunningWorkers", cnt); } return cnt; } WorkerRegistryV2(); static Props props(); void clearState(); @Override int getNumRunningWorkers(); @Override Set<WorkerId> getAllRunningWorkers(); @Override Map<WorkerId, String> getAllRunningWorkerSlaveIdMappings(); @Override boolean isWorkerValid(WorkerId workerId); @Override Optional<Long> getAcceptedAt(WorkerId workerId); @Override void process(LifecycleEventsProto.WorkerListChangedEvent event); @Override void process(LifecycleEventsProto.JobStatusEvent statusEvent); }
WorkerRegistryV2 implements WorkerRegistry, WorkerEventSubscriber { @Override public int getNumRunningWorkers() { if(logger.isDebugEnabled()) { logger.debug("In getNumRunningWorkers"); } int cnt = jobToWorkerInfoMap.values().stream() .map(workerList -> workerList.stream() .filter(wm -> WorkerState.isRunningState(wm.getState())) .collect(Collectors.toList()) .size() ) .reduce(0,(a, b) -> a + b); if(logger.isDebugEnabled()) { logger.debug("Returning {} from getNumRunningWorkers", cnt); } return cnt; } WorkerRegistryV2(); static Props props(); void clearState(); @Override int getNumRunningWorkers(); @Override Set<WorkerId> getAllRunningWorkers(); @Override Map<WorkerId, String> getAllRunningWorkerSlaveIdMappings(); @Override boolean isWorkerValid(WorkerId workerId); @Override Optional<Long> getAcceptedAt(WorkerId workerId); @Override void process(LifecycleEventsProto.WorkerListChangedEvent event); @Override void process(LifecycleEventsProto.JobStatusEvent statusEvent); static final WorkerRegistryV2 INSTANCE; }