Timo Walther created FLINK-41244:
------------------------------------
Summary: 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
{{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)
