code: @SuppressWarnings({ "unchecked", "rawtypes" })
  	public DataStream<T> closeWith(DataStream<T> feedbackStream) {
  
  		Collection<StreamTransformation<?>> predecessors = feedbackStream.getTransformation().getTransitivePredecessors();
  
  		if (!predecessors.contains(this.transformation)) {
  			throw new UnsupportedOperationException(
  					"Cannot close an iteration with a feedback DataStream that does not originate from said iteration.");
  		}
  
  		((FeedbackTransformation) getTransformation()).addFeedbackEdge(feedbackStream.getTransformation());
  
  		return feedbackStream;
  	}
docs: Closes the iteration. This method defines the end of the iterative
  program part that will be fed back to the start of the iteration.
  
  <p>A common usage pattern for streaming iterations is to use output
  splitting to send a part of the closing data stream to the head. Refer to
  {@link DataStream#split(org.apache.flink.streaming.api.collector.selector.OutputSelector)}
  for more information.
  
  @param feedbackStream
  {@link DataStream} that will be used as input to the iteration
  head.
  
  @return The feedback stream.