@Test public void testConcurrentAccess() throws Exception { final Time timeout = Time.milliseconds(100L); final Time timeToLive = Time.hours(1L); final CountingRestfulGateway restfulGateway = createCountingRestfulGateway(expectedJobId, CompletableFuture.completedFuture(expectedExecutionGraph)); 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(expectedJobId, restfulGateway), 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(expectedExecutionGraph, executionGraph); } assertThat(restfulGateway.getNumRequestJobCalls(), Matchers.equalTo(1)); } finally { ExecutorUtils.gracefulShutdown(5000L, TimeUnit.MILLISECONDS, executor); } }
public CompletableFuture<AccessExecutionGraph> getExecutionGraph(JobID jobId, RestfulGateway restfulGateway) { return getExecutionGraphInternal(jobId, restfulGateway).thenApply(Function.identity()); }
ExecutionGraphCache implements Closeable { public CompletableFuture<AccessExecutionGraph> getExecutionGraph(JobID jobId, RestfulGateway restfulGateway) { return getExecutionGraphInternal(jobId, restfulGateway).thenApply(Function.identity()); } }
ExecutionGraphCache implements Closeable { public CompletableFuture<AccessExecutionGraph> getExecutionGraph(JobID jobId, RestfulGateway restfulGateway) { return getExecutionGraphInternal(jobId, restfulGateway).thenApply(Function.identity()); } ExecutionGraphCache(
			Time timeout,
			Time timeToLive); }
ExecutionGraphCache implements Closeable { public CompletableFuture<AccessExecutionGraph> getExecutionGraph(JobID jobId, RestfulGateway restfulGateway) { return getExecutionGraphInternal(jobId, restfulGateway).thenApply(Function.identity()); } ExecutionGraphCache(
			Time timeout,
			Time timeToLive); @Override void close(); int size(); CompletableFuture<AccessExecutionGraph> getExecutionGraph(JobID jobId, RestfulGateway restfulGateway); void cleanup(); }
ExecutionGraphCache implements Closeable { public CompletableFuture<AccessExecutionGraph> getExecutionGraph(JobID jobId, RestfulGateway restfulGateway) { return getExecutionGraphInternal(jobId, restfulGateway).thenApply(Function.identity()); } ExecutionGraphCache(
			Time timeout,
			Time timeToLive); @Override void close(); int size(); CompletableFuture<AccessExecutionGraph> getExecutionGraph(JobID jobId, RestfulGateway restfulGateway); void cleanup();  }