Github user sunjincheng121 commented on the issue:
https://github.com/apache/flink/pull/3585
Hi @fhueske, in our point of view, register timer should be correlated to
the event time of the CURRENT data, but not currentWaterMark (which is
correlated to the event time of the PREVIOUS datas) . Using the PREVIOUS event
time (`currentWatermark + 1`) to register the timer for CURRENT data may cause
some problems IMO. I do not have whole lots of counterexamples for now, just
intuitively think this is not the correct way. One example I have is:
Data line -> ... 5 7 4 3 2
Watermark line -> ... 5 4 3 2 1
In this case, assuming when watermark is `3`ï¼we get `4` and `7`, then we
use ` currentWatermark + 1` to register timer `timer(3+1)` for both `4` and
`7`. When watermark goes to `4`ï¼timer `timer(3+1)` be triggered (then `4` and
`7` will be emitted). After this, data `5` comes (while the watermark is still
`4`), then we register timer `timer(4+1)` for 5. This is incorrect, as we
should always ensure data `5` is triggered before data `7`. I run a test case,
and printed the results of [event time][currentWatermark] as below. I hope it
can help to explain my point.
Datas:
```
(1L, 1, "Hello"),
(2L, 2, "Hello"),
(4L, 4, "Hello"),
(3L, 3, "Hello"),
(5L, 3, "Hello"),
(7L, 7, "Hello"),
(6L, 7, "Hello"),
(8L, 8, "Hello World"),
(7L, 8, "Hello"),
(5L, 5, "Hello"),
(20L, 20, "Hello World"),
(9L, 9, "Hello World"))
```
Event time of each data & Watermark: (newTS_currentWM[event
time][currentWatermark])
```
newTS_currentWM[1][-9223372036854775808]
newTS_currentWM[2][-1]
newTS_currentWM[4][0]
newTS_currentWM[3][2]
newTS_currentWM[5][2]
newTS_currentWM[7][3] // a data with event time 7, it will be registered
timer to (3+1)
newTS_currentWM[6][5] // a data with event time 6, it will be registered
timer to (5+1)
newTS_currentWM[8][5]
newTS_currentWM[7][6]
newTS_currentWM[5][6]
newTS_currentWM[20][6]
newTS_currentWM[9][18]
```
Another important reason of using allowedLateness to configure the
watermark in my design is that, in our point of view, watermark is a semantics
which is correlated to the stream lateness SLA. User exactly knows the tuple of
(SLA, allowedLateness). For instance:
******************************************
A system which always ensure the order:
SLA(100%) = Watermark(eventTime)
******************************************
For flink system process out of order data, and user have different
watermarks indicating different SLA levels:
(SLA, allowedLateness) = Watermark(eventTime - allowedLateness)
(SLA(80%), 0) = Watermark(eventTime)
(SLA(90%), 2) = Watermark(eventTime - 2)
(SLA(95%), 5) = Watermark(eventTime - 5)
We want to keep the interface to let user configure the
allowedLateness(AssignerWithPeriodicWatermarks|AssignerWithPunctuatedWatermarks),
thereby the correlated SLA.
What do you think ?
Thanks,
SunJincheng
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
