pnowojski commented on code in PR #22432:
URL: https://github.com/apache/flink/pull/22432#discussion_r1177575706
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/BroadcastingOutputCollector.java:
##########
@@ -17,52 +17,77 @@
package org.apache.flink.streaming.runtime.tasks;
+import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.Gauge;
import org.apache.flink.streaming.api.operators.Output;
import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.io.RecordWriterOutput;
import org.apache.flink.streaming.runtime.metrics.WatermarkGauge;
import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
import org.apache.flink.util.OutputTag;
import org.apache.flink.util.XORShiftRandom;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Random;
class BroadcastingOutputCollector<T> implements
WatermarkGaugeExposingOutput<StreamRecord<T>> {
- protected final Output<StreamRecord<T>>[] outputs;
+ protected final Output<StreamRecord<T>>[] allOutputs;
+
+ protected final Output<StreamRecord<T>>[] chainedOutputs;
+
+ protected final RecordWriterOutput<T>[] nonChainedOutputs;
+
+ protected final Counter numRecordsOutForTask;
private final Random random = new XORShiftRandom();
private final WatermarkGauge watermarkGauge = new WatermarkGauge();
- public BroadcastingOutputCollector(Output<StreamRecord<T>>[] outputs) {
- this.outputs = outputs;
+ @SuppressWarnings({"unchecked"})
+ public BroadcastingOutputCollector(
+ Output<StreamRecord<T>>[] allOutputs, Counter
numRecordsOutForTask) {
+ this.allOutputs = allOutputs;
+ this.numRecordsOutForTask = numRecordsOutForTask;
+
+ List<Output<StreamRecord<T>>> chainedOutputs = new ArrayList<>(4);
+ List<RecordWriterOutput<T>> nonChainedOutputs = new ArrayList<>(4);
+ for (Output<StreamRecord<T>> output : allOutputs) {
+ if (output instanceof RecordWriterOutput) {
+ nonChainedOutputs.add((RecordWriterOutput<T>) output);
+ } else {
+ chainedOutputs.add(output);
+ }
+ }
+ this.chainedOutputs = chainedOutputs.toArray(new Output[0]);
+ this.nonChainedOutputs = nonChainedOutputs.toArray(new
RecordWriterOutput[0]);
Review Comment:
I see, maybe instead of using `if/else` check, we could use some object
oriented wrapper? Something like or `OutputWithRecordsEmittedCheck`, that would
be an `@Internal` interface. Implementation could be:
```
class ChainingOutput implements OutputWithRecordsEmittedCheck {
boolean collectAndCheckIfEmitted(...) {
this.underlyingOutput.collect(...);
return false;
}
}
```
Using this interface, we could implement it in the existing internal
implementations/wrappers (to avoid adding another layer), like `ChainingOutput`.
We could also avoid `RecordWriterCountingOutput` wrapper, by adding the
records emitted counter directly to the `RecordWriterOutput`.
Maybe those changes wouldn't be too complicated? WDYT? If so, it might make
sense trying to split this PR into at least two smaller commits? For example
the first commit could be introducing `OutputWithRecordsEmittedCheck` and it's
implementations, and the actual bug fix using `collectAndCheckIfEmitted` in the
following commit?
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
