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_r322537457
##########
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:
Thanks for this concern. I actually also hesitated to this issue before and
passed the `output` in specific methods in `StatusWatermarkValve` instead of
constructor in previous version.
I think there are two different possibilities here:
- Make `StatusWatermarkValve` completely coupled with
`StreamTaskNetworkInput`, then we could even construct `StatusWatermarkValve`
inside constructor of `StreamTaskNetworkInput`. This is a positive point. And
we assume that the `output` is same for both `emitNext` and
`StatusWatermarkValve` now. But if the `output` instances are different in
future as you mentioned, then we need to pass another `output` into
`StreamTaskNetworkInput` for `StatusWatermarkValve`. Especially it would cause
many changes in `StatusWatermarkValve` because there are four internal methods
involved in passing `output` argument in `StatusWatermarkValve`.
- Construct the different `output` instances in the upper layer of
`StreamInputProcessor` for both `StreamTaskNetworkInput` and
`StatusWatermarkValve` as the current implementation in this PR. In other words
`StatusWatermarkValve` could be regarded as a `output` wrapper which is
actually used in `StatusWatermarkValve`, and the wrapped `output` inside
`StatusWatermarkValve` could be anything else different from the `output` in
`emitNext`. If it changes we only need to construct a different `output`
instance in `StreamInputProcessor` and pass it into `StatusWatermarkValve`, no
need to touch four internal methods inside it.
I have no obvious preference for above two ways, because we could think of
side effects for any way.
For the alternative of encapsulating `StatusWatermarkValve` behind `output`,
I wonder it is not good to let `output` see any channel index concept,
especially considering the source output future. And the current
`StatusWatermarkValve` is a bit heavy-weight, so encapsulating it into `output`
would complicate `output` implementation. In contrast, making `output` as an
field of `StatusWatermarkValve` seems more light-weight.
----------------------------------------------------------------
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
