HunterXHunter

After modification, only another error will be thrown. > Caused by: org.apache.flink.runtime.taskmanager.AsynchronousException: Caught exception while processing timer. at org.apache.flink.streaming.runtime.tasks.StreamTask$StreamTaskAsyncExceptionHandler.handleAsyncException(StreamTask.java:1535) ~[flink-dist-1.15.1-vip.jar:1.15.1-vip] at org.apache.flink.streaming.runtime.tasks.StreamTask.handleAsyncException(StreamTask.java:1510) ~[flink-dist-1.15.1-vip.jar:1.15.1-vip] at org.apache.flink.streaming.runtime.tasks.StreamTask.invokeProcessingTimeCallback(StreamTask.java:1650) ~[flink-dist-1.15.1-vip.jar:1.15.1-vip] at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$null$21(StreamTask.java:1639) ~[flink-dist-1.15.1-vip.jar:1.15.1-vip] at...

I'm not sure. The problem occurred after upgrading 3.1.1.

I also encountered this exception. I read Kafka data in batches.Therefore, we want to use latest when the timestamp is exceeded.

its work , i had try.

As I know, flink 15+ version has support checkpointing with finished tasks (https://flink.apache.org/2022/07/11/flip-147-support-checkpoints-after-tasks-finished-part-one/#support-checkpointing-with-finished-tasks). so commit will be executed once in `notifyCheckpointComplete`. And this was true in my tests.

fixed #3960