private void saveHandleInState(final long checkpointId, final long timestamp) throws Exception {

		//only add handle if a new OperatorState was created since the last snapshot
		if (out != null) {
			int subtaskIdx = getRuntimeContext().getIndexOfThisSubtask();
			StreamStateHandle handle = out.closeAndGetHandle();

			PendingCheckpoint pendingCheckpoint = new PendingCheckpoint(
				checkpointId, subtaskIdx, timestamp, handle);

			if (pendingCheckpoints.contains(pendingCheckpoint)) {
				//we already have a checkpoint stored for that ID that may have been partially written,
				//so we discard this "alternate version" and use the stored checkpoint
				handle.discardState();
			} else {
				pendingCheckpoints.add(pendingCheckpoint);
			}
			out = null;
		}
	}
private void saveHandleInState ( final long checkpointId , final long timestamp ) throws Exception { if ( out != null ) { int subtaskIdx = getRuntimeContext ( ) . getIndexOfThisSubtask ( ) ; StreamStateHandle handle = out . closeAndGetHandle ( ) ; PendingCheckpoint pendingCheckpoint = new PendingCheckpoint ( checkpointId , subtaskIdx , timestamp , handle ) ; if ( pendingCheckpoints . contains ( pendingCheckpoint ) ) { handle . discardState ( ) ; } else { pendingCheckpoints . add ( pendingCheckpoint ) ; } out = null ; } }
Called when a checkpoint barrier arrives. It closes any open streams to the backend
and marks them as pending for committing to the external, third-party storage system.

@param checkpointId the id of the latest received checkpoint.
@throws IOException in case something went wrong when handling the stream to the backend.