Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5688#discussion_r190136036
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/KeyedProcessFunctionWithCleanupState.scala
---
@@ -44,8 +45,20 @@ abstract class KeyedProcessFunctionWithCleanupState[K,
I, O](queryConfig: Stream
protected def registerProcessingCleanupTimer(
ctx: KeyedProcessFunction[K, I, O]#Context,
currentTime: Long): Unit = {
- if (stateCleaningEnabled) {
+ registerCleanupTimer(ctx, currentTime, TimeDomain.PROCESSING_TIME)
+ }
+ protected def registerEventCleanupTimer(
--- End diff --
@fhueske is there a reason why we have no event-time clean-up state so far?
I think it would make sense to move this change to a separate PR. We should
make the notion of time configurable through `StreamQueryConfig`.
---
