QueueScheduler extends AbstractScheduler implements JobStatusListener, ExecutionStageListener { @Override public void schedulJob(final ExecutionGraph executionGraph) throws SchedulingException { final Map<InstanceType, InstanceTypeDescription> availableInstances = getInstanceManager() .getMapOfAvailableInstanceTypes(); final Iterator<ExecutionStage> stageIt = executionGraph.iterator(); while (stageIt.hasNext()) { final InstanceRequestMap instanceRequestMap = new InstanceRequestMap(); final ExecutionStage stage = stageIt.next(); stage.collectRequiredInstanceTypes(instanceRequestMap, ExecutionState.CREATED); final Iterator<Map.Entry<InstanceType, Integer>> it = instanceRequestMap.getMinimumIterator(); while (it.hasNext()) { final Map.Entry<InstanceType, Integer> entry = it.next(); final InstanceTypeDescription descr = availableInstances.get(entry.getKey()); if (descr == null) { throw new SchedulingException("Unable to schedule job: No instance of type " + entry.getKey() + " available"); } if (descr.getMaximumNumberOfAvailableInstances() != -1 && descr.getMaximumNumberOfAvailableInstances() < entry.getValue().intValue()) { throw new SchedulingException("Unable to schedule job: " + entry.getValue().intValue() + " instances of type " + entry.getKey() + " required, but only " + descr.getMaximumNumberOfAvailableInstances() + " are available"); } } } executionGraph.registerJobStatusListener(this); final ExecutionGraphIterator it2 = new ExecutionGraphIterator(executionGraph, true); while (it2.hasNext()) { final ExecutionVertex vertex = it2.next(); vertex.registerExecutionListener(new QueueExecutionListener(this, vertex)); } executionGraph.registerExecutionStageListener(this); synchronized (this.jobQueue) { this.jobQueue.add(executionGraph); } final ExecutionStage executionStage = executionGraph.getCurrentExecutionStage(); try { requestInstances(executionStage); } catch (InstanceException e) { final String exceptionMessage = StringUtils.stringifyException(e); LOG.error(exceptionMessage); this.jobQueue.remove(executionGraph); throw new SchedulingException(exceptionMessage); } } QueueScheduler(final DeploymentManager deploymentManager, final InstanceManager instanceManager); @Override void schedulJob(final ExecutionGraph executionGraph); @Override ExecutionGraph getExecutionGraphByID(final JobID jobID); @Override void shutdown(); @Override void jobStatusHasChanged(final ExecutionGraph executionGraph, final InternalJobStatus newJobStatus,
			final String optionalMessage); @Override void nextExecutionStageEntered(final JobID jobID, final ExecutionStage executionStage);  }
@Test public void testSchedulJobWithInMemoryChannel() { final TestInstanceManager tim = new TestInstanceManager(); final TestDeploymentManager tdm = new TestDeploymentManager(); final QueueScheduler scheduler = new QueueScheduler(tdm, tim); final ExecutionGraph executionGraph = createExecutionGraph(ChannelType.INMEMORY, tim); try { try { scheduler.schedulJob(executionGraph); } catch (SchedulingException e) { fail(StringUtils.stringifyException(e)); } tdm.waitForDeployment(); assertEquals(executionGraph.getJobID(), tdm.getIDOfLastDeployedJob()); final List<ExecutionVertex> listOfDeployedVertices = tdm.getListOfLastDeployedVertices(); assertNotNull(listOfDeployedVertices); assertEquals(2, listOfDeployedVertices.size()); assertEquals(0, tim.getNumberOfReleaseMethodCalls()); for (final ExecutionVertex vertex : listOfDeployedVertices) { vertex.updateExecutionState(ExecutionState.STARTING); vertex.updateExecutionState(ExecutionState.RUNNING); vertex.updateExecutionState(ExecutionState.FINISHING); vertex.updateExecutionState(ExecutionState.FINISHED); } assertEquals(1, tim.getNumberOfReleaseMethodCalls()); } finally { try { LibraryCacheManager.unregister(executionGraph.getJobID()); } catch (IOException ioe) { } } }
@Test public void testSchedulJobWithFileChannel() { final TestInstanceManager tim = new TestInstanceManager(); final TestDeploymentManager tdm = new TestDeploymentManager(); final QueueScheduler scheduler = new QueueScheduler(tdm, tim); final ExecutionGraph executionGraph = createExecutionGraph(ChannelType.FILE, tim); try { try { scheduler.schedulJob(executionGraph); } catch (SchedulingException e) { fail(StringUtils.stringifyException(e)); } tdm.waitForDeployment(); assertEquals(executionGraph.getJobID(), tdm.getIDOfLastDeployedJob()); List<ExecutionVertex> listOfDeployedVertices = tdm.getListOfLastDeployedVertices(); assertNotNull(listOfDeployedVertices); assertEquals(1, listOfDeployedVertices.size()); assertEquals(0, tim.getNumberOfReleaseMethodCalls()); tdm.clear(); ExecutionVertex vertex = listOfDeployedVertices.get(0); vertex.updateExecutionState(ExecutionState.STARTING); vertex.updateExecutionState(ExecutionState.RUNNING); vertex.updateExecutionState(ExecutionState.FINISHING); vertex.updateExecutionState(ExecutionState.FINISHED); assertEquals(0, tim.getNumberOfReleaseMethodCalls()); tdm.waitForDeployment(); assertEquals(executionGraph.getJobID(), tdm.getIDOfLastDeployedJob()); listOfDeployedVertices = tdm.getListOfLastDeployedVertices(); assertNotNull(listOfDeployedVertices); assertEquals(1, listOfDeployedVertices.size()); vertex = listOfDeployedVertices.get(0); vertex.updateExecutionState(ExecutionState.STARTING); vertex.updateExecutionState(ExecutionState.RUNNING); vertex.updateExecutionState(ExecutionState.FINISHING); vertex.updateExecutionState(ExecutionState.FINISHED); assertEquals(1, tim.getNumberOfReleaseMethodCalls()); } finally { try { LibraryCacheManager.unregister(executionGraph.getJobID()); } catch (IOException ioe) { } } }