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 failure in requesting an AccessExecutionGraph from the gateway, will not create
	 * a cache entry --> another cache request will trigger a new gateway request.
	 */
Tests that a failure in requesting an AccessExecutionGraph from the gateway, will not create
a cache entry --> another cache request will trigger a new gateway request.
Tests that a failure in requesting an AccessExecutionGraph from the gateway will not create a cache entry -- > another cache request will trigger a new gateway request .
@Test
	public void testImmediateCacheInvalidationAfterFailure() 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 JobManagerGateway jobManagerGateway = mock(JobManagerGateway.class);
		when(jobManagerGateway.requestJob(eq(jobId), any(Time.class))).thenReturn(
			FutureUtils.completedExceptionally(new FlinkJobNotFoundException(jobId)),
			CompletableFuture.completedFuture(accessExecutionGraph));
		try (ExecutionGraphCache executionGraphCache = new ExecutionGraphCache(timeout, timeToLive)) {
			CompletableFuture<AccessExecutionGraph> executionGraphFuture = executionGraphCache.getExecutionGraph(jobId, jobManagerGateway);
			try {
				executionGraphFuture.get();
				fail("The execution graph future should have been completed exceptionally.");
			} catch (ExecutionException ee) {
				assertTrue(ee.getCause() instanceof FlinkException);
			}
			CompletableFuture<AccessExecutionGraph> executionGraphFuture2 = executionGraphCache.getExecutionGraph(jobId, jobManagerGateway);
			assertEquals(accessExecutionGraph, executionGraphFuture2.get());
		}
	}
@ Test public void testImmediateCacheInvalidationAfterFailure ( ) 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 JobManagerGateway jobManagerGateway = mock ( JobManagerGateway . class ) ; when ( jobManagerGateway . requestJob ( eq ( jobId ) , any ( Time . class ) ) ) . thenReturn ( FutureUtils . completedExceptionally ( new FlinkJobNotFoundException ( jobId ) ) , CompletableFuture . completedFuture ( accessExecutionGraph ) ) ; try ( ExecutionGraphCache executionGraphCache = new ExecutionGraphCache ( timeout , timeToLive ) ) { CompletableFuture < AccessExecutionGraph > executionGraphFuture = executionGraphCache . getExecutionGraph ( jobId , jobManagerGateway ) ; try { executionGraphFuture . get ( ) ; fail ( "The execution graph future should have been completed exceptionally." ) ; } catch ( ExecutionException ee ) { assertTrue ( ee . getCause ( ) instanceof FlinkException ) ; } CompletableFuture < AccessExecutionGraph > executionGraphFuture2 = executionGraphCache . getExecutionGraph ( jobId , jobManagerGateway ) ; assertEquals ( accessExecutionGraph , executionGraphFuture2 . get ( ) ) ; } }
Tests that a failure in requesting an AccessExecutionGraph from the gateway, will not create
 a cache entry --> another cache request will trigger a new gateway request.
Tests that a failure in requesting an AccessExecutionGraph from the gateway will not create a cache entry -- > another cache request will trigger a new gateway request .
// let's first answer with a JobNotFoundException and then only with the correct result