@Override
    public CompletableFuture<Void> globalCleanupAsync(JobID jobID, Executor executor) {
        return CompletableFuture.runAsync(
                () -> {
                    logger.info("Clean up the high availability data for job {}.", jobID);
                    try {
                        internalCleanupJobData(jobID);
                    } catch (Exception e) {
                        throw new CompletionException(e);
                    }
                    logger.info(
                            "Finished cleaning up the high availability data for job {}.", jobID);
                },
                executor);
    }
@Test
    void testCleanupJobData() throws Exception {
        final Queue<CloseOperations> closeOperations = new ArrayDeque<>(3);
        final TestingBlobStoreService testingBlobStoreService =
                new TestingBlobStoreService(closeOperations);

        JobID jobID = new JobID();
        CompletableFuture<JobID> jobCleanupFuture = new CompletableFuture<>();

        final TestingHaServices haServices =
                new TestingHaServices(
                        new Configuration(),
                        Executors.directExecutor(),
                        testingBlobStoreService,
                        closeOperations,
                        () -> {},
                        jobCleanupFuture::complete);

        haServices.globalCleanupAsync(jobID, Executors.directExecutor()).join();
        JobID jobIDCleaned = jobCleanupFuture.get();
        assertThat(jobIDCleaned).isEqualTo(jobID);
    }