@Override
    public void onAddedJobGraph(JobID jobId) {
        runIfStateIs(State.RUNNING, () -> handleAddedJobGraph(jobId));
    }
@Test
    void onAddedJobGraph_submitsRecoveredJob() throws Exception {
        final CompletableFuture<JobGraph> submittedJobFuture = new CompletableFuture<>();
        final TestingDispatcherGateway testingDispatcherGateway =
                TestingDispatcherGateway.newBuilder()
                        .setSubmitFunction(
                                submittedJob -> {
                                    submittedJobFuture.complete(submittedJob);
                                    return CompletableFuture.completedFuture(Acknowledge.get());
                                })
                        .build();

        dispatcherServiceFactory =
                createFactoryBasedOnGenericSupplier(
                        () ->
                                TestingDispatcherGatewayService.newBuilder()
                                        .setDispatcherGateway(testingDispatcherGateway)
                                        .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());

            final JobGraph submittedJobGraph = submittedJobFuture.get();

            assertThat(submittedJobGraph.getJobID()).isEqualTo(JOB_GRAPH.getJobID());
        }
    }