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_r323269375
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/streamstatus/StatusWatermarkValve.java
##########
@@ -29,24 +30,13 @@
/**
* A {@code StatusWatermarkValve} embodies the logic of how {@link Watermark}
and {@link StreamStatus} are propagated to
* downstream outputs, given a set of one or multiple input channels that
continuously receive them. Usages of this
- * class need to define the number of input channels that the valve needs to
handle, as well as provide a customized
- * implementation of {@link ValveOutputHandler}, which is called by the valve
only when it determines a new watermark or
- * stream status can be propagated.
+ * class need to define the number of input channels that the valve needs to
handle, as well as provide a implementation of
+ * {@link DataOutput}, which is called by the valve only when it determines a
new watermark or stream status can be propagated.
*/
@Internal
public class StatusWatermarkValve {
- /**
- * Usages of {@code StatusWatermarkValve} should implement a {@code
ValveOutputHandler}
- * to handle watermark and stream status outputs from the valve.
- */
- public interface ValveOutputHandler {
- void handleWatermark(Watermark watermark) throws Exception;
-
- void handleStreamStatus(StreamStatus streamStatus) throws
Exception;
- }
-
- private final ValveOutputHandler outputHandler;
+ private final DataOutput output;
Review comment:
It can not directly work well to make `StatusWatermarkValve` as a private
field of `output`.
The overall process is `StreamTaskNetworkInput#processElement` -> `if
(recordOrMark.isWatermark())` -> `output.emitWatermark` ->
StatusWatermarkValve#inputWatermark`.
The key problem is that the interface of `output.emitWatermark` can not
expose any channel index info, but `StatusWatermarkValve` has to know the
current channel index to judge the internal logics. So we make the
`StatusWatermarkValve` in the upper layer of `output`. In other words, the
`StatusWatermarkValve` can wrap the `output`, but `output` can not wrap the
`StatusWatermarkValve`.
As for passing the `output` in the constructor of `StatusWatermarkValve` vs
passing it in four internal methods of `StatusWatermarkValve`. I still prefer
to the current way for two reasons:
- Once `StatusWatermarkValve` needs a different output future, we only need
to change one place, otherwise we need to touch four internal methods.
- From design aspect, I think it is better to reduce the unnecessary
dependency between `StatusWatermarkValve` and `StreamTaskNetworkInput`.
Currently we have to rely on the channel index from `StreamTaskNetworkInput`.
But if we regard the `output` as a more general concept which might not be
consistent with the `output` in `emitNext`, then we can pass it into the
constructor in free from outside. Just like the previous
`ForwardingValveOutputHandler` which was also passed into the constructor of
`StatusWatermarkValve` from outside.
----------------------------------------------------------------
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
