@Override
    public CompletableFuture<Void> cleanupAsync(JobID jobId) {
        mainThreadExecutor.assertRunningInMainThread();

        CompletableFuture<Void> cleanupFuture = FutureUtils.completedVoidFuture();
        for (CleanupWithLabel<T> cleanupWithLabel : prioritizedCleanup) {
            cleanupFuture =
                    cleanupFuture.thenCompose(
                            ignoredValue ->
                                    withRetry(
                                            jobId,
                                            cleanupWithLabel.getLabel(),
                                            cleanupWithLabel.getCleanup()));
        }

        return cleanupFuture.thenCompose(
                ignoredValue ->
                        FutureUtils.completeAll(
                                regularCleanup.stream()
                                        .map(
                                                cleanupWithLabel ->
                                                        withRetry(
                                                                jobId,
                                                                cleanupWithLabel.getLabel(),
                                                                cleanupWithLabel.getCleanup()))
                                        .collect(Collectors.toList())));
    }
@Test
    void testMediumPriorityCleanupBlocksAllLowerPrioritizedCleanups() {
        final SingleCallCleanup highPriorityCleanup = SingleCallCleanup.withCompletionOnCleanup();
        final SingleCallCleanup lowerThanHighPriorityCleanup =
                SingleCallCleanup.withoutCompletionOnCleanup();
        final SingleCallCleanup noPriorityCleanup0 = SingleCallCleanup.withCompletionOnCleanup();
        final SingleCallCleanup noPriorityCleanup1 = SingleCallCleanup.withCompletionOnCleanup();

        final DefaultResourceCleaner<CleanupCallback> testInstance =
                createTestInstanceBuilder()
                        .withPrioritizedCleanup("Prio #0", highPriorityCleanup)
                        .withPrioritizedCleanup("Prio #1", lowerThanHighPriorityCleanup)
                        .withRegularCleanup("Reg #0", noPriorityCleanup0)
                        .withRegularCleanup("Reg #1", noPriorityCleanup1)
                        .build();

        assertThat(highPriorityCleanup.isDone()).isFalse();

        final CompletableFuture<Void> overallCleanupResult = testInstance.cleanupAsync(JOB_ID);

        assertThat(highPriorityCleanup.isDone()).isTrue();
        assertThat(lowerThanHighPriorityCleanup.isDone()).isFalse();
        assertThat(noPriorityCleanup0.isDone()).isFalse();
        assertThat(noPriorityCleanup1.isDone()).isFalse();

        assertThat(overallCleanupResult.isDone()).isFalse();

        lowerThanHighPriorityCleanup.completeCleanup();

        assertThat(overallCleanupResult).isCompleted();

        assertThat(highPriorityCleanup.isDone()).isTrue();
        assertThat(lowerThanHighPriorityCleanup.isDone()).isTrue();
        assertThat(noPriorityCleanup0.isDone()).isTrue();
        assertThat(noPriorityCleanup1.isDone()).isTrue();
    }