@Override
    public CompletableFuture<ExecutionGraphInfo> getExecutionGraphInfo(
            JobID jobId, RestfulGateway restfulGateway) {
        return getExecutionGraphInternal(jobId, restfulGateway).thenApply(Function.identity());
    }
@Test
    void testConcurrentAccess() throws Exception {
        final Time timeout = Time.milliseconds(100L);
        final Time timeToLive = Time.hours(1L);

        final CountingRestfulGateway restfulGateway =
                createCountingRestfulGateway(
                        expectedJobId,
                        CompletableFuture.completedFuture(expectedExecutionGraphInfo));

        final int numConcurrentAccesses = 10;

        final ArrayList<CompletableFuture<ExecutionGraphInfo>> executionGraphFutures =
                new ArrayList<>(numConcurrentAccesses);

        final ExecutorService executor =
                java.util.concurrent.Executors.newFixedThreadPool(numConcurrentAccesses);

        try (ExecutionGraphCache executionGraphCache =
                new DefaultExecutionGraphCache(timeout, timeToLive)) {
            for (int i = 0; i < numConcurrentAccesses; i++) {
                CompletableFuture<ExecutionGraphInfo> executionGraphFuture =
                        CompletableFuture.supplyAsync(
                                        () ->
                                                executionGraphCache.getExecutionGraphInfo(
                                                        expectedJobId, restfulGateway),
                                        executor)
                                .thenCompose(Function.identity());

                executionGraphFutures.add(executionGraphFuture);
            }

            final CompletableFuture<Collection<ExecutionGraphInfo>> allExecutionGraphFutures =
                    FutureUtils.combineAll(executionGraphFutures);

            Collection<ExecutionGraphInfo> allExecutionGraphs = allExecutionGraphFutures.get();

            for (ExecutionGraphInfo executionGraph : allExecutionGraphs) {
                assertThat(executionGraph).isEqualTo(expectedExecutionGraphInfo);
            }

            assertThat(restfulGateway.getNumRequestJobCalls()).isOne();
        } finally {
            ExecutorUtils.gracefulShutdown(5000L, TimeUnit.MILLISECONDS, executor);
        }
    }