dawidwys commented on a change in pull request #13529:
URL: https://github.com/apache/flink/pull/13529#discussion_r501079844
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTwoInputProcessorFactory.java
##########
@@ -40,38 +48,78 @@
*/
public class StreamTwoInputProcessorFactory {
public static <IN1, IN2> StreamTwoInputProcessor<IN1, IN2> create(
+ AbstractInvokable ownerTask,
CheckpointedInputGate[] checkpointedInputGates,
- TypeSerializer<IN1> inputSerializer1,
- TypeSerializer<IN2> inputSerializer2,
IOManager ioManager,
+ MemoryManager memoryManager,
TaskIOMetricGroup taskIOMetricGroup,
StreamStatusMaintainer streamStatusMaintainer,
TwoInputStreamOperator<IN1, IN2, ?> streamOperator,
TwoInputSelectionHandler inputSelectionHandler,
WatermarkGauge input1WatermarkGauge,
WatermarkGauge input2WatermarkGauge,
OperatorChain<?, ?> operatorChain,
+ StreamConfig streamConfig,
+ Configuration taskManagerConfig,
+ Configuration jobConfig,
+ ExecutionConfig executionConfig,
+ ClassLoader userClassloader,
Counter numRecordsIn) {
checkNotNull(operatorChain);
checkNotNull(inputSelectionHandler);
+
StreamStatusTracker statusTracker = new StreamStatusTracker();
taskIOMetricGroup.reuseRecordsInputCounter(numRecordsIn);
+ TypeSerializer<IN1> typeSerializer1 =
streamConfig.getTypeSerializerIn(0, userClassloader);
+ StreamTaskInput<IN1> input1 = new StreamTaskNetworkInput<>(
+ checkpointedInputGates[0],
+ typeSerializer1,
+ ioManager,
+ new
StatusWatermarkValve(checkpointedInputGates[0].getNumberOfInputChannels()),
+ 0);
+ TypeSerializer<IN2> typeSerializer2 =
streamConfig.getTypeSerializerIn(1, userClassloader);
+ StreamTaskInput<IN2> input2 = new StreamTaskNetworkInput<>(
+ checkpointedInputGates[1],
+ typeSerializer2,
+ ioManager,
+ new
StatusWatermarkValve(checkpointedInputGates[1].getNumberOfInputChannels()),
+ 1);
+
+ if (streamConfig.shouldSortInputs()) {
+ @SuppressWarnings("unchecked")
+ MultiInputSortingDataInputs<Object> multiInputs = new
MultiInputSortingDataInputs<Object>(
Review comment:
Yes, I must admit I was a bit harsh on this one :sweat_smile:
----------------------------------------------------------------
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]
