@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 testConcurrentCleanupWithExceptionFirst() {
        final SingleCallCleanup cleanup0 = SingleCallCleanup.withoutCompletionOnCleanup();
        final SingleCallCleanup cleanup1 = SingleCallCleanup.withoutCompletionOnCleanup();

        final CompletableFuture<Void> cleanupResult =
                createTestInstanceBuilder()
                        .withRegularCleanup("Reg #0", cleanup0)
                        .withRegularCleanup("Reg #1", cleanup1)
                        .build()
                        .cleanupAsync(JOB_ID);

        assertThat(cleanupResult).isNotCompleted();
        assertThat(cleanup0).extracting(SingleCallCleanup::getProcessedJobId).isEqualTo(JOB_ID);
        assertThat(cleanup1).extracting(SingleCallCleanup::getProcessedJobId).isEqualTo(JOB_ID);

        final RuntimeException expectedException = new RuntimeException("Expected exception");
        cleanup0.completeCleanupExceptionally(expectedException);
        assertThat(cleanupResult).isNotCompleted();

        cleanup1.completeCleanup();
        assertThatFuture(cleanupResult)
                .eventuallyFailsWith(ExecutionException.class)
                .extracting(FlinkAssertions::chainOfCauses, STREAM_THROWABLE)
                .hasExactlyElementsOfTypes(
                        ExecutionException.class,
                        FutureUtils.RetryException.class,
                        CompletionException.class,
                        expectedException.getClass())
                .last()
                .isEqualTo(expectedException);
    }