@Override
    public CompletableFuture<Void> globalCleanupAsync(JobID jobId, Executor executor) {
        checkNotNull(jobId, "Job ID");

        return runAsyncWithLockAssertRunning(
                () -> {
                    LOG.debug("Removing job graph {} from {}.", jobId, jobGraphStateHandleStore);

                    final String name = jobGraphStoreUtil.jobIDToName(jobId);
                    releaseAndRemoveOrThrowCompletionException(jobId, name);

                    addedJobGraphs.remove(jobId);

                    LOG.info("Removed job graph {} from {}.", jobId, jobGraphStateHandleStore);
                },
                executor);
    }
@Test
    public void testGlobalCleanupFailsIfRemovalReturnsFalse() throws Exception {
        final TestingStateHandleStore<JobGraph> stateHandleStore =
                builder.setRemoveFunction(name -> false).build();

        final JobGraphStore jobGraphStore = createAndStartJobGraphStore(stateHandleStore);
        assertThrows(
                ExecutionException.class,
                () ->
                        jobGraphStore
                                .globalCleanupAsync(
                                        testingJobGraph.getJobID(), Executors.directExecutor())
                                .get());
    }