@Test @SuppressWarnings("unchecked") public void testAbortDiscardsState() throws Exception { CheckpointProperties props = new CheckpointProperties(false, CheckpointType.SAVEPOINT, false, false, false, false, false); QueueExecutor executor = new QueueExecutor(); OperatorState state = mock(OperatorState.class); doNothing().when(state).registerSharedStates(any(SharedStateRegistry.class)); PendingCheckpoint pending = createPendingCheckpoint(props, executor); setTaskState(pending, state); pending.abort(CheckpointFailureReason.CHECKPOINT_DECLINED); executor.runQueuedCommands(); verify(state, times(1)).discardState(); Mockito.reset(state); pending = createPendingCheckpoint(props, executor); setTaskState(pending, state); pending.abort(CheckpointFailureReason.CHECKPOINT_DECLINED, new Exception("Expected Test Exception")); executor.runQueuedCommands(); verify(state, times(1)).discardState(); Mockito.reset(state); pending = createPendingCheckpoint(props, executor); setTaskState(pending, state); pending.abort(CheckpointFailureReason.CHECKPOINT_EXPIRED); executor.runQueuedCommands(); verify(state, times(1)).discardState(); Mockito.reset(state); pending = createPendingCheckpoint(props, executor); setTaskState(pending, state); pending.abort(CheckpointFailureReason.CHECKPOINT_SUBSUMED); executor.runQueuedCommands(); verify(state, times(1)).discardState(); }
public void abort(CheckpointFailureReason reason, @Nullable Throwable cause) { try { CheckpointException exception = new CheckpointException(reason, cause); onCompletionPromise.completeExceptionally(exception); reportFailedCheckpoint(exception); assertAbortSubsumedForced(reason); } finally { dispose(true); } }
PendingCheckpoint { public void abort(CheckpointFailureReason reason, @Nullable Throwable cause) { try { CheckpointException exception = new CheckpointException(reason, cause); onCompletionPromise.completeExceptionally(exception); reportFailedCheckpoint(exception); assertAbortSubsumedForced(reason); } finally { dispose(true); } } }
PendingCheckpoint { public void abort(CheckpointFailureReason reason, @Nullable Throwable cause) { try { CheckpointException exception = new CheckpointException(reason, cause); onCompletionPromise.completeExceptionally(exception); reportFailedCheckpoint(exception); assertAbortSubsumedForced(reason); } finally { dispose(true); } } PendingCheckpoint(
			JobID jobId,
			long checkpointId,
			long checkpointTimestamp,
			Map<ExecutionAttemptID, ExecutionVertex> verticesToConfirm,
			CheckpointProperties props,
			CheckpointStorageLocation targetLocation,
			Executor executor); }
PendingCheckpoint { public void abort(CheckpointFailureReason reason, @Nullable Throwable cause) { try { CheckpointException exception = new CheckpointException(reason, cause); onCompletionPromise.completeExceptionally(exception); reportFailedCheckpoint(exception); assertAbortSubsumedForced(reason); } finally { dispose(true); } } PendingCheckpoint(
			JobID jobId,
			long checkpointId,
			long checkpointTimestamp,
			Map<ExecutionAttemptID, ExecutionVertex> verticesToConfirm,
			CheckpointProperties props,
			CheckpointStorageLocation targetLocation,
			Executor executor); JobID getJobId(); long getCheckpointId(); long getCheckpointTimestamp(); int getNumberOfNonAcknowledgedTasks(); int getNumberOfAcknowledgedTasks(); Map<OperatorID, OperatorState> getOperatorStates(); boolean isFullyAcknowledged(); boolean isAcknowledgedBy(ExecutionAttemptID executionAttemptId); boolean isDiscarded(); boolean canBeSubsumed(); boolean setCancellerHandle(ScheduledFuture<?> cancellerHandle); CompletableFuture<CompletedCheckpoint> getCompletionFuture(); CompletedCheckpoint finalizeCheckpoint(); TaskAcknowledgeResult acknowledgeTask(
			ExecutionAttemptID executionAttemptId,
			TaskStateSnapshot operatorSubtaskStates,
			CheckpointMetrics metrics); void addMasterState(MasterState state); void abort(CheckpointFailureReason reason, @Nullable Throwable cause); void abort(CheckpointFailureReason reason); @Override String toString(); }
PendingCheckpoint { public void abort(CheckpointFailureReason reason, @Nullable Throwable cause) { try { CheckpointException exception = new CheckpointException(reason, cause); onCompletionPromise.completeExceptionally(exception); reportFailedCheckpoint(exception); assertAbortSubsumedForced(reason); } finally { dispose(true); } } PendingCheckpoint(
			JobID jobId,
			long checkpointId,
			long checkpointTimestamp,
			Map<ExecutionAttemptID, ExecutionVertex> verticesToConfirm,
			CheckpointProperties props,
			CheckpointStorageLocation targetLocation,
			Executor executor); JobID getJobId(); long getCheckpointId(); long getCheckpointTimestamp(); int getNumberOfNonAcknowledgedTasks(); int getNumberOfAcknowledgedTasks(); Map<OperatorID, OperatorState> getOperatorStates(); boolean isFullyAcknowledged(); boolean isAcknowledgedBy(ExecutionAttemptID executionAttemptId); boolean isDiscarded(); boolean canBeSubsumed(); boolean setCancellerHandle(ScheduledFuture<?> cancellerHandle); CompletableFuture<CompletedCheckpoint> getCompletionFuture(); CompletedCheckpoint finalizeCheckpoint(); TaskAcknowledgeResult acknowledgeTask(
			ExecutionAttemptID executionAttemptId,
			TaskStateSnapshot operatorSubtaskStates,
			CheckpointMetrics metrics); void addMasterState(MasterState state); void abort(CheckpointFailureReason reason, @Nullable Throwable cause); void abort(CheckpointFailureReason reason); @Override String toString();  }