public void abort(
            CheckpointFailureReason reason,
            @Nullable Throwable cause,
            CheckpointsCleaner checkpointsCleaner,
            Runnable postCleanup,
            Executor executor,
            CheckpointStatsTracker statsTracker) {
        try {
            failureCause = new CheckpointException(reason, cause);
            onCompletionPromise.completeExceptionally(failureCause);
            masterTriggerCompletionPromise.completeExceptionally(failureCause);
            assertAbortSubsumedForced(reason);
        } finally {
            dispose(true, checkpointsCleaner, postCleanup, executor);
        }
    }
@Test
    void testAbortDiscardsState() throws Exception {
        CheckpointProperties props =
                new CheckpointProperties(
                        false, CheckpointType.CHECKPOINT, false, false, false, false, false, false);
        QueueExecutor executor = new QueueExecutor();

        OperatorState state = new OperatorState(new OperatorID(), 1, 256);
        OperatorSubtaskStateMock subtaskStateMock = new OperatorSubtaskStateMock();
        OperatorSubtaskState subtaskState = subtaskStateMock.getSubtaskState();
        state.putState(0, subtaskState);

        // Abort declined
        PendingCheckpoint pending = createPendingCheckpoint(props, executor);
        setTaskState(pending, state);

        abort(pending, CheckpointFailureReason.CHECKPOINT_DECLINED);
        // execute asynchronous discard operation
        executor.runQueuedCommands();
        subtaskStateMock.verifyDiscard();

        // Abort error
        subtaskStateMock.reset();

        pending = createPendingCheckpoint(props, executor);
        setTaskState(pending, state);

        abort(pending, CheckpointFailureReason.CHECKPOINT_DECLINED);
        // execute asynchronous discard operation
        executor.runQueuedCommands();
        subtaskStateMock.verifyDiscard();

        // Abort expired
        subtaskStateMock.reset();

        pending = createPendingCheckpoint(props, executor);
        setTaskState(pending, state);

        abort(pending, CheckpointFailureReason.CHECKPOINT_EXPIRED);
        // execute asynchronous discard operation
        executor.runQueuedCommands();
        subtaskStateMock.verifyDiscard();

        // Abort subsumed
        subtaskStateMock.reset();

        pending = createPendingCheckpoint(props, executor);
        setTaskState(pending, state);

        abort(pending, CheckpointFailureReason.CHECKPOINT_SUBSUMED);
        // execute asynchronous discard operation
        executor.runQueuedCommands();
        subtaskStateMock.verifyDiscard();
    }