@Test
public void testCacheInvalidationIfSuspended() throws Exception {

	final Time timeout = Time.milliseconds(100L);
	final Time timeToLive = Time.hours(1L);
	final JobID jobId = new JobID();
	final AccessExecutionGraph accessExecutionGraph = mock(AccessExecutionGraph.class);
	final AccessExecutionGraph suspendedExecutionGraph = mock(AccessExecutionGraph.class);
	when(suspendedExecutionGraph.getState()).thenReturn(JobStatus.SUSPENDED);
	final JobManagerGateway jobManagerGateway = mock(JobManagerGateway.class);
	// let's first answer with a suspended ExecutionGraph and then only with the correct result
	when(jobManagerGateway.requestJob(eq(jobId), any(Time.class))).thenReturn(
		CompletableFuture.completedFuture(suspendedExecutionGraph),
		CompletableFuture.completedFuture(accessExecutionGraph));
	try (ExecutionGraphCache executionGraphCache = new ExecutionGraphCache(timeout, timeToLive)) {
		CompletableFuture<AccessExecutionGraph> executionGraphFuture = executionGraphCache.getExecutionGraph(jobId, jobManagerGateway);
		assertEquals(suspendedExecutionGraph, executionGraphFuture.get());
		CompletableFuture<AccessExecutionGraph> executionGraphFuture2 = executionGraphCache.getExecutionGraph(jobId, jobManagerGateway);
		assertEquals(accessExecutionGraph, executionGraphFuture2.get());
	}
}