public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(boolean isPeriodic) {
        return triggerCheckpointFromCheckpointThread(checkpointProperties, null, isPeriodic);
    }
@Test
    void testCheckpointStatsTrackerPendingCheckpointCallback() throws Exception {
        // set up the coordinator and validate the initial state
        CheckpointStatsTracker tracker = mock(CheckpointStatsTracker.class);
        CheckpointCoordinator checkpointCoordinator =
                new CheckpointCoordinatorBuilder()
                        .setTimer(manuallyTriggeredScheduledExecutor)
                        .setCheckpointStatsTracker(tracker)
                        .build(EXECUTOR_RESOURCE.getExecutor());

        when(tracker.reportPendingCheckpoint(
                        anyLong(), anyLong(), any(CheckpointProperties.class), any(Map.class)))
                .thenReturn(mock(PendingCheckpointStats.class));

        // Trigger a checkpoint and verify callback
        CompletableFuture<CompletedCheckpoint> checkpointFuture =
                checkpointCoordinator.triggerCheckpoint(false);
        manuallyTriggeredScheduledExecutor.triggerAll();
        FutureUtils.throwIfCompletedExceptionally(checkpointFuture);

        verify(tracker, times(1))
                .reportPendingCheckpoint(
                        eq(1L),
                        any(Long.class),
                        eq(
                                CheckpointProperties.forCheckpoint(
                                        CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION)),
                        any());
    }