@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);
		}
	}
@ 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 ) ; BlockingQueueBroker . INSTANCE . handIn ( brokerID , dataChannel ) ; LOG . info ( "Iteration head {} added feedback queue under {}" , getName ( ) , brokerID ) ; try { @ SuppressWarnings ( "unchecked" ) RecordWriterOutput < OUT > [ ] outputs = ( RecordWriterOutput < OUT > [ ] ) getStreamOutputs ( ) ; 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 { break ; } } } finally { BlockingQueueBroker . INSTANCE . remove ( brokerID ) ; LOG . info ( "Iteration head {} removed feedback queue under {}" , getName ( ) , brokerID ) ; } }
------------------------------------------------------------------------