Repository: apache/flink
File: flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java (Java)

Code before:
                        .setCheckpointStatsTracker(tracker)
                        .setTimer(manuallyTriggeredScheduledExecutor)
                        .build(graph);

        CompletableFuture<CompletedCheckpoint> checkpointFuture =
                coordinator.triggerCheckpoint(false);
        manuallyTriggeredScheduledExecutor.triggerAll();

        CompletableFuture<Void> ackTask =
                CompletableFuture.runAsync(
                        () -> {
                            try {
                                ackCheckpoint(
                                        1L,
                                        coordinator,
                                        jobVertexID,
                                        graph,
                                        handle(),
                                        handle(),
                                        handle());
                            } catch (Exception e) {
                                throw new RuntimeException(e);
                            }
                        });

        assertThat(tracker.getReportStartedFuture().get(20, TimeUnit.SECONDS))
                .as("reportCompletedCheckpoint should be started soon when checkpoint is acked.")
                .isNull();

        for (int i = 0; i < 30; i++) {
            assertThat(checkpointFuture)
                    .as(
                            "Checkpoint future should not complete while reportCompletedCheckpoint is blocked")
                    .isNotDone();
            Thread.sleep(100);
        }

        tracker.getReportBlockingFuture().complete(null);

        CompletedCheckpoint result = checkpointFuture.get(5, TimeUnit.SECONDS);
        assertThat(result)
                .as("Checkpoint future should complete after reportCompletedCheckpoint finishes")
                .isNotNull();

        ackTask.get(5, TimeUnit.SECONDS);
    }

    /**
     * A controllable checkpoint stats tracker for testing purposes. Allows precise control over
     * when reportCompletedCheckpoint() completes, enabling verification of execution order and
     * timing in tests.

Review comment (suggestion):
That is likely to end up being flaky test. Test in CI could freeze for 15min and more, so 20 seconds timeout may not be sufficient in general. 
I suggest to use indefinite timeout of at least a few hours

Code after:
        CheckpointCoordinator coordinator =
                new CheckpointCoordinatorBuilder()
                        .setCheckpointStatsTracker(tracker)
                        .setTimer(manuallyTriggeredScheduledExecutor)
                        .build(graph);

        CompletableFuture<CompletedCheckpoint> checkpointFuture =
                coordinator.triggerCheckpoint(false);
        manuallyTriggeredScheduledExecutor.triggerAll();

        CompletableFuture<Void> ackTask =
                CompletableFuture.runAsync(
                        () -> {
                            try {
                                ackCheckpoint(
                                        1L,
                                        coordinator,
                                        jobVertexID,
                                        graph,
                                        handle(),
                                        handle(),
                                        handle());
                            } catch (Exception e) {
                                throw new RuntimeException(e);
                            }
                        });

        assertThat(tracker.getReportStartedFuture().get(TIMEOUT_SECONDS, TimeUnit.SECONDS))
                .as("reportCompletedCheckpoint should be started soon when checkpoint is acked.")
                .isNull();

        for (int i = 0; i < 30; i++) {
            assertThat(checkpointFuture)
                    .as(
                            "Checkpoint future should not complete while reportCompletedCheckpoint is blocked")
                    .isNotDone();
            Thread.sleep(100);
        }

        tracker.getReportBlockingFuture().complete(null);

        CompletedCheckpoint result = checkpointFuture.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
        assertThat(result)
                .as("Checkpoint future should complete after reportCompletedCheckpoint finishes")
                .isNotNull();

        ackTask.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
    }

    /**
     * A controllable checkpoint stats tracker for testing purposes. Allows precise control over