@Test public void assertRunOneIteration() { when(facadeService.getEligibleJobContext()).thenReturn( Collections.singletonList(JobContext.from(CloudJobConfigurationBuilder.createCloudJobConfiguration("failover_job", CloudJobExecutionType.DAEMON, 1), ExecutionType.FAILOVER))); Map<String, VMAssignmentResult> vmAssignmentResultMap = new HashMap<>(); vmAssignmentResultMap.put("rs1", new VMAssignmentResult("localhost", Collections.singletonList(new VMLeaseObject(OfferBuilder.createOffer("offer_0"))), Sets.newHashSet(mockTaskAssignmentResult("failover_job", ExecutionType.FAILOVER)))); when(taskScheduler.scheduleOnce(ArgumentMatchers.anyList(), ArgumentMatchers.anyList())).thenReturn(new SchedulingResult(vmAssignmentResultMap)); when(facadeService.load("failover_job")).thenReturn(Optional.of(CloudJobConfigurationBuilder.createCloudJobConfiguration("failover_job"))); when(facadeService.getFailoverTaskId(any(MetaInfo.class))).thenReturn(Optional.of(String.format("%s@-@0@-@%s@-@unassigned-slave@-@0", "failover_job", ExecutionType.FAILOVER.name()))); when(taskScheduler.getTaskAssigner()).thenReturn(mock(Action2.class)); taskLaunchScheduledService.runOneIteration(); verify(facadeService).removeLaunchTasksFromQueue(ArgumentMatchers.anyList()); verify(facadeService).loadAppConfig("test_app"); verify(jobEventBus).post(ArgumentMatchers.<JobStatusTraceEvent>any()); }
@Override protected void runOneIteration() { try { LaunchingTasks launchingTasks = new LaunchingTasks(facadeService.getEligibleJobContext()); List<TaskRequest> taskRequests = launchingTasks.getPendingTasks(); if (!taskRequests.isEmpty()) { AppConstraintEvaluator.getInstance().loadAppRunningState(); } Collection<VMAssignmentResult> vmAssignmentResults = taskScheduler.scheduleOnce(taskRequests, LeasesQueue.getInstance().drainTo()).getResultMap().values(); List<TaskContext> taskContextsList = new LinkedList<>(); Map<List<Protos.OfferID>, List<Protos.TaskInfo>> offerIdTaskInfoMap = new HashMap<>(); for (VMAssignmentResult each: vmAssignmentResults) { List<VirtualMachineLease> leasesUsed = each.getLeasesUsed(); List<Protos.TaskInfo> taskInfoList = new ArrayList<>(each.getTasksAssigned().size() * 10); taskInfoList.addAll(getTaskInfoList(launchingTasks.getIntegrityViolationJobs(vmAssignmentResults), each, leasesUsed.get(0).hostname(), leasesUsed.get(0).getOffer())); for (Protos.TaskInfo taskInfo : taskInfoList) { taskContextsList.add(TaskContext.from(taskInfo.getTaskId().getValue())); } offerIdTaskInfoMap.put(getOfferIDs(leasesUsed), taskInfoList); } for (TaskContext each : taskContextsList) { facadeService.addRunning(each); jobEventBus.post(createJobStatusTraceEvent(each)); } facadeService.removeLaunchTasksFromQueue(taskContextsList); for (Entry<List<OfferID>, List<TaskInfo>> each : offerIdTaskInfoMap.entrySet()) { schedulerDriver.launchTasks(each.getKey(), each.getValue()); } } catch (Throwable throwable) { log.error("Launch task error", throwable); } finally { AppConstraintEvaluator.getInstance().clearAppRunningState(); } }
TaskLaunchScheduledService extends AbstractScheduledService { @Override protected void runOneIteration() { try { LaunchingTasks launchingTasks = new LaunchingTasks(facadeService.getEligibleJobContext()); List<TaskRequest> taskRequests = launchingTasks.getPendingTasks(); if (!taskRequests.isEmpty()) { AppConstraintEvaluator.getInstance().loadAppRunningState(); } Collection<VMAssignmentResult> vmAssignmentResults = taskScheduler.scheduleOnce(taskRequests, LeasesQueue.getInstance().drainTo()).getResultMap().values(); List<TaskContext> taskContextsList = new LinkedList<>(); Map<List<Protos.OfferID>, List<Protos.TaskInfo>> offerIdTaskInfoMap = new HashMap<>(); for (VMAssignmentResult each: vmAssignmentResults) { List<VirtualMachineLease> leasesUsed = each.getLeasesUsed(); List<Protos.TaskInfo> taskInfoList = new ArrayList<>(each.getTasksAssigned().size() * 10); taskInfoList.addAll(getTaskInfoList(launchingTasks.getIntegrityViolationJobs(vmAssignmentResults), each, leasesUsed.get(0).hostname(), leasesUsed.get(0).getOffer())); for (Protos.TaskInfo taskInfo : taskInfoList) { taskContextsList.add(TaskContext.from(taskInfo.getTaskId().getValue())); } offerIdTaskInfoMap.put(getOfferIDs(leasesUsed), taskInfoList); } for (TaskContext each : taskContextsList) { facadeService.addRunning(each); jobEventBus.post(createJobStatusTraceEvent(each)); } facadeService.removeLaunchTasksFromQueue(taskContextsList); for (Entry<List<OfferID>, List<TaskInfo>> each : offerIdTaskInfoMap.entrySet()) { schedulerDriver.launchTasks(each.getKey(), each.getValue()); } } catch (Throwable throwable) { log.error("Launch task error", throwable); } finally { AppConstraintEvaluator.getInstance().clearAppRunningState(); } } }
TaskLaunchScheduledService extends AbstractScheduledService { @Override protected void runOneIteration() { try { LaunchingTasks launchingTasks = new LaunchingTasks(facadeService.getEligibleJobContext()); List<TaskRequest> taskRequests = launchingTasks.getPendingTasks(); if (!taskRequests.isEmpty()) { AppConstraintEvaluator.getInstance().loadAppRunningState(); } Collection<VMAssignmentResult> vmAssignmentResults = taskScheduler.scheduleOnce(taskRequests, LeasesQueue.getInstance().drainTo()).getResultMap().values(); List<TaskContext> taskContextsList = new LinkedList<>(); Map<List<Protos.OfferID>, List<Protos.TaskInfo>> offerIdTaskInfoMap = new HashMap<>(); for (VMAssignmentResult each: vmAssignmentResults) { List<VirtualMachineLease> leasesUsed = each.getLeasesUsed(); List<Protos.TaskInfo> taskInfoList = new ArrayList<>(each.getTasksAssigned().size() * 10); taskInfoList.addAll(getTaskInfoList(launchingTasks.getIntegrityViolationJobs(vmAssignmentResults), each, leasesUsed.get(0).hostname(), leasesUsed.get(0).getOffer())); for (Protos.TaskInfo taskInfo : taskInfoList) { taskContextsList.add(TaskContext.from(taskInfo.getTaskId().getValue())); } offerIdTaskInfoMap.put(getOfferIDs(leasesUsed), taskInfoList); } for (TaskContext each : taskContextsList) { facadeService.addRunning(each); jobEventBus.post(createJobStatusTraceEvent(each)); } facadeService.removeLaunchTasksFromQueue(taskContextsList); for (Entry<List<OfferID>, List<TaskInfo>> each : offerIdTaskInfoMap.entrySet()) { schedulerDriver.launchTasks(each.getKey(), each.getValue()); } } catch (Throwable throwable) { log.error("Launch task error", throwable); } finally { AppConstraintEvaluator.getInstance().clearAppRunningState(); } }  }
TaskLaunchScheduledService extends AbstractScheduledService { @Override protected void runOneIteration() { try { LaunchingTasks launchingTasks = new LaunchingTasks(facadeService.getEligibleJobContext()); List<TaskRequest> taskRequests = launchingTasks.getPendingTasks(); if (!taskRequests.isEmpty()) { AppConstraintEvaluator.getInstance().loadAppRunningState(); } Collection<VMAssignmentResult> vmAssignmentResults = taskScheduler.scheduleOnce(taskRequests, LeasesQueue.getInstance().drainTo()).getResultMap().values(); List<TaskContext> taskContextsList = new LinkedList<>(); Map<List<Protos.OfferID>, List<Protos.TaskInfo>> offerIdTaskInfoMap = new HashMap<>(); for (VMAssignmentResult each: vmAssignmentResults) { List<VirtualMachineLease> leasesUsed = each.getLeasesUsed(); List<Protos.TaskInfo> taskInfoList = new ArrayList<>(each.getTasksAssigned().size() * 10); taskInfoList.addAll(getTaskInfoList(launchingTasks.getIntegrityViolationJobs(vmAssignmentResults), each, leasesUsed.get(0).hostname(), leasesUsed.get(0).getOffer())); for (Protos.TaskInfo taskInfo : taskInfoList) { taskContextsList.add(TaskContext.from(taskInfo.getTaskId().getValue())); } offerIdTaskInfoMap.put(getOfferIDs(leasesUsed), taskInfoList); } for (TaskContext each : taskContextsList) { facadeService.addRunning(each); jobEventBus.post(createJobStatusTraceEvent(each)); } facadeService.removeLaunchTasksFromQueue(taskContextsList); for (Entry<List<OfferID>, List<TaskInfo>> each : offerIdTaskInfoMap.entrySet()) { schedulerDriver.launchTasks(each.getKey(), each.getValue()); } } catch (Throwable throwable) { log.error("Launch task error", throwable); } finally { AppConstraintEvaluator.getInstance().clearAppRunningState(); } }   }
TaskLaunchScheduledService extends AbstractScheduledService { @Override protected void runOneIteration() { try { LaunchingTasks launchingTasks = new LaunchingTasks(facadeService.getEligibleJobContext()); List<TaskRequest> taskRequests = launchingTasks.getPendingTasks(); if (!taskRequests.isEmpty()) { AppConstraintEvaluator.getInstance().loadAppRunningState(); } Collection<VMAssignmentResult> vmAssignmentResults = taskScheduler.scheduleOnce(taskRequests, LeasesQueue.getInstance().drainTo()).getResultMap().values(); List<TaskContext> taskContextsList = new LinkedList<>(); Map<List<Protos.OfferID>, List<Protos.TaskInfo>> offerIdTaskInfoMap = new HashMap<>(); for (VMAssignmentResult each: vmAssignmentResults) { List<VirtualMachineLease> leasesUsed = each.getLeasesUsed(); List<Protos.TaskInfo> taskInfoList = new ArrayList<>(each.getTasksAssigned().size() * 10); taskInfoList.addAll(getTaskInfoList(launchingTasks.getIntegrityViolationJobs(vmAssignmentResults), each, leasesUsed.get(0).hostname(), leasesUsed.get(0).getOffer())); for (Protos.TaskInfo taskInfo : taskInfoList) { taskContextsList.add(TaskContext.from(taskInfo.getTaskId().getValue())); } offerIdTaskInfoMap.put(getOfferIDs(leasesUsed), taskInfoList); } for (TaskContext each : taskContextsList) { facadeService.addRunning(each); jobEventBus.post(createJobStatusTraceEvent(each)); } facadeService.removeLaunchTasksFromQueue(taskContextsList); for (Entry<List<OfferID>, List<TaskInfo>> each : offerIdTaskInfoMap.entrySet()) { schedulerDriver.launchTasks(each.getKey(), each.getValue()); } } catch (Throwable throwable) { log.error("Launch task error", throwable); } finally { AppConstraintEvaluator.getInstance().clearAppRunningState(); } }    }