zhijiangW commented on a change in pull request #9478: [FLINK-13766][task]
Refactor the implementation of StreamInputProcessor based on
StreamTaskInput#emitNext
URL: https://github.com/apache/flink/pull/9478#discussion_r323275935
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTwoInputProcessor.java
##########
@@ -54,92 +50,98 @@
@Internal
public final class StreamTwoInputProcessor<IN1, IN2> implements
StreamInputProcessor {
- private final TwoInputStreamOperator<IN1, IN2, ?> streamOperator;
-
private final TwoInputSelectionHandler inputSelectionHandler;
private final Object lock;
- private final StreamTaskInput input1;
- private final StreamTaskInput input2;
+ private final StreamTaskInput<IN1> input1;
+ private final StreamTaskInput<IN2> input2;
private final OperatorChain<?, ?> operatorChain;
- /**
- * Valves that control how watermarks and stream statuses from the 2
inputs are forwarded.
- */
- private final StatusWatermarkValve statusWatermarkValve1;
- private final StatusWatermarkValve statusWatermarkValve2;
+ private final DataOutput<IN1> output1;
+ private final DataOutput<IN2> output2;
/**
* Stream status for the two inputs. We need to keep track for
determining when
* to forward stream status changes downstream.
*/
- private StreamStatus firstStatus;
- private StreamStatus secondStatus;
+ private StreamStatus firstStatus = StreamStatus.ACTIVE;
+ private StreamStatus secondStatus = StreamStatus.ACTIVE;
- private int lastReadInputIndex;
-
- private final Counter numRecordsIn;
+ /** Always try to read from the first input. */
+ private int lastReadInputIndex = 1;
private boolean isPrepared;
public StreamTwoInputProcessor(
- Collection<InputGate> inputGates1,
- Collection<InputGate> inputGates2,
- TypeSerializer<IN1> inputSerializer1,
- TypeSerializer<IN2> inputSerializer2,
- StreamTask<?, ?> streamTask,
- CheckpointingMode checkpointingMode,
- Object lock,
- IOManager ioManager,
- Configuration taskManagerConfig,
- StreamStatusMaintainer streamStatusMaintainer,
- TwoInputStreamOperator<IN1, IN2, ?> streamOperator,
- TwoInputSelectionHandler inputSelectionHandler,
- WatermarkGauge input1WatermarkGauge,
- WatermarkGauge input2WatermarkGauge,
- String taskName,
- OperatorChain<?, ?> operatorChain,
- Counter numRecordsIn) throws IOException {
-
- this.streamOperator = checkNotNull(streamOperator);
- this.inputSelectionHandler =
checkNotNull(inputSelectionHandler);
+ CheckpointedInputGate[] checkpointedInputGates,
+ TypeSerializer<IN1> inputSerializer1,
+ TypeSerializer<IN2> inputSerializer2,
+ Object lock,
+ IOManager ioManager,
+ StreamStatusMaintainer streamStatusMaintainer,
+ TwoInputStreamOperator<IN1, IN2, ?> streamOperator,
+ TwoInputSelectionHandler inputSelectionHandler,
+ WatermarkGauge input1WatermarkGauge,
+ WatermarkGauge input2WatermarkGauge,
+ OperatorChain<?, ?> operatorChain,
+ Counter numRecordsIn) {
this.lock = checkNotNull(lock);
+ this.inputSelectionHandler =
checkNotNull(inputSelectionHandler);
- InputGate unionedInputGate1 =
InputGateUtil.createInputGate(inputGates1.toArray(new InputGate[0]));
- InputGate unionedInputGate2 =
InputGateUtil.createInputGate(inputGates2.toArray(new InputGate[0]));
-
- // create a Input instance for each input
- CheckpointedInputGate[] checkpointedInputGates =
InputProcessorUtil.createCheckpointedInputGatePair(
- streamTask,
- checkpointingMode,
- ioManager,
- unionedInputGate1,
- unionedInputGate2,
- taskManagerConfig,
- taskName);
- checkState(checkpointedInputGates.length == 2);
- this.input1 = new
StreamTaskNetworkInput(checkpointedInputGates[0], inputSerializer1, ioManager,
0);
- this.input2 = new
StreamTaskNetworkInput(checkpointedInputGates[1], inputSerializer2, ioManager,
1);
-
- this.statusWatermarkValve1 = new StatusWatermarkValve(
- unionedInputGate1.getNumberOfInputChannels(),
- new ForwardingValveOutputHandler(streamOperator, lock,
streamStatusMaintainer, input1WatermarkGauge, 0));
- this.statusWatermarkValve2 = new StatusWatermarkValve(
- unionedInputGate2.getNumberOfInputChannels(),
- new ForwardingValveOutputHandler(streamOperator, lock,
streamStatusMaintainer, input2WatermarkGauge, 1));
+ this.output1 = new StreamTaskNetworkOutput<>(
+ streamOperator,
+ record -> processRecord1(record, streamOperator,
numRecordsIn),
+ lock,
+ streamStatusMaintainer,
+ input1WatermarkGauge,
+ 0);
+
+ this.output2 = new StreamTaskNetworkOutput<>(
+ streamOperator,
+ record -> processRecord2(record, streamOperator,
numRecordsIn),
+ lock,
+ streamStatusMaintainer,
+ input2WatermarkGauge,
+ 1);
+
+ StatusWatermarkValve statusWatermarkValve1 = new
StatusWatermarkValve(
+ checkpointedInputGates[0].getNumberOfInputChannels(),
output1);
+ StatusWatermarkValve statusWatermarkValve2 = new
StatusWatermarkValve(
+ checkpointedInputGates[1].getNumberOfInputChannels(),
output2);
+
+ this.input1 = new StreamTaskNetworkInput<>(
+ checkpointedInputGates[0], inputSerializer1, ioManager,
statusWatermarkValve1, 0);
+ this.input2 = new StreamTaskNetworkInput<>(
+ checkpointedInputGates[1], inputSerializer2, ioManager,
statusWatermarkValve2, 1);
Review comment:
Thanks. In my previous memories, it was both ok for arguments either in the
same line or in separate line. Now the new formatting seems only accept the
separate argument line if breaking. I should catch up with it and fix based on
that. :)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
