private void declineCheckpoint(long checkpointID, CheckpointFailureReason failureReason) {
        declineCheckpoint(checkpointID, failureReason, null);
    }
@Test
    public void testDeclineCheckpoint() throws Exception {
        TestCheckpointResponder testCheckpointResponder = new TestCheckpointResponder();
        final Task task =
                createTaskBuilder()
                        .setInvokable(InvokableDecliningCheckpoints.class)
                        .setCheckpointResponder(testCheckpointResponder)
                        .build(Executors.directExecutor());
        assertCheckpointDeclined(
                task,
                testCheckpointResponder,
                1,
                CheckpointFailureReason.CHECKPOINT_DECLINED_TASK_NOT_READY);

        task.startTaskThread();
        try {
            awaitInvokableLatch(task);
            assertEquals(ExecutionState.RUNNING, task.getExecutionState());

            assertCheckpointDeclined(
                    task,
                    testCheckpointResponder,
                    InvokableDecliningCheckpoints.REJECTED_EXECUTION_CHECKPOINT_ID,
                    CheckpointFailureReason.CHECKPOINT_DECLINED_TASK_CLOSING);
            assertCheckpointDeclined(
                    task,
                    testCheckpointResponder,
                    InvokableDecliningCheckpoints.THROWING_CHECKPOINT_ID,
                    CheckpointFailureReason.TASK_FAILURE);
            assertCheckpointDeclined(
                    task,
                    testCheckpointResponder,
                    InvokableDecliningCheckpoints.TRIGGERING_FAILED_CHECKPOINT_ID,
                    CheckpointFailureReason.TASK_FAILURE);
        } finally {
            triggerInvokableLatch(task);
            task.getExecutingThread().join();
        }
        assertEquals(ExecutionState.FINISHED, task.getTerminationFuture().getNow(null));
    }