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 concurrent accesses only trigger a single AccessExecutionGraph request.
	 */
Tests that concurrent accesses only trigger a single AccessExecutionGraph request.
Tests that concurrent accesses only trigger a single AccessExecutionGraph request .
@Test
	public void testConcurrentAccess() 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(CompletableFuture.completedFuture(accessExecutionGraph));
		final int numConcurrentAccesses = 10;
		final ArrayList<CompletableFuture<AccessExecutionGraph>> executionGraphFutures = new ArrayList<>(numConcurrentAccesses);
		final ExecutorService executor = java.util.concurrent.Executors.newFixedThreadPool(numConcurrentAccesses);
		try (ExecutionGraphCache executionGraphCache = new ExecutionGraphCache(timeout, timeToLive)) {
			for (int i = 0; i < numConcurrentAccesses; i++) {
				CompletableFuture<AccessExecutionGraph> executionGraphFuture = CompletableFuture
					.supplyAsync(
						() -> executionGraphCache.getExecutionGraph(jobId, jobManagerGateway),
						executor)
					.thenCompose(Function.identity());
				executionGraphFutures.add(executionGraphFuture);
			}
			final CompletableFuture<Collection<AccessExecutionGraph>> allExecutionGraphFutures = FutureUtils.combineAll(executionGraphFutures);
			Collection<AccessExecutionGraph> allExecutionGraphs = allExecutionGraphFutures.get();
			for (AccessExecutionGraph executionGraph : allExecutionGraphs) {
				assertEquals(accessExecutionGraph, executionGraph);
			}
			verify(jobManagerGateway, times(1)).requestJob(eq(jobId), any(Time.class));
		} finally {
			ExecutorUtils.gracefulShutdown(5000L, TimeUnit.MILLISECONDS, executor);
		}
	}
@ Test public void testConcurrentAccess ( ) 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 ( CompletableFuture . completedFuture ( accessExecutionGraph ) ) ; final int numConcurrentAccesses = 10 ; final ArrayList < CompletableFuture < AccessExecutionGraph > > executionGraphFutures = new ArrayList < > ( numConcurrentAccesses ) ; final ExecutorService executor = java . util . concurrent . Executors . newFixedThreadPool ( numConcurrentAccesses ) ; try ( ExecutionGraphCache executionGraphCache = new ExecutionGraphCache ( timeout , timeToLive ) ) { for ( int i = 0 ; i < numConcurrentAccesses ; i ++ ) { CompletableFuture < AccessExecutionGraph > executionGraphFuture = CompletableFuture . supplyAsync ( ( ) -> executionGraphCache . getExecutionGraph ( jobId , jobManagerGateway ) , executor ) . thenCompose ( Function . identity ( ) ) ; executionGraphFutures . add ( executionGraphFuture ) ; } final CompletableFuture < Collection < AccessExecutionGraph > > allExecutionGraphFutures = FutureUtils . combineAll ( executionGraphFutures ) ; Collection < AccessExecutionGraph > allExecutionGraphs = allExecutionGraphFutures . get ( ) ; for ( AccessExecutionGraph executionGraph : allExecutionGraphs ) { assertEquals ( accessExecutionGraph , executionGraph ) ; } verify ( jobManagerGateway , times ( 1 ) ) . requestJob ( eq ( jobId ) , any ( Time . class ) ) ; } finally { ExecutorUtils . gracefulShutdown ( 5000L , TimeUnit . MILLISECONDS , executor ) ; } }
Tests that concurrent accesses only trigger a single AccessExecutionGraph request.
Tests that concurrent accesses only trigger a single AccessExecutionGraph request .