public interface IterationBody {
    
    ....


    /**
     * onEpochWatermarkIncremented(), then the epoch@param inputs The inputs of this record = epochWatermark + 1the subgraph.
     *
 @param perRoundSubBody The computational *logic Supposethat therewant isto abe coordinatorexecuted operator which takes all feedback variable streams (emitted by the iteration body)as per-round.
     * @return The output of the subgraph.
     */
 and the termination criteriastatic streamDataStreamList forEachRound(if
 not null) as inputs. The execution of theDataStreamList graphinputs, createdPerRoundSubBody byperRoundSubBody) the{
     * iteration body willreturn terminatenull;
 when all input streams}


 have been fully consumed/** ANDThe anysub-graph ofinside the followingiteration conditions
body that should be executed * is met:as per-round. */
    interface PerRoundSubBody {

        DataStreamList process(DataStreamList input);
    }
}

package org.apache.flink.iteration;

public class DataStreamList {
	public static DataStreamList of(DataStream<?>... streams);

    // Returns the number of data streams in this list* 1) The termination criteria stream is not null. And the coordinator operator has not observed any new value from
     * the termination criteria stream between two consecutive onEpochWatermarkIncremented invocations.
    public *int 2size() The coordinator operator has not observed any new value from any feedback variable stream between two
     * consecutive onEpochWatermarkIncremented invocations.
     *{...}

    // Returns the data stream at the given index in this list.
    public <T> DataStream<T> get(int index) {...}
}  

public class ReplayableDataStreamList {

    private * Required:final List<DataStream<?>> replayedDataStreams;

    private * 1) All the init variable streams and the data streams must be bounded.
final List<DataStream<?>> nonReplayedStreams;

    private ReplayableDataStreamList(
           * 2) The parallelism of any stream in the initial variable streams must equal the parallelism of the stream at the
     * same index of the feedback variable streams returned by the IterationBody.
     *
     * @param initVariableStreams The initial variable streams. These streams will be merged with the feedback variableList<DataStream<?>> replayedDataStreams, List<DataStream<?>> nonReplayedStreams) {
        this.replayedDataStreams = replayedDataStreams;
        this.nonReplayedStreams = nonReplayedStreams;
    }

    public static ReplayedDataStreamList replay(DataStream<?>... dataStreams) {
        return new ReplayedDataStreamList(Arrays.asList(dataStreams));
    }

    public static NonReplayedDataStreamList notReplay(DataStream<?>... dataStreams) {
     *   return new NonReplayedDataStreamList(Arrays.asList(dataStreams));
    }

    List<DataStream<?>> getReplayedDataStreams() {
        return Collections.unmodifiableList(replayedDataStreams);
    streams before being used as the 1st parameter to invoke the iteration body.
     * @param dataStreams The data streams. These streams will be used as the 2nd parameter to invoke the iteration}

    List<DataStream<?>> getNonReplayedStreams() {
        return Collections.unmodifiableList(nonReplayedStreams);
    }

    private static class ReplayedDataStreamList extends ReplayableDataStreamList {

        public ReplayedDataStreamList(List<DataStream<?>> replayedDataStreams) {
     *       super(replayedDataStreams, Collections.emptyList());
        }

    body.
    public * @param body The computation logic which takes variable/data streams and returns variable/output streams.
     * @return The list of output streams returned by the iteration boy.
ReplayableDataStreamList andNotReplay(DataStream<?>... nonReplayedStreams) {
            return new ReplayableDataStreamList(
                 */
    static DataStreamList iterateBoundedStreamsUntilTermination(DataStreamList initVariableStreams, DataStreamList dataStreams, IterationBody body) {...}
getReplayedDataStreams(), Arrays.asList(nonReplayedStreams));
        }
    /**}

    private *static Thisclass methodNonReplayedDataStreamList canextends useReplayableDataStreamList an{

 iteration body to process records in some bounded data streams iteratively until a
public NonReplayedDataStreamList(List<DataStream<?>> nonReplayedDataStreams) {
       * termination criteria is reached super(e.g. the given number of rounds is completed or no further variable update is
Collections.emptyList(), nonReplayedDataStreams);
        }

        public * needed). Because this method replays records in the data streams, the iteration body does not need to cache thoseReplayableDataStreamList andReplay(DataStream<?>... replayedStreams) {
            return new ReplayableDataStreamList(
     * records to visit those records repeatedly.
     *
     * This method invokes the iteration body with the following parameters:
     * 1) The 1st parameter is a list of input variable streams, which are created as the union of the initial Arrays.asList(replayedStreams), getNonReplayedStreams());
        }
    }
}

package org.apache.flink.iteration;

/** The config for an iteration. */
public class IterationConfig {

    private final OperatorLifeCycle operatorRoundMode;

    private IterationConfig(OperatorLifeCycle operatorRoundMode) { variable
     * streams and thethis.operatorRoundMode corresponding= feedbackoperatorRoundMode;
 variable streams (returned by}

 the iteration body).
 public static IterationConfigBuilder  * 2) The 2nd parameter is a list of replayed data streams, which are created by replaying the initial data streamsnewBuilder() {

        return new IterationConfigBuilder();
    }

    public static class IterationConfigBuilder {

     * round by roundprivate untilOperatorLifeCycle theoperatorRoundMode iteration= terminates. The records in the Nth round will be emitted into the iterationOperatorLifeCycle.ALL_ROUND;

        private IterationConfigBuilder() {}

     * body only ifpublic the low watermark of the first operator in the iteration body >= N - 1.
     *
IterationConfigBuilder setOperatorRoundMode(OperatorLifeCycle operatorRoundMode) {
            this.operatorRoundMode = operatorRoundMode;
        * The epoch values arereturn determinedthis;
 as described below. See IterationListener for how the}

 epoch values are used.
    public *IterationConfig 1build() All{
 records in the initial variable streams has epoch=1.
    return *new 2IterationConfig(operatorRoundMode);
 The records from the initial data streams will}
 be replayed round by}

 round into the iterationpublic body.enum TheOperatorLifeCycle records{
     * in the first round have epoch=1. And records in the Nth round have epoch = N.
     * 3) For any record emitted by this operator into a non-feedback stream, the epoch of this emitted record = the
     * epoch of the input record that triggers this emission. If this record is emitted by
     * onEpochWatermarkIncremented(), then the epoch of this record = epochWatermark.
     * 4) For any record emitted by this operator into a feedback stream, the epoch of the emitted record = the epoch
     * of the input record that triggers this emission + 1. If this record is emitted by onEpochWatermarkIncremented(),
     * then the epoch of this record = epochWatermark + 1.
     *
     * Suppose there is a coordinator operator which takes all feedback variable streams (emitted by the iteration body)
     * and the termination criteria stream (if not null) as inputs. The execution of the graph created by the
     * iteration body will terminate when all input streams have been fully consumed AND any of the following conditions
     * is met:
     * 1) The termination criteria stream is not null. And the coordinator operator has not observed any new value from
     * the termination criteria stream between two consecutive onEpochWatermarkIncremented invocations.
     * 2) The coordinator operator has not observed any new value from any feedback variable stream between two
     * consecutive onEpochWatermarkIncremented invocations.
     *
     * Required:
     * 1) All the init variable streams and the data streams must be bounded.
     * 2) The parallelism of any stream in the initial variable streams must equal the parallelism of the stream at the
     * same index of the feedback variable streams returned by the IterationBody.
     *
     * @param initVariableStreams The initial variable streams. These streams will be merged with the feedback variable
     *                            streams before being used as the 1st parameter to invoke the iteration body.
     * @param initDataStreams The initial data streams. Records from these streams will be repeatedly replayed and used
     *                        as the 2nd parameter to invoke the iteration body.
     * @param body The computation logic which takes variable/data streams and returns variable/output streams.
     * @return The list of output streams returned by the iteration boy.
     */
    static DataStreamList iterateAndReplayBoundedStreamsUntilTermination(DataStreamList initVariableStreams, DataStreamList initDataStreams, IterationBody body) {...}
}

public class SynchronousBoundedLinearRegression {
    private static final int N_DIM = 50;
    private static final int N_EPOCH = 5;
    private static final int N_BATCH_PER_EPOCH = 10;
    private static final OutputTag<double[]> FINAL_MODEL_OUTPUT_TAG = new OutputTag<double[]>{};

    public static void main(String[] args) {
        DataStream<double[]> initParameters = loadParameters().setParallelism(1);
        DataStream<Tuple2<double[], Double>> dataset = loadDataSet().setParallelism(1);

        DataStreamList resultStreams = 
            Iterations.iterateBoundedStreamsUntilTermination(
				DataStreamList.of(initParameters), 
				ReplayableDataStreamList.notReplay(dataset), 
				IterationConfig.newBuilder().setOperatorRoundMode(ALL_ROUND).build();
				(variableStreams, dataStreams) -> {
                DataStream<double[]> parameterUpdates = variableStreams.get(0);
                	DataStream<Tuple2<double[], Double>> dataset = dataStreams.get(0);

                	SingleOutputStreamOperator<double[]> parameters = parameterUpdates.process(new ParametersCacheFunction());
                	DataStream<double[]> modelUpdate = parameters.setParallelism(1)
                   		.broadcast()
	                    .connect(dataset)
    	                .coProcess(new TrainFunction())
        	            .setParallelism(10)
						.process(new ReduceFunction())
						.setParallelism(1)
	
                	return new IterationBodyResult(DataStreamList.of(modelUpdate), DataStreamList.of(parameters.getSideOut(FINAL_MODEL_OUTPUT_TAG)));
            	});
        
        DataStream<double[]> finalModel = resultStreams.get("final_model");
        finalModel.print();
    }

    public static class ParametersCacheFunction extends ProcessFunction<double[], double[]>
        implements IterationListener<double[]> {  
        
        private final double[] parameters = new double[N_DIM];

        public void processElement(double[] update, Context ctx, Collector<O> output) {
            // Suppose we have a util to add the second array to the first.
            ArrayUtils.addWith(parameters, update

 			DataStreamList resultStreams = 
            dataset.add(trainSample)
        }

   Iterations.iterateBoundedStreamsUntilTermination(
				DataStreamList.of(initParameters), 
				ReplayableDataStreamList.notReplay(dataset), 
				IterationConfig.newBuilder().setOperatorRoundMode(ALL_ROUND).build();
				(variableStreams, dataStreams) -> {
     void onEpochWatermarkIncremented(int epochWatermark, Context context, Collector<T> collector) {
            if (epochWatermark == 0) {DataStream<double[]> parameterUpdates = variableStreams.get(0);
                calculateModelUpdate(firstRoundCachedParameter, output);	DataStream<Tuple2<double[], Double>> dataset = dataStreams.get(0);

                firstRoundCachedParameter	SingleOutputStreamOperator<double[]> parameters = null parameterUpdates.process(new ParametersCacheFunction());
            }
    	DataStream<double[]> modelUpdate   }
= parameters.setParallelism(1)
        private void calculateModelUpdate(double[] parameters, Collector<O> output) {
       		.broadcast()
	     List<Tuple2<double[], Double>> samples = sample(dataset);

            double[] modelUpdate = new double[N_DIM];
.connect(dataset)
    	            for (Tuple2<double[], Double> record : samples) {
.coProcess(new TrainFunction())
        	         double diff = (ArrayUtils.muladd(record.f0, parameters) - record.f1);
                ArrayUtils.addWith(modelUpdate, ArrayUtils.multiply(record.f0, diff));setParallelism(10)
					
			     	DataStream<double[]> reduced = forEachRound(DataStreamList.of(modelUpdate), streams -> {
						return streams.<double[]>get(0).windowAll().reduce((x, y) -> ArrayUtils.add(x, y));
					}).<double[]>get(0);
	
            }

    	return new       output.collect(modelUpdateIterationBodyResult(DataStreamList.of(modelUpdate), DataStreamList.of(parameters.getSideOut(FINAL_MODEL_OUTPUT_TAG)));
        }
    	}
});