@Override
    public CompletableFuture<CheckpointStatsSnapshot> requestCheckpointStats(
            JobID jobId, Time timeout) {
        return performOperationOnJobMasterGateway(
                        jobId, gateway -> gateway.requestCheckpointStats(timeout))
                .exceptionally(
                        t ->
                                getExecutionGraphInfoFromStore(t, jobId)
                                        .getArchivedExecutionGraph()
                                        .getCheckpointStatsSnapshot());
    }
@Test
    public void testRetrieveCheckpointStatsOnNonExistentJob() throws Exception {
        TestingJobMasterGateway testingJobMasterGateway =
                new TestingJobMasterGatewayBuilder().build();

        dispatcher =
                createAndStartDispatcher(
                        heartbeatServices,
                        haServices,
                        new TestingJobMasterGatewayJobManagerRunnerFactory(
                                testingJobMasterGateway));

        CompletableFuture<CheckpointStatsSnapshot> resultsFuture =
                dispatcher.callAsyncInMainThread(
                        () -> dispatcher.requestCheckpointStats(jobId, TIMEOUT));

        Assertions.assertThat(resultsFuture).failsWithin(Duration.ofSeconds(1));
        Assertions.assertThat(resultsFuture).isCompletedExceptionally();

        Assertions.assertThatThrownBy(resultsFuture::get)
                .hasCauseInstanceOf(FlinkJobNotFoundException.class)
                .hasMessageContaining("Could not find Flink job");
    }