@Override
    public ResourceCleaner createLocalResourceCleaner(
            ComponentMainThreadExecutor mainThreadExecutor) {
        return DefaultResourceCleaner.forLocallyCleanableResources(
                        mainThreadExecutor, cleanupExecutor, retryStrategy)
                .withPrioritizedCleanup(JOB_MANAGER_RUNNER_REGISTRY_LABEL, jobManagerRunnerRegistry)
                .withRegularCleanup(JOB_GRAPH_STORE_LABEL, jobGraphWriter)
                .withRegularCleanup(BLOB_SERVER_LABEL, blobServer)
                .withRegularCleanup(JOB_MANAGER_METRIC_GROUP_LABEL, jobManagerMetricGroup)
                .build();
    }
@Test
    public void testLocalResourceCleaning() {
        assertCleanupNotTriggered();

        final CompletableFuture<Void> cleanupResultFuture =
                testInstance
                        .createLocalResourceCleaner(
                                ComponentMainThreadExecutorServiceAdapter.forMainThread())
                        .cleanupAsync(JOB_ID);

        assertWaitingForPrioritizedCleanupToFinish();
        assertThat(cleanupResultFuture).isNotCompleted();

        // makes the prioritized JobManagerRunner cleanup result terminate so that other cleanups
        // are triggered
        jobManagerRunnerRegistryLocalCleanupResultFuture.complete(null);

        assertThat(jobManagerRunnerRegistryLocalCleanupFuture).isCompleted();

        assertThat(jobGraphWriterLocalCleanupFuture).isCompleted();
        assertThat(jobGraphWriterGlobalCleanupFuture).isNotDone();

        assertThat(blobServer.getLocalCleanupFuture()).isCompleted();
        assertThat(blobServer.getGlobalCleanupFuture()).isNotDone();

        assertThat(highAvailabilityServicesGlobalCleanupFuture).isNotDone();

        assertJobMetricGroupCleanedUp();

        assertThat(cleanupResultFuture).isCompleted();
    }