@Test
public void testCacheInvalidationIfSwitchToSuspended() 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 SuspendableAccessExecutionGraph toBeSuspendedExecutionGraph = new SuspendableAccessExecutionGraph(jobId);
	final JobManagerGateway jobManagerGateway = mock(JobManagerGateway.class);
	// let's first answer with a JobNotFoundException and then only with the correct result
	when(jobManagerGateway.requestJob(eq(jobId), any(Time.class))).thenReturn(
		CompletableFuture.completedFuture(toBeSuspendedExecutionGraph),
		CompletableFuture.completedFuture(accessExecutionGraph));
	try (ExecutionGraphCache executionGraphCache = new ExecutionGraphCache(timeout, timeToLive)) {
		CompletableFuture<AccessExecutionGraph> executionGraphFuture = executionGraphCache.getExecutionGraph(jobId, jobManagerGateway);
		assertEquals(toBeSuspendedExecutionGraph, executionGraphFuture.get());
		toBeSuspendedExecutionGraph.setJobStatus(JobStatus.SUSPENDED);
		// retrieve the same job from the cache again --> this should return it and invalidate the cache entry
		CompletableFuture<AccessExecutionGraph> executionGraphFuture2 = executionGraphCache.getExecutionGraph(jobId, jobManagerGateway);
		assertEquals(accessExecutionGraph, executionGraphFuture2.get());
		CompletableFuture<AccessExecutionGraph> executionGraphFuture3 = executionGraphCache.getExecutionGraph(jobId, jobManagerGateway);
		assertEquals(accessExecutionGraph, executionGraphFuture3.get());
		verify(jobManagerGateway, times(2)).requestJob(eq(jobId), any(Time.class));
	}
}