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

        // let's first answer with a JobNotFoundException and then only with the correct result
        final CountingRestfulGateway restfulGateway =
                createCountingRestfulGateway(
                        expectedJobId,
                        FutureUtils.completedExceptionally(
                                new FlinkJobNotFoundException(expectedJobId)),
                        CompletableFuture.completedFuture(expectedExecutionGraphInfo));

        try (ExecutionGraphCache executionGraphCache =
                new DefaultExecutionGraphCache(timeout, timeToLive)) {
            CompletableFuture<ExecutionGraphInfo> executionGraphFuture =
                    executionGraphCache.getExecutionGraphInfo(expectedJobId, restfulGateway);

            assertThatThrownBy(() -> executionGraphFuture.get())
                    .hasCauseInstanceOf(FlinkException.class);

            CompletableFuture<ExecutionGraphInfo> executionGraphFuture2 =
                    executionGraphCache.getExecutionGraphInfo(expectedJobId, restfulGateway);

            assertThatFuture(executionGraphFuture2)
                    .eventuallySucceeds()
                    .isEqualTo(expectedExecutionGraphInfo);
        }
    }