[
https://issues.apache.org/jira/browse/FLINK-10327?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16617287#comment-16617287
]
Kostas Kloudas commented on FLINK-10327:
----------------------------------------
[~pnowojski] as you mention, in terms of code, the change may be just a couple
of lines, but the implications of exposing such functionality at the function
level are more than that.
As I said in the PR:
1) manipulating the watermark should be explicitly done from a WatermarkEmitter
(which is exposed to the user).
2) if you want to run a "callback" upon watermark, then so far the trick is to
register a timer for watermark + 1. This is even more "powerful" than just
exposing the `processWatermark` to the processFunction, as by reacting to a
timer, you are already in a keyed context (so you have access to state).
3) if you want to go even lower than that, then you should write your own
operator, and be explicit on how to handle the different parts that constitute
an operator, i.e. collectors, timestamps of emitted elements
(collector.setTimestamp), etc.
If we agree on that, then I would recommend to close the issue as "won't fix".
> Pass processWatermark calls from (Co)ProcessOperator to (Co)ProcessFunction
> ---------------------------------------------------------------------------
>
> Key: FLINK-10327
> URL: https://issues.apache.org/jira/browse/FLINK-10327
> Project: Flink
> Issue Type: New Feature
> Components: Streaming
> Reporter: Piotr Nowojski
> Assignee: Piotr Nowojski
> Priority: Major
> Labels: pull-request-available
>
> Currently {{CoProcessFunction}} can not react to changes watermark
> advancement. By passing {{processWatermark}} calls to function we would give
> a way to perform some actions on watermark advancement, like state clean up
> or emitting some results after accumulating some data.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
