@Override
    public void close() {
        if (asyncCheckpointState.compareAndSet(
                AsyncCheckpointState.RUNNING, AsyncCheckpointState.DISCARDED)) {

            try {
                final Tuple2<Long, Long> tuple = cleanup();
                reportAbortedSnapshotStats(tuple.f0, tuple.f1);
            } catch (Exception cleanupException) {
                LOG.warn(
                        "Could not properly clean up the async checkpoint runnable.",
                        cleanupException);
            }
        } else {
            logFailedCleanupAttempt();
        }
    }
@Test
    void testReportIncompleteStats() {
        long checkpointId = 1000L;
        TestEnvironment env = new TestEnvironment();
        new AsyncCheckpointRunnable(
                        new HashMap<>(),
                        new CheckpointMetaData(checkpointId, 1),
                        new CheckpointMetricsBuilder(),
                        0,
                        "Task Name",
                        r -> {},
                        env,
                        (msg, ex) -> {},
                        false,
                        false,
                        () -> true)
                .close();
        assertThat(((TestTaskStateManager) env.getTaskStateManager()).getReportedCheckpointId())
                .isEqualTo(checkpointId);
    }