pnowojski commented on a change in pull request #13234:
URL: https://github.com/apache/flink/pull/13234#discussion_r478513685
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java
##########
@@ -547,302 +675,38 @@ private void
linkOperatorWrappers(List<StreamOperatorWrapper<?, ?>> allOperatorW
return (tailOperatorWrapper == null) ? null :
tailOperatorWrapper.getStreamOperator();
}
- //
------------------------------------------------------------------------
- // Collectors for output chaining
- //
------------------------------------------------------------------------
-
/**
- * An {@link Output} that measures the last emitted watermark with a
{@link WatermarkGauge}.
- *
- * @param <T> The type of the elements that can be emitted.
+ * Wrapper class to access the chained sources and their's outputs.
*/
- public interface WatermarkGaugeExposingOutput<T> extends Output<T> {
- Gauge<Long> getWatermarkGauge();
- }
-
- static class ChainingOutput<T> implements
WatermarkGaugeExposingOutput<StreamRecord<T>> {
-
- protected final OneInputStreamOperator<T, ?> operator;
- protected final Counter numRecordsIn;
- protected final WatermarkGauge watermarkGauge = new
WatermarkGauge();
-
- protected final StreamStatusProvider streamStatusProvider;
-
- @Nullable
- protected final OutputTag<T> outputTag;
-
- public ChainingOutput(
- OneInputStreamOperator<T, ?> operator,
- StreamStatusProvider streamStatusProvider,
- @Nullable OutputTag<T> outputTag) {
- this.operator = operator;
-
- {
- Counter tmpNumRecordsIn;
- try {
- OperatorIOMetricGroup ioMetricGroup =
((OperatorMetricGroup) operator.getMetricGroup()).getIOMetricGroup();
- tmpNumRecordsIn =
ioMetricGroup.getNumRecordsInCounter();
- } catch (Exception e) {
- LOG.warn("An exception occurred during
the metrics setup.", e);
- tmpNumRecordsIn = new SimpleCounter();
- }
- numRecordsIn = tmpNumRecordsIn;
- }
+ public static class ChainedSourceOutputs {
+ private final Map<Integer,
WatermarkGaugeExposingOutput<StreamRecord<?>>> chainedSourceOutputs;
+ private final Map<Integer, SourceOperator<?, ?>>
sourceOperators;
Review comment:
Good idea with `Input` - I also didn't like the integer. About moving
`Map` to an upper level, I don't mind one way or another, so let it be as you
prefer.
----------------------------------------------------------------
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]
