@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 testGlobalCleanupWithNonExistName() throws Exception {
        final CompletableFuture<JobID> removeFuture = new CompletableFuture<>();
        final TestingStateHandleStore<JobGraph> stateHandleStore =
                builder.setRemoveFunction(name -> removeFuture.complete(JobID.fromHexString(name)))
                        .build();

        final JobGraphStore jobGraphStore = createAndStartJobGraphStore(stateHandleStore);
        jobGraphStore
                .globalCleanupAsync(testingJobGraph.getJobID(), Executors.directExecutor())
                .join();

        assertThat(removeFuture.isDone(), is(true));
    }