@Override
    public void onAddedJobGraph(JobID jobId) {
        runIfStateIs(State.RUNNING, () -> handleAddedJobGraph(jobId));
    }
@Test
    void onAddedJobGraph_ifNotRunning_isBeingIgnored() throws Exception {
        final CompletableFuture<JobID> recoveredJobFuture = new CompletableFuture<>();
        jobGraphStore =
                TestingJobGraphStore.newBuilder()
                        .setRecoverJobGraphFunction(
                                (jobId, jobGraphs) -> {
                                    recoveredJobFuture.complete(jobId);
                                    return jobGraphs.get(jobId);
                                })
                        .build();

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

            // wait until the process has started the dispatcher
            dispatcherLeaderProcess.getDispatcherGateway().get();

            // now add the job graph
            jobGraphStore.putJobGraph(JOB_GRAPH);

            dispatcherLeaderProcess.closeAsync();

            dispatcherLeaderProcess.onAddedJobGraph(JOB_GRAPH.getJobID());

            assertThatThrownBy(
                            () -> recoveredJobFuture.get(10L, TimeUnit.MILLISECONDS),
                            "onAddedJobGraph should be ignored if the leader process is not running.")
                    .isInstanceOf(TimeoutException.class);
        }
    }