flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/ExecutionGraphCacheTest.java
/**
	 * Tests that a cache entry is invalidated if the retrieved {@link AccessExecutionGraph} is in
	 * state {@link JobStatus#SUSPENDED}.
	 *
	 * <p>This test can be removed once we no longer request the actual {@link ExecutionGraph} from the
	 * {@link JobManager}.
	 */
Tests that a cache entry is invalidated if the retrieved  AccessExecutionGraph is in
state  JobStatus#SUSPENDED.

This test can be removed once we no longer request the actual  ExecutionGraph from the
JobManager.
Tests that a cache entry is invalidated if the retrieved AccessExecutionGraph is in state JobStatus#SUSPENDED . This test can be removed once we no longer request the actual ExecutionGraph from the JobManager .
@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);
		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());
		}
	}
@ 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 ) ; 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 ( ) ) ; } }
Tests that a cache entry is invalidated if the retrieved {@link AccessExecutionGraph} is in
 state {@link JobStatus#SUSPENDED}.
Tests that a cache entry is invalidated if the retrieved { @link AccessExecutionGraph } is in state { @link JobStatus#SUSPENDED } .
// let's first answer with a suspended ExecutionGraph and then only with the correct result