Github user StefanRRichter commented on the issue:
https://github.com/apache/flink/pull/4900
I have one quick question, but maybe I miss something big here: wouldn't it
be possible to simply run `quiesceAndAwaitPending()` before `Operator::close()`
and avoid the locking/flag checking and not run into exceptions? If we ensure
that the task is no longer in running state, the "forgotten" timers can no
longer reflect in any checkpoint/savepoint and I think there is no guarantee of
a consistent output after canceling a job anyways?
---
