code: @Override
  	OperatorStateHandle closeAndGetHandle() throws IOException {
  		StreamStateHandle streamStateHandle = delegate.closeAndGetHandle();
  
  		if (null == streamStateHandle) {
  			return null;
  		}
  
  		if (partitionOffsets.isEmpty() && delegate.getPos() > initialPosition) {
  			startNewPartition();
  		}
  
  		Map<String, OperatorStateHandle.StateMetaInfo> offsetsMap = new HashMap<>(1);
  
  		OperatorStateHandle.StateMetaInfo metaInfo =
  				new OperatorStateHandle.StateMetaInfo(
  						partitionOffsets.toArray(),
  					OperatorStateHandle.Mode.SPLIT_DISTRIBUTE);
  
  		offsetsMap.put(DefaultOperatorStateBackend.DEFAULT_OPERATOR_STATE_NAME, metaInfo);
  
  		return new OperatorStreamStateHandle(offsetsMap, streamStateHandle);
  	}
docs: This method should not be public so as to not expose internals to user code.