/** This class take response for collecting data to output stream. */
public interface Collector<OUT> {
    /**
     * Collect record to output stream.
     *
     * @param record to be collected.
     */
    void collect(OUT record);      
    
    /**
     * CollectOverwrite recordthe totimestamp output of this record and collect it to output stream.
     *
     * @param record to be collected.
     * @param timestamp of the processed data.
     */
    void collectcollectAndOverwriteTimestamp(OUT record, long timestamp); 
}

/** This class contains all logical related to process and emit records tofrom two outputsnon-broadcast input. */
public interface TwoOutputStreamProcessFunction<INTwoInputNonBroadcastStreamProcessFunction<IN1, OUT1IN2, OUT2>OUT> extends ProcessFunction {
    /**
     * Process andrecord emitfrom recordfirst toinput theand first/secondemit outputdata through {@link Collector}s.
     *
     * @param record to process.
     * @param output1output to emit processed records.
 to the first output.
     * @param output2 to emit processed records to the second output.
     * @param ctx, runtime context in which this function is executed.
     */
    void processRecordprocessRecordFromFirstInput(
IN1 record, Collector<OUT> output, RuntimeContext ctx)
       IN record, Collector<OUT1> output1, Collector<OUT2> output2, RuntimeContext ctx)throws Exception;

    /**
     * Process Thisrecord isfrom asecond life-cycleinput methodand indicatesemit thatdata thisthrough function will no longer receive any input{@link Collector}.
     *
     * data.@param Thisrecord allowing the ProcessFunction to emit results at once rather than upon each recordto process.
     *
     * @param output1 to emit record to the first output.
     * @param output2 to emit record to the second outputprocessed records.
     * @param ctx, runtime context in which this function is executed.
     */
    default void endInputprocessRecordFromSecondInput(
IN2 record, Collector<OUT> output, RuntimeContext ctx)
       Collector<OUT1> output1, Collector<OUT2> output2, RuntimeContext ctx) {}
}

throws Exception;

    /**
     * This 

is 

a 

life-cycle 

method 

indicates 

that 

this 

function 

will 

no 

longer 

receive 

any 

data 

from
 

* 

the 

first 

input.
 

 *

  * @param ctx, the context 

in which this function is executed.
     */
 

default 

void 

endFirstInput(NonPartitionedContext<OUT> ctx) {

}

/**
     * 

This 

is 

a 

life-cycle method indicates that this function will no longer receive any data from
     *

 the second input.
 

  *
     * @param ctx, the context 

in 

which this function is executed.
     */
 

default 

void 

endSecondInput(NonPartitionedContext<OUT> ctx) {

}

}

all 

logical 

related to 

process 

records 

from 

a broadcast stream and a
 * 

non-broadcast stream

.
 

*/
public interface TwoInputBroadcastStreamProcessFunction<IN1, IN2, 

OUT> 

extends 

ProcessFunction {
    

/**
     * 

Process 

record 

from 

non-broadcast input 

and 

emit 

data 

through 

{@link 

Collector}.
     *
     * @param 

record to 

process.
     * @param 

output to emit 

processed 

records.
     * 

@param ctx, runtime context in which this 

function is executed.
     */
    

void processRecordFromNonBroadcastInput(IN1 record, Collector<OUT> output, RuntimeContext ctx)
            

throws Exception;

    /**
     * 

Process record 

from 

broadcast input

.In general, the broadcast side is not allowed to
     *

 manipulate state and output data 

because 

it 

corresponds to 

all 

partitions instead of a single
 

* 

partition. But you could use broadcast context to process all the partitions at once.
     *
     * @param record 

to process.
     * @param ctx, the context in which 

this function is executed.
     */

    void 

processRecordFromBroadcastInput(IN2 record, NonPartitionedContext<OUT> ctx) throws Exception;

    

/**
     * 

This 

is 

a 

life-cycle method indicates that this function will no longer receive any data from
     * 

the 

non-broadcast input.
     *
     * @param 

ctx, 

the 

context 

in 

which 

this 

function 

is 

executed.
     */
  

default 

void 

endNonBroadcastInput(NonPartitionedContext<OUT> ctx) {}

    

/**

* 

This is a life-cycle method indicates that this function will no longer receive any data from
     * 

the 

broadcast 

input.
     *
     * @param 

ctx, the

 context in which this function is executed.
     */
    

default void 

endBroadcastInput(NonPartitionedContext<OUT> ctx)

 {}
}

 Process and emit record to the first/second output through {@link 

Collector}s.
     *

 * @param record to process.
     * 

@param output1 to emit processed records to the first output.
     * @param output2 to emit processed records to the second output.
     * @param 

ctx, 

runtime 

context 

in 

which this 

function is executed.
     */
    void 

processRecord(

IN 

record, 

Collector<OUT1> 

output1, 

Collector<OUT2> 

output2, RuntimeContext ctx);

   /**
     * 

This is 

a 

life-cycle 

method 

indicates 

that 

this 

function 

will no longer receive any 

input
    

* data.
     

*

 *

 @param ctx, the context in which this function 

is executed.
     */
    default void endInput(TwoOutputNonPartitionedContext<OUT1, OUT2> 

ctx) {}
}

each 

parallelism 

is a
 * partition, and the partition to which the data belongs is 

random.
 */
public interface 

NonKeyedPartitionStream<T> {
    /**
     * Apply an operation to this {@link 

NonKeyedPartitionStream}

;
     *
     * 

@param 

processFunction 

to 

perform operation.
     * @return new stream with this operation.
     */
    

<OUT> NonKeyedPartitionStream<OUT> process(
            OneInputStreamProcessFunction<T, OUT> processFunction);

    /**
     * 

Apply a two output operation to this {@link 

NonKeyedPartitionStream}.
     *
     * @param processFunction to perform two output operation.
     * 

@return 

new 

stream 

with 

this operation.
     *

/
    <OUT1, 

OUT2> TwoNonKeyedPartitionStreams<OUT1, 

OUT2> process(
            

TwoOutputStreamProcessFunction<T, 

OUT1,

OUT2> processFunction);

    /**
     * Apply 

to a two input operation 

on this and other {@link 

NonKeyedPartitionStream}

.
     *
     * @param

 other {@link NonKeyedPartitionStream} to perform operation with two input.
     * 

@param 

processFunction 

to perform operation.
     * @return new stream with this operation.
     */
    

<T_OTHER, OUT> NonKeyedPartitionStream<OUT> 

connectAndProcess(
            NonKeyedPartitionStream<T_OTHER> other,
            

TwoInputNonBroadcastStreamProcessFunction<T, T_OTHER, OUT> processFunction);

    /**
     * Apply a two 

input operation to this and other {@link 

BroadcastStream}.
     *
     * @param 

processFunction 

to 

perform 

operation.
 

* 

@return 

new 

stream 

with 

this 

operation.
     */
 

 <T_OTHER, OUT> NonKeyedPartitionStream<OUT> connectAndProcess(
            BroadcastStream<T_OTHER> other,
      

TwoInputBroadcastStreamProcessFunction<T, T_OTHER, OUT> processFunction);

    

/**
     * 

Coalesce this 

stream to 

a 

{@link GlobalStream}.
     *
   

* 

@return the 

coalesced 

global 

stream.
     */
 

GlobalStream<T> global();

    /**
     * 

Transform this stream to a {@link 

KeyedPartitionStream}.
     *

* 

@param 

keySelector 

to 

decide how to map data 

to partition.
     * @return the transformed stream partitioned 

by key.
     */
    <K> KeyedPartitionStream<K, T> 

keyBy(KeySelector<T, K> 

keySelector);

    /**
     * 

Transform 

this 

stream 

to 

a 

new {@link 

NonKeyedPartitionStream}

, data will be shuffled 

between
     * 

these two streams.
     *
     * @return 

the 

transformed 

stream 

after 

shuffle.
     */
    

NonKeyedPartitionStream<T> shuffle();

    /**
     * 

Transform 

this stream to 

a 

new 

{@link 

BroadcastStream}.
 

  *
   

* 

@return the 

transformed 

{@link BroadcastStream}.
     */
    BroadcastStream<T> 

broadcast();

    

/**
 

* 

Sink 

data 

from 

this 

stream.
     *

     * 

@param sink 

to 

receive 

data 

from this 

stream.
     */
    

void toSink(Sink<T> sink);

    /**
     * This class 

represents a combination of two {@link NonKeyedPartitionStream}. It will be used
     

* as the return value of operation with two output.
     */
 

interface 

TwoNonKeyedPartitionStreams<T1, 

T2> 

{
 

  /** Get the 

first 

stream. */
        NonKeyedPartitionStream<T1> getFirst();

        /**

 Get the second stream. */
  

NonKeyedPartitionStream<T2> getSecond();
    }
}

 Apply an operation to 

this 

{@link KeyedPartitionStream}.
 

*
     * 

<p>This 

method 

is 

used 

to avoid shuffle after applying 

the 

process 

function. 

It 

is required
     * 

that for the same record, the new {@link 

KeySelector} 

must extract the same key as the
     *

 original {@link 

KeySelector} on this {@link KeyedPartitionStream}.
     *
     * @param processFunction 

to perform operation.
     * @param newKeySelector to select the key 

after process.
     * @return new {@link KeyedPartitionStream} with this operation.
  

   */
    <OUT> KeyedPartitionStream<K, OUT> process(
            OneInputStreamProcessFunction<T, OUT> processFunction,
            KeySelector<OUT, K> newKeySelector);

    /**
     * Apply 

an 

operation to this 

{@link 

KeyedPartitionStream}

;
     *
     * @param processFunction to perform operation.
     * @return new 

{@link NonKeyedPartitionStream} with this operation.
     */
    

<OUT> NonKeyedPartitionStream<OUT> 

process(
            

OneInputStreamProcessFunction<T, OUT> processFunction);

    /**
     * Apply a 

two output operation to this {@link KeyedPartitionStream}.
     *
     * 

<p>This 

method 

is 

used 

to avoid shuffle after applying 

the process function. It 

is required
     

*

 that for the same 

record, 

these 

new two {@link 

KeySelector}

s must extract the same 

key as the
     *

 original {@link KeySelector}s on this {@link KeyedPartitionStream}.
     *
     * 

@param 

processFunction 

to 

perform 

two 

output 

operation.
     *

 @param keySelector1 to 

select the key of first output.
     

*

 @param keySelector2 to select 

the 

key 

of 

second 

output.
 

 * @return new {@link TwoKeyedPartitionStreams} with this operation.
     *

/
    <OUT1, OUT2> TwoKeyedPartitionStreams<K, OUT1, OUT2> process(
       

     TwoOutputStreamProcessFunction<T, OUT1, OUT2> processFunction,
     

KeySelector<OUT1, 

K> 

keySelector1,
         

   KeySelector<OUT2, 

K> 

keySelector2);

    /**
     * Apply a 

two 

output 

operation to 

this {@link 

KeyedPartitionStream}.
     *
     * 

@param 

processFunction 

to 

perform two output operation.
    

*

 @return new {@link TwoNonKeyedPartitionStreams} 

with this 

operation.
     */
    <OUT1, 

OUT2> 

TwoNonKeyedPartitionStreams<OUT1, 

OUT2> 

process(
 

  TwoOutputStreamProcessFunction<T, OUT1, 

OUT2> processFunction);

    /**
     * 

Apply 

a 

two 

input 

operation 

to 

this 

and other {@link KeyedPartitionStream}. The 

two 

keyed
 

  * streams must 

have the 

same partitions, otherwise it makes no sense to connect them.
     *

* 

@param 

other {

@link KeyedPartitionStream} to perform operation with two input.
     

*

 @param 

processFunction 

to 

perform 

operation.

     * @return new 

{@link NonKeyedPartitionStream} with this operation.
     

*/
    <T_OTHER, OUT> NonKeyedPartitionStream<OUT> connectAndProcess(
            KeyedPartitionStream<K, 

T_OTHER> other,
    

        TwoInputNonBroadcastStreamProcessFunction<T, T_OTHER, OUT> processFunction);

     /**
     * Apply a two 

input operation to this and other {@link 

KeyedPartitionStream}.The two keyed
     *

 streams must have the 

same 

partitions, otherwise it makes no sense to 

connect 

them.
     *
 

* 

<p>This 

method 

is 

used to avoid shuffle after 

applying the process function. 

It 

is 

required
     * that for the same record, 

the new {@link KeySelector} must extract the same key as the
     * 

original 

{@link KeySelector}s on these two {@link 

KeyedPartitionStream}s.
     *
     * @param 

other {@link KeyedPartitionStream} to perform operation with two 

input.
     * 

@param 

processFunction 

to 

perform operation.
     *

 @param newKeySelector to 

select 

the 

key 

after process

.
     * @return new {@link KeyedPartitionStream} with this 

operation.
     */
    

<T_OTHER, OUT> KeyedPartitionStream<K, OUT> connectAndProcess(
            KeyedPartitionStream<K, T_OTHER> other,
     

       TwoInputNonBroadcastStreamProcessFunction<T, T_OTHER, OUT> processFunction,
            KeySelector<OUT, K> newKeySelector);

    /**
     * Apply a two input operation to this and other {@link 

BroadcastStream}

.
 

*
     * @param processFunction to perform operation.
     * @return new stream with this operation.
     */
    <T_OTHER, OUT> 

NonKeyedPartitionStream<OUT> connectAndProcess(
            

BroadcastStream<T_OTHER> other,
            

TwoInputBroadcastStreamProcessFunction<T, T_OTHER, OUT> processFunction);

    /**
     * Apply a 

two 

input 

operation to 

this and other {@link BroadcastStream}.
     *
     * 

<p>This 

method 

is 

used to 

avoid 

shuffle 

after applying the process function. It is required
     *

 that for the 

record 

from non-broadcast input, the new {@link KeySelector} must extract the
     * 

same 

key 

as 

the 

original 

{@link 

KeySelector}

s 

on 

the {@link KeyedPartitionStream}. For the
     * 

record 

from 

broadcast input, the output key 

from keyed partition itself instead 

of 

new key
     *

 selector, so it 

is safe already.
     

*
     * 

@param 

other {@link BroadcastStream}

 to perform operation with two 

input.
     * 

@param 

processFunction 

to 

perform 

operation.
     *

 @param newKeySelector to 

select the key after process.
     * 

@return new {@link KeyedPartitionStream} with this 

operation.
     */
    <T_OTHER, 

OUT> 

KeyedPartitionStream<K, 

OUT> 

connectAndProcess(
 

  BroadcastStream<T_OTHER> other,
 

TwoInputBroadcastStreamProcessFunction<T, T_OTHER, OUT> processFunction,
            KeySelector<OUT, K> newKeySelector);

    /**
     * Coalesce this stream to a {@link GlobalStream}.
    

 *
     

* 

@return the 

coalesced global stream.
 

    */
    

GlobalStream<T> 

global();

    /**
    

 *

Transform 

this stream

to a new {@link KeyedPartitionStream}.
    

*
     

* @param keySelector to decide how to map data to partition.
     * @return the transformed stream partitioned by key.
     */
 

<NEW_KEY> KeyedPartitionStream<NEW_KEY, T> keyBy(KeySelector<T, NEW_KEY> keySelector);

    

/**
     * 

Transform 

this 

stream 

to 

a 

new {@link NonKeyedPartitionStream}, data will be shuffled between
     * 

these two streams.
     *
     * @return 

the transformed stream 

after 

shuffle.
     */
    

NonKeyedPartitionStream<T> shuffle();

    /**
     * Transform this stream to a new 

{@link BroadcastStream}.
     *
     * @return 

the transformed {@link BroadcastStream}.
     */

    BroadcastStream<T> broadcast();

    /**
 

* 

Sink 

data 

from 

this stream.
     *
     * @param 

sink to 

receive 

data 

from 

this 

stream.
     */
 

void 

toSink(Sink<T> sink);

    /**
 

* 

This 

class represents a combination of 

two {@link KeyedPartitionStream}. It 

will be used as
     * the return value of operation 

with two output.
     */
    interface TwoKeyedPartitionStreams<K, T1, T2> {
        /** Get the 

first stream. */
        KeyedPartitionStream<K, T1> getFirst();

        /** Get the second stream. */
        KeyedPartitionStream<K, T2> getSecond();
    }
}