CompletedCheckpoint implements Serializable { public boolean discardOnShutdown(JobStatus jobStatus) throws Exception { if (jobStatus == JobStatus.FINISHED && props.discardOnJobFinished() || jobStatus == JobStatus.CANCELED && props.discardOnJobCancelled() || jobStatus == JobStatus.FAILED && props.discardOnJobFailed() || jobStatus == JobStatus.SUSPENDED && props.discardOnJobSuspended()) { doDiscard(); return true; } else { if (externalPointer != null) { LOG.info("Persistent checkpoint with ID {} at '{}' not discarded.", checkpointID, externalPointer); } return false; } } CompletedCheckpoint(
			JobID job,
			long checkpointID,
			long timestamp,
			long completionTimestamp,
			Map<OperatorID, OperatorState> operatorStates,
			@Nullable Collection<MasterState> masterHookStates,
			CheckpointProperties props,
			@Nullable StreamStateHandle externalizedMetadata,
			@Nullable String externalPointer); JobID getJobId(); long getCheckpointID(); long getTimestamp(); long getDuration(); CheckpointProperties getProperties(); void discardOnFailedStoring(); boolean discardOnSubsume(); boolean discardOnShutdown(JobStatus jobStatus); long getStateSize(); Map<OperatorID, OperatorState> getOperatorStates(); Collection<MasterState> getMasterHookStates(); boolean isExternalized(); @Nullable StreamStateHandle getExternalizedMetadata(); @Nullable String getExternalPointer(); static boolean checkpointsMatch(
		Collection<CompletedCheckpoint> first,
		Collection<CompletedCheckpoint> second); void registerSharedStatesAfterRestored(SharedStateRegistry sharedStateRegistry); @Override String toString(); @Override boolean equals(Object o); @Override int hashCode();  }
@Test public void testDiscard() throws Exception { File file = tmpFolder.newFile(); assertEquals(true, file.exists()); OperatorState state = mock(OperatorState.class); Map<OperatorID, OperatorState> taskStates = new HashMap<>(); taskStates.put(new OperatorID(), state); CompletedCheckpoint checkpoint = new CompletedCheckpoint( new JobID(), 0, 0, 1, taskStates, Collections.<MasterState>emptyList(), CheckpointProperties.forStandardCheckpoint(), new FileStateHandle(new Path(file.toURI()), file.length()), file.getAbsolutePath()); checkpoint.discardOnShutdown(JobStatus.FAILED); assertEquals(false, file.exists()); }