@Override
    public void onRemovedJobGraph(JobID jobId) {
        runIfStateIs(State.RUNNING, () -> handleRemovedJobGraph(jobId));
    }
@Test
    void onRemovedJobGraph_terminatesRunningJob() throws Exception {
        jobGraphStore =
                TestingJobGraphStore.newBuilder()
                        .setInitialJobGraphs(Collections.singleton(JOB_GRAPH))
                        .build();

        final CompletableFuture<JobID> terminateJobFuture = new CompletableFuture<>();
        final TestingDispatcherGatewayService testingDispatcherService =
                TestingDispatcherGatewayService.newBuilder()
                        .setOnRemovedJobGraphFunction(
                                jobID -> {
                                    terminateJobFuture.complete(jobID);
                                    return FutureUtils.completedVoidFuture();
                                })
                        .build();

        dispatcherServiceFactory =
                createFactoryBasedOnGenericSupplier(() -> testingDispatcherService);

        final ExecutorService executorService = Executors.newSingleThreadExecutor();
        try (final SessionDispatcherLeaderProcess dispatcherLeaderProcess =
                createDispatcherLeaderProcess()) {
            dispatcherLeaderProcess.start();

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

            // now remove the Job from the JobGraphStore and notify the dispatcher service
            jobGraphStore.globalCleanupAsync(JOB_GRAPH.getJobID(), executorService).join();
            dispatcherLeaderProcess.onRemovedJobGraph(JOB_GRAPH.getJobID());

            assertThat(terminateJobFuture.get()).isEqualTo(JOB_GRAPH.getJobID());
        } finally {
            assertThat(executorService.shutdownNow()).isEmpty();
        }
    }