code: @Override
  	protected void run() throws Exception {
  
  		final String iterationId = getConfiguration().getIterationId();
  		if (iterationId == null || iterationId.length() == 0) {
  			throw new Exception("Missing iteration ID in the task configuration");
  		}
  
  		final String brokerID = createBrokerIdString(getEnvironment().getJobID(), iterationId ,
  				getEnvironment().getTaskInfo().getIndexOfThisSubtask());
  
  		final long iterationWaitTime = getConfiguration().getIterationWaitTime();
  		final boolean shouldWait = iterationWaitTime > 0;
  
  		final BlockingQueue<StreamRecord<OUT>> dataChannel = new ArrayBlockingQueue<StreamRecord<OUT>>(1);
  
  		// offer the queue for the tail
  		BlockingQueueBroker.INSTANCE.handIn(brokerID, dataChannel);
  		LOG.info("Iteration head {} added feedback queue under {}", getName(), brokerID);
  
  		// do the work
  		try {
  			@SuppressWarnings("unchecked")
  			RecordWriterOutput<OUT>[] outputs = (RecordWriterOutput<OUT>[]) getStreamOutputs();
  
  			// If timestamps are enabled we make sure to remove cyclic watermark dependencies
  			if (isSerializingTimestamps()) {
  				synchronized (getCheckpointLock()) {
  					for (RecordWriterOutput<OUT> output : outputs) {
  						output.emitWatermark(new Watermark(Long.MAX_VALUE));
  					}
  				}
  			}
  
  			while (running) {
  				StreamRecord<OUT> nextRecord = shouldWait ?
  					dataChannel.poll(iterationWaitTime, TimeUnit.MILLISECONDS) :
  					dataChannel.take();
  
  				if (nextRecord != null) {
  					synchronized (getCheckpointLock()) {
  						for (RecordWriterOutput<OUT> output : outputs) {
  							output.collect(nextRecord);
  						}
  					}
  				}
  				else {
  					// done
  					break;
  				}
  			}
  		}
  		finally {
  			// make sure that we remove the queue from the broker, to prevent a resource leak
  			BlockingQueueBroker.INSTANCE.remove(brokerID);
  			LOG.info("Iteration head {} removed feedback queue under {}", getName(), brokerID);
  		}
  	}
docs: ------------------------------------------------------------------------