@Override
    public void onRemovedJobGraph(JobID jobId) {
        runIfStateIs(State.RUNNING, () -> handleRemovedJobGraph(jobId));
    }
@Test
    void onRemovedJobGraph_failingRemovalCall_failsFatally() throws Exception {
        final FlinkException testException = new FlinkException("Test exception");

        final TestingDispatcherGatewayService testingDispatcherService =
                TestingDispatcherGatewayService.newBuilder()
                        .setOnRemovedJobGraphFunction(
                                jobID -> FutureUtils.completedExceptionally(testException))
                        .build();

        dispatcherServiceFactory =
                createFactoryBasedOnGenericSupplier(() -> testingDispatcherService);

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

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

            // now notify the dispatcher service
            dispatcherLeaderProcess.onRemovedJobGraph(JOB_GRAPH.getJobID());

            final Throwable fatalError = fatalErrorHandler.getErrorFuture().join();

            assertThat(fatalError).hasCause(testException);

            fatalErrorHandler.clearError();
        }
    }