TL;DR: Issue with Flink’s registerProcessingTimeTimer after Task Manager restarts Hey folks, I hope you’re all doing well! I have a question regarding Flink’s Streaming API and the behavior of

registerProcessingTimeTimer

. I’ve encountered an issue with the

onTimer

method not getting fired in certain scenarios, and I was wondering if anyone has come across a similar problem and could offer some guidance. Here’s some context: we have a Stream API named “Consolidation,” which utilizes

registerProcessingTimeTimer

to schedule delayed processing for a specific field based on certain criteria. During testing on our staging environment with a low processing time (10 minutes), everything works fine. However, when we move to production with a larger processing time window (72 hours), it seems that the

onTimer

method is not triggered as expected. Upon further investigation, I’ve noticed that our Flink instances occasionally restart due to our k8s node assignment policy. This leads me to suspect that the restarts may be causing the created timers to be wiped out. Is it a valid expectation for

registerProcessingTimeTimer

to maintain the timers’ state and trigger them even after a restart? Should we be implementing something differently to ensure this expected behavior? Just to provide some additional information, our deployment process is based on Flink Operator, and the

upgradeMode

is set to “savepoint.” Our state backend is configured as follows:

state.backend: rocksdb
state.backend.incremental: true

On a side note, I have experimented with smaller window times, and without the restarts, the timers work as expected. While researching this issue, I came across the following excerpt from the documentation:

Timers are always asynchronously checkpointed, except for the combination of RocksDB backend / with incremental snapshots / with heap-based timers (will be resolved with FLINK-10026). Notice that large numbers of timers can increase the checkpointing time because timers are part of the checkpointed state. See the “Timer Coalescing” section for advice on how to reduce the number of timers.

If anyone has any insights or experiences to share regarding the behavior of

registerProcessingTimeTimer

after Task Manager restarts or any suggestions on how to tackle this issue, I would greatly appreciate your input. Thank you all in advance for your help!

Timers are included in state snapshots (checkpoints and savepoints), and are restored during restarts, provided you do a stateful restart. On the other hand, if you throw away the state, then the timers will be lost. I don't have any experience with the k8s operator and its upgradeMode, but I would expect it to work. Are you using the default state.backend.rocksdb.timer-service.factory, which keeps timers in RocksDB when using the rocksdb state backend? If so, then the docs you quoted aren't relevant -- that only applies if you set state.backend.rocksdb.timer-service.factory to HEAP.