@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);
	}
}