code: 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;
  		}
  	}
docs: 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.