[
https://issues.apache.org/jira/browse/FLINK-41244?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Timo Walther closed FLINK-41244.
--------------------------------
Fix Version/s: 2.2.2
2.3.1
2.4.0
2.1.4
Resolution: Fixed
Fixed:
- master: 6968bc461effbd8dc6f94d6bbb517ccc0a930146
- release-2.4: eb107bf98e5bf84f2de69d2bd8989538365982a8
- release-2.3: 640e2193b5a230fbab419f4f8bcb1bbf0ed8e6b3
- release-2.2: a9d367e0fd5275d7b934fcd26f0985d688b5c934
- release-2.1: 66e2d75a6bdc3356cef1ae4b4be5f7112aa569de
> PTF clearAllTimers() deletes unnamed timers across all keys and corrupts the
> key context
> ----------------------------------------------------------------------------------------
>
> Key: FLINK-41244
> URL: https://issues.apache.org/jira/browse/FLINK-41244
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
> Reporter: Timo Walther
> Assignee: Timo Walther
> Priority: Critical
> Labels: pull-request-available
> Fix For: 2.2.2, 2.3.1, 2.4.0, 2.1.4
>
>
> {{ProcessTableFunction.Context#clearAllTimers()}} (and {{{}clearAll(){}}})
> does not scope the removal of unnamed timers to the current partition key.
> Named timers are not affected.
> h3. Cause
> {{WritableInternalTimeContext.clearUnnamedTimers()}} calls
> {{InternalTimerService#forEachEventTimeTimer}} to collect timers, then
> deletes them:
> {code:java}
> unnamedTimerService.forEachEventTimeTimer((namespace, time) ->
> unnamedTimers.add(time));
> unnamedTimers.forEach(time ->
> unnamedTimerService.deleteEventTimeTimer(VoidNamespace.INSTANCE, time));
> {code}
> {{InternalTimerServiceImpl#foreachTimer}} iterates the timers of *all keys*
> in the subtask. It also calls {{keyContext.setCurrentKey(timer.getKey())}}
> for each timer. As a result:
> # The collected list contains timestamps from every key, not just the
> current one.
> # The deletes run under the key of the last iterated timer, not the current
> key. The current key's unnamed timers usually survive and fire later. A
> different key can lose a timer that happens to share a timestamp.
> # The operator's key context is left on an arbitrary key after the call.
> Subsequent state access in the same {{{}eval(){}}}/{{{}onTimer(){}}} call,
> such as {{clearAllState()}} in {{clearAll()}} or user {{@StateHint}} handles,
> can read or write another key's state.
> Named timers are not affected, because {{clearNamedTimers()}} goes through
> the keyed {{{}internal-named-timers-map{}}}.
> h3. Impact
> * Unnamed timers fire after {{{}clearAllTimers(){}}}/{{{}clearAll(){}}} was
> called.
> * Timers of other keys are silently deleted.
> * State corruption across keys is possible.
> * The cost is O(all timers in the subtask) per call.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
