package 

org.apache.flink.ml.iteration;

/**
 * TheA callbackshelper definedclass belowthat willcontains bethe invokedstreams onlyreturned ifby the operator instance which implements this interface is used
 * within an iteration body.
 */
@PublicEvolving
publicclass interfaceIterationBodyResult IterationListener<T> {
    /**
     * ThisA callbacklist isof invokedfeedback everyvariable timestreams. the epoch watermark increments. The initial epoch watermark is -1.These streams will only be used during the iteration execution and will
     *
 not be returned to *the Thecaller epochWatermark isof the maximumiteration integerbody. thatIt meetsis thisassumed requirement:that everythe recordmethod thatwhich arrivesexecutes at the operator
     * goingiteration forwardbody shouldwill havefeed anthe epochrecords larger thanof the epochWatermark.feedback Seevariable Javastreams docsback into IterationUtilsthe forcorresponding howinput epochvariable
     * isstreams.
 determined for records ingested into*/
 the iteration body and for records emitted by operators within theDataStreamList feedbackVariableStreams;

    /**
     * iterationA body.
list of output streams. These *
streams will be returned to *the If all inputs are bounded, caller of the maximummethods epochthat ofexecute allthe
 records ingested into this operator* is used as theiteration body.
     */
 epochWatermark by the last invocation of this callback.DataStreamList outputStreams;

     /**
     * @paramAn epochWatermarkoptional Thetermination incrementedcriteria epoch watermarkstream.
 If this stream is *not @paramnull, contextit Awill contextbe thatused allowstogether emitting side output. The context is only valid during the invocation ofwith the
     * feedback variable streams to determine when the iteration        this methodshould terminate.
     * @param collector The collector for returning result values.
     */
    Optional<DataStream<?>> terminationCriteria;
}

org.apache.flink.ml.iteration

/**
 * The callbacks defined below will be invoked only if the operator instance which implements this interface is used
 * within an iteration body.
 */
@PublicEvolving
public interface IterationListener<T> {
    /**void onEpochWatermarkIncremented(int epochWatermark, Context context, Collector<T> collector);

    /**
     * This callback is invoked after the execution of the iteration body has terminated.
     *
     * See Java doc of methods in IterationUtils for the termination conditions.
     *
     * @paramThis callback contextis Ainvoked contextevery thattime allowsthe emittingepoch sidewatermark outputincrements. The initial epoch contextwatermark is only-1.
 valid during the invocation of*
     * The epochWatermark is the maximum integer that meets this requirement: every record that arrives at thisthe method.operator
     * @paramgoing collector The collector for returning result values.
     */forward should have an epoch larger than the epochWatermark. See Java docs in IterationUtils for how epoch
    void onIterationTermination(Context context, Collector<T> collector);

    /**
     * Information available in an invocation of the callbacks defined in the IterationProgressListener* is determined for records ingested into the iteration body and for records emitted by operators within the
     * iteration body.
     */
    interface Context* {
If all inputs are bounded, the maximum epoch /**
of all records ingested into this operator is used *as Emitsthe
 a record to the side* output identifiedepochWatermark by the {@link OutputTag} last invocation of this callback.
     *
     *
 @param epochWatermark The incremented epoch watermark.
     * @param outputTagcontext the {@code OutputTag}A context that identifiesallows theemitting side output to emit to.
. The context is only valid during the invocation of
     *         *   @param value The record tothis emitmethod.
     * @param collector The collector for returning result */values.
     */
   <X> void output(OutputTag<X> outputTag, X value);onEpochWatermarkIncremented(int epochWatermark, Context context, Collector<T> collector);

    }
}

package org.apache.flink.ml.iteration;

/**
 * A helper class to apply {@link IterationBody} to data streams.
 */
@PublicEvolving
public class IterationUtils {
    /**the epoch of this emitted record = the
     * epoch of the input record that triggers this emission. If this record is emitted by onEpochLWIncremented(), then
     * the epoch of this record = incrementedEpochLW - 1.
     * 4)This Formethod anycan recorduse emittedan byiteration thisbody operatorto intoprocess arecords feedbackin variableunbounded stream, the epoch of the emitted record =data streams.
     *
     * min(theThis epochmethod ofinvokes the iteration inputbody recordwith thatthe triggersfollowing thisparameters:
 emission, MAX_LONG - 1) +* 1.) IfThe this1st recordparameter is a emittedlist by
of input variable streams, which * onEpochLWIncremented(), thenare created as the epochunion of thisthe record = incrementedEpochLW.initial variable
     *
 streams and the corresponding *feedback Thevariable execution of the graph createdstreams (returned by the iteration body will not terminate by itself. This is because at least
     * one of its data streams is unbounded).
     * 2) The 2nd parameter is the data streams given to this method.
     *
     * Required:
The epoch values are determined *as 1)described Allbelow. theSee initIterationListener variablefor streamshow must be boundedthe epoch values are used.
     * 21) ThereAll is at least one unbounded stream records in the datainitial variable streams listhas epoch=0.
     * 32) TheAll parallelism of any stream records in the data initialstreams variable streams must equal the parallelism ofhas epoch=MAX_LONG. In this case, records in the data stream atwon't theaffect
     * sameany indexoperator's of the feedback variable streams returned by the IterationBodylow watermark.
     *
 3) For any record *emitted @paramby initVariableStreamsthis Theoperator initialinto variable streams. These streams will be merged with the feedback variablea 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 onEpochLWIncremented(), then
     * the epoch of this record streams= beforeincrementedEpochLW being used as the 1st parameter to invoke the iteration body- 1.
     * 4) For @paramany dataStreamsrecord Theemitted databy streams.this Theseoperator streamsinto willa befeedback usedvariable asstream, the 2ndepoch parameterof tothe invokeemitted therecord iteration=
     * min(the epoch of the input record that triggers this emission, MAX_LONG - 1) + 1. If this record is emitted body.by
     * @param body The computation logic which takes variable/data streams and returns variable/output streams.onEpochLWIncremented(), then the epoch of this record = incrementedEpochLW.
     *
     * @return The listexecution of outputthe streamsgraph returnedcreated by the iteration boy.
     */ body will not terminate by itself. This is because at least
    static DataStreamList* iterateUnboundedStreams(DataStreamList initVariableStreams, DataStreamList dataStreams, IterationBody body) {...}

one of its data streams is unbounded.
     /**
     * ThisRequired:
 method can use an iteration* body1) toAll processthe records in some bounded data init variable streams iterativelymust untilbe abounded.
     * termination2) criteriaThere is reached (e.g. the given number of rounds is completed or no further variable update isat least one unbounded stream in the data streams list.
     * needed3). BecauseThe thisparallelism methodof doesany not replaystream records in the initial datavariable streams, must equal the iterationparallelism bodyof needsthe tostream cacheat thosethe
     * recordssame inindex orderof tothe visit those records repeatedlyfeedback variable streams returned by the IterationBody.
     *
     * This method invokes the iteration body @param initVariableStreams The initial variable streams. These streams will be merged with the followingfeedback parameters:variable
     *  1) The 1st parameter is a list of input variable streams, which are created as the union of the initial variable
     * streams before andbeing theused correspondingas feedbackthe variable1st streamsparameter (returnedto byinvoke the iteration body).
     * @param 2)dataStreams The 2nddata parameterstreams. isThese thestreams datawill streamsbe givenused toas thisthe method.
2nd parameter to invoke the *iteration
     * The epoch    values are determined as described below. See IterationListener for how the epoch values are usedbody.
     * @param 1)body AllThe recordscomputation inlogic thewhich initialtakes variable/data streams has epoch=0and returns variable/output streams.
     * 2) All records in the data streams has epoch=0 @return The list of output streams returned by the iteration boy.
     */
 3) For any recordstatic emittedDataStreamList by this operator into a non-feedback stream, the epoch of this emitted record = theiterateUnboundedStreams(DataStreamList initVariableStreams, DataStreamList dataStreams, IterationBody body) {...}

    /**
     * This epochmethod ofcan theuse inputan recorditeration thatbody triggersto thisprocess emission.records Ifin thissome recordbounded isdata emittedstreams by onEpochLWIncremented(), theniteratively until a
     * termination criteria is reached (e.g. the epochgiven number of this record = incrementedEpochLW - 1. rounds is completed or no further variable update is
     * 4needed). ForBecause anythis recordmethod emitteddoes bynot thisreplay operatorrecords intoin athe feedbackdata variable streamstreams, the epochiteration ofbody theneeds emittedto recordcache = thethose
     * epochrecords ofin theorder inputto recordvisit thatthose triggers this emission + 1. If this record is emitted by onEpochLWIncremented(),
     * then the epoch of this record = incrementedEpochLW.records repeatedly.
     *
     * This method invokes the iteration body with the following parameters:
     *
 1) The 1st parameter *is Supposea therelist isof ainput coordinatorvariable operatorstreams, which takesare allcreated feedbackas variablethe streams (emitted byunion of the iterationinitial body)variable
     * streams and the corresponding terminationfeedback criteriavariable streamstreams (ifreturned notby null)the asiteration inputs. The execution of the graph created by thebody).
     * iteration2) bodyThe will2nd terminateparameter whenis allthe inputdata streams havegiven beento fully AND any of the following conditions is met:this method.
     * consumed:
     * 1)The Theepoch terminationvalues criteriaare streamdetermined isas notdescribed nullbelow. AndSee theIterationListener coordinatorfor operatorhow hasthe notepoch observedvalues any new value fromare used.
     * the termination criteria stream between two consecutive onEpochLWIncremented invocations1) All records in the initial variable streams has epoch=0.
     * 2) The coordinator operatorAll records in the data streams has not observed any new value fromepoch=0.
     * 3) For any feedbackrecord variableemitted streamby betweenthis two
operator into a non-feedback stream, *the consecutiveepoch onEpochLWIncrementedof invocations.
this emitted record =  *the
     * Required:
epoch of the input record *that 1)triggers Allthis theemission. initIf variablethis streamsrecord andis theemitted data streams must be bounded.by onEpochLWIncremented(), then
     * 2) The parallelismthe epoch of anythis streamrecord in= theincrementedEpochLW initial- variable1.
 streams must equal the parallelism* of4) theFor streamany atrecord the
emitted by this operator into *a samefeedback indexvariable ofstream, the epoch feedbackof variablethe streamsemitted returnedrecord by= the IterationBody.
     *
 epoch of the input *record @paramthat initVariableStreamstriggers Thethis initialemission variable+ streams1. TheseIf streamsthis willrecord beis mergedemitted with the feedback variableby onEpochLWIncremented(),
     * then the epoch of this record = incrementedEpochLW.
     *
     * Suppose there is a coordinator operator which takes all streamsfeedback beforevariable beingstreams used as the 1st parameter to invoke(emitted by the iteration body.)
     * @param dataStreams The data streams. These streams will be used as the 2nd parameter to invoke the iterationand 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 AND any of the following conditions is  body.met:
     * consumed:
 @param body The computation logic* which1) takesThe variable/datatermination streamscriteria andstream returnsis variable/outputnot streamsnull.
 And the coordinator operator *has @returnnot Theobserved listany ofnew outputvalue streamsfrom
 returned by the iteration boy.
* the termination criteria stream */
between two consecutive onEpochLWIncremented staticinvocations.
 DataStreamList iterateBoundedStreamsUntilTermination(DataStreamList initVariableStreams, DataStreamList dataStreams,* IterationBody2) body) {...}

    /**
     * This method can use an iteration body to process records in some bounded data streams iteratively until aThe coordinator operator has not observed any new value from any feedback variable stream between two
     * consecutive onEpochLWIncremented invocations.
     *
     * terminationRequired:
 criteria is reached (e.g. the* given1) numberAll ofthe roundsinit variable isstreams completedand orthe nodata furtherstreams variablemust updatebe isbounded.
     * needed2). BecauseThe thisparallelism methodof replaysany recordsstream in the initial datavariable streams, themust iterationequal bodythe doesparallelism notof needthe tostream cacheat thosethe
     * records to visit those records repeatedlysame index of the feedback variable streams returned by the IterationBody.
     *
     * This method invokes the iteration body @param initVariableStreams The initial variable streams. These streams will be merged with the followingfeedback parameters:variable
     * 1) The 1st parameter is a list of input variable streams, which are  created as the union of the initial variable
     * streams before andbeing theused correspondingas feedbackthe variable1st streamsparameter (returnedto byinvoke the iteration body).
     * @param 2)dataStreams The 2nddata parameterstreams. isThese astreams listwill ofbe replayedused dataas streams,the which2nd areparameter createdto by replayinginvoke the initial data streamsiteration
     * round  by round until the iteration terminates. The records in the Nth round will be emitted into the iterationbody.
     * @param body onlyThe ifcomputation thelogic lowwhich watermarktakes ofvariable/data thestreams firstand operatorreturns in the iteration body >= N - 1.
     *
     * The epoch values are determined as described below. See IterationListener for how the epoch values are used.
     * 1) All records in the initial variable streams has epoch=0.
     * 2) The records from the initial data streams will be replayed round by round into the iteration body. The records
     * in the first round have epoch=0. And records in the Nth round have epoch = N - 1.
     * 3) For any record emitted by this operator into a non-feedback stream, the epoch of this emitted record = thevariable/output streams.
     * @return The list of output streams returned by the iteration boy.
     */
    static DataStreamList iterateBoundedStreamsUntilTermination(DataStreamList initVariableStreams, DataStreamList dataStreams, IterationBody body) {...}

    /**
     * This method can use an iteration body to process records in some bounded data streams iteratively until a
     * termination criteria is reached (e.g. the given number of rounds is completed or no further variable update is
     * needed). Because this method replays records in the data streams, the iteration body does not need to cache those
     * records to visit those records repeatedly.
     *
 epoch of the input record* thatThis triggersmethod thisinvokes emission.the Ifiteration thisbody recordwith isthe emittedfollowing by onEpochLWIncremented(), then
parameters:
     * 1) *The the1st epochparameter ofis thisa recordlist =of incrementedEpochLWinput - 1.
     * 4) For any record emitted by this operator into a feedback stream, the epoch of the emitted record = the epochvariable streams, which are created as the union of the initial variable
     * streams and the corresponding feedback variable streams (returned by the iteration body).
     * of2) theThe input2nd recordparameter thatis triggersa thislist emission + 1. If this record is emitted by onEpochLWIncremented(), thenof replayed data streams, which are created by replaying the initial data streams
     * theround epochby ofround thisuntil recordthe =iteration incrementedEpochLW.
     *
     * Suppose there is a coordinator operator which takes all feedback variable streams (emitted byterminates. The records in the Nth round will be emitted into the iteration
     * body only if the low watermark of the first operator in the iteration body)
 >= N - 1.
 * and the termination criteria*
 stream (if not null) as inputs. * The executionepoch ofvalues theare graphdetermined createdas bydescribed the
below. See IterationListener for how *the iterationepoch bodyvalues willare terminateused.
 when all input streams have* been1) fullyAll consumed AND any ofrecords in the followinginitial conditions
variable streams    * is met:has epoch=0.
     * 12) The terminationrecords from criteriathe streaminitial isdata notstreams null.will Andbe thereplayed coordinatorround operatorby hasround notinto observedthe anyiteration newbody. valueThe fromrecords
     * in the terminationfirst criteriaround stream between two consecutive onEpochLWIncremented invocations.
     * 2) The coordinator operator has not observed any new value from any feedback variable stream between two
     * consecutive onEpochLWIncremented invocations.
     *have epoch=0. And records in the Nth round have epoch = N - 1.
     * 3) For any record emitted by this operator into a non-feedback stream, the epoch of this emitted record = the
     * Required:
epoch of the input record *that 1)triggers Allthis theemission. initIf variablethis streamsrecord andis theemitted data streams must be bounded.by onEpochLWIncremented(), then
     * 2)the Theepoch parallelism of any stream in the initial variable streams must equal the parallelism of the stream at the
     * same indexthis record = incrementedEpochLW - 1.
     * 4) For any record emitted by this operator into a feedback stream, the epoch of the feedbackemitted variablerecord streams returned= by the IterationBody.epoch
     *
 of the input record *that @paramtriggers initVariableStreamsthis Theemission initial+ variable streams1. TheseIf streamsthis willrecord beis mergedemitted with the feedback variableby onEpochLWIncremented(), then
     * the epoch of this record = incrementedEpochLW.
     *
     * Suppose there is a coordinator operator which takes all feedback variable streams before being used as(emitted by the 1st parameter to invoke the iteration body.)
     * @param initDataStreams The initial data streams. Records from these streams will be repeatedly replayed and usedand 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
     * asis themet:
 2nd parameter to invoke the* iteration body.
     * @param body The computation logic which takes variable/data streams and returns variable/output streams1) 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 onEpochLWIncremented invocations.
     * @return2) The list of output streams returned by the iteration boy coordinator operator has not observed any new value from any feedback variable stream between two
     * consecutive onEpochLWIncremented invocations.
     */
    static DataStreamList iterateAndReplayBoundedStreamsUntilTermination(DataStreamList initVariableStreams, DataStreamList initDataStreams, IterationBody body) {...}

}

package org.apache.flink.ml.iteration;

public class DataStreamList {
    */
// Returns the number of data streams in this list.
     Optional<DataStream<?>> terminationCriteria;
}