IN1
- The input type of the operatorIN2
- The input type of the operatorOUT
- The output type of the operator@PublicEvolving public interface TwoInputStreamOperator<IN1,IN2,OUT> extends StreamOperator<OUT>
AbstractStreamOperator
as a base class if you want to
implement a custom operator.Modifier and Type | Method and Description |
---|---|
void |
processElement1(StreamRecord<IN1> element)
Processes one element that arrived on the first input of this two-input operator.
|
void |
processElement2(StreamRecord<IN2> element)
Processes one element that arrived on the second input of this two-input operator.
|
void |
processLatencyMarker1(LatencyMarker latencyMarker)
Processes a
LatencyMarker that arrived on the first input of this two-input operator. |
void |
processLatencyMarker2(LatencyMarker latencyMarker)
Processes a
LatencyMarker that arrived on the second input of this two-input
operator. |
default void |
processRecordAttributes1(RecordAttributes recordAttributes)
Processes a
RecordAttributes that arrived on the first input of this operator. |
default void |
processRecordAttributes2(RecordAttributes recordAttributes)
Processes a
RecordAttributes that arrived on the second input of this operator. |
void |
processWatermark1(Watermark mark)
Processes a
Watermark that arrived on the first input of this two-input operator. |
void |
processWatermark2(Watermark mark)
Processes a
Watermark that arrived on the second input of this two-input operator. |
void |
processWatermarkStatus1(WatermarkStatus watermarkStatus)
Processes a
WatermarkStatus that arrived on the first input of this two-input
operator. |
void |
processWatermarkStatus2(WatermarkStatus watermarkStatus)
Processes a
WatermarkStatus that arrived on the second input of this two-input
operator. |
close, finish, getMetricGroup, getOperatorAttributes, getOperatorID, initializeState, open, prepareSnapshotPreBarrier, setKeyContextElement1, setKeyContextElement2, snapshotState
notifyCheckpointAborted, notifyCheckpointComplete
getCurrentKey, setCurrentKey
void processElement1(StreamRecord<IN1> element) throws Exception
Exception
void processElement2(StreamRecord<IN2> element) throws Exception
Exception
void processWatermark1(Watermark mark) throws Exception
Watermark
that arrived on the first input of this two-input operator.
This method is guaranteed to not be called concurrently with other methods of the operator.void processWatermark2(Watermark mark) throws Exception
Watermark
that arrived on the second input of this two-input operator.
This method is guaranteed to not be called concurrently with other methods of the operator.void processLatencyMarker1(LatencyMarker latencyMarker) throws Exception
LatencyMarker
that arrived on the first input of this two-input operator.
This method is guaranteed to not be called concurrently with other methods of the operator.Exception
LatencyMarker
void processLatencyMarker2(LatencyMarker latencyMarker) throws Exception
LatencyMarker
that arrived on the second input of this two-input
operator. This method is guaranteed to not be called concurrently with other methods of the
operator.Exception
LatencyMarker
void processWatermarkStatus1(WatermarkStatus watermarkStatus) throws Exception
WatermarkStatus
that arrived on the first input of this two-input
operator. This method is guaranteed to not be called concurrently with other methods of the
operator.Exception
WatermarkStatus
void processWatermarkStatus2(WatermarkStatus watermarkStatus) throws Exception
WatermarkStatus
that arrived on the second input of this two-input
operator. This method is guaranteed to not be called concurrently with other methods of the
operator.Exception
WatermarkStatus
@Experimental default void processRecordAttributes1(RecordAttributes recordAttributes) throws Exception
RecordAttributes
that arrived on the first input of this operator. This
method is guaranteed to not be called concurrently with other methods of the operator.Exception
@Experimental default void processRecordAttributes2(RecordAttributes recordAttributes) throws Exception
RecordAttributes
that arrived on the second input of this operator. This
method is guaranteed to not be called concurrently with other methods of the operator.Exception
Copyright © 2014–2024 The Apache Software Foundation. All rights reserved.