@Override
    public void onAddedJobGraph(JobID jobId) {
        runIfStateIs(State.RUNNING, () -> handleAddedJobGraph(jobId));
    }
@Test
    void onAddedJobGraph_failingRecovery_propagatesTheFailure() throws Exception {
        final FlinkException expectedFailure = new FlinkException("Expected failure");
        jobGraphStore =
                TestingJobGraphStore.newBuilder()
                        .setRecoverJobGraphFunction(
                                (ignoredA, ignoredB) -> {
                                    throw expectedFailure;
                                })
                        .build();

        try (final SessionDispatcherLeaderProcess dispatcherLeaderProcess =
                createDispatcherLeaderProcess()) {
            dispatcherLeaderProcess.start();

            // wait first for the dispatcher service to be created
            dispatcherLeaderProcess.getDispatcherGateway().get();

            jobGraphStore.putJobGraph(JOB_GRAPH);
            dispatcherLeaderProcess.onAddedJobGraph(JOB_GRAPH.getJobID());

            assertThatFuture(fatalErrorHandler.getErrorFuture())
                    .eventuallySucceeds()
                    .extracting(FlinkAssertions::chainOfCauses, STREAM_THROWABLE)
                    .contains(expectedFailure);

            assertThat(dispatcherLeaderProcess.getState())
                    .isEqualTo(SessionDispatcherLeaderProcess.State.STOPPED);

            fatalErrorHandler.clearError();
        }
    }