Github user dianfu commented on a diff in the pull request:
https://github.com/apache/flink/pull/5063#discussion_r152955709
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/RowTimeUnboundedOver.scala
---
@@ -116,7 +116,7 @@ abstract class RowTimeUnboundedOver(
// discard late record
if (timestamp > curWatermark) {
// ensure every key just registers one timer
- ctx.timerService.registerEventTimeTimer(curWatermark + 1)
+ ctx.timerService.registerEventTimeTimer(timestamp)
--- End diff --
@fhueske Thanks a lot for your comments. Your concern makes sense to me. I
think the current implementation is ok under periodic watermark. But I'm not
sure if it's optimal under punctuated watermark. We will perform some
performance test for unbounded over under punctuated watermark and share the
results.
---
