@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.IN_MEMORY, tim);
	try {
		try {
			scheduler.schedulJob(executionGraph);
		} catch (SchedulingException e) {
			fail(StringUtils.stringifyException(e));
		}
		// Wait for the deployment to complete
		tdm.waitForDeployment();
		assertEquals(executionGraph.getJobID(), tdm.getIDOfLastDeployedJob());
		final List<ExecutionVertex> listOfDeployedVertices = tdm.getListOfLastDeployedVertices();
		assertNotNull(listOfDeployedVertices);
		// Vertices connected via in-memory channels must be deployed in a single cycle.
		assertEquals(2, listOfDeployedVertices.size());
		// Check if the release of the allocated resources works properly by simulating the vertices' life cycle
		assertEquals(0, tim.getNumberOfReleaseMethodCalls());
		// Simulate vertex life cycle
		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) {
			// Ignore exception here
		}
	}
}