ASF GitHub Bot commented on FLINK-5544:

Github user StefanRRichter commented on the issue:

    @shixiaogang I had a brief look into the updated PR. It seems like this 
current version does not include integrating timers with the other keyed state? 
I am asking because my plan would be to integrate timers more deeply with the 
keyed backends so that we can also use asynchronous and incremental snapshots. 
One of your previous comments mentioned something in this direction as well and 
I wanted to ask what your plans about that are? Otherwise,  I would take over 
and probably cherrypick some parts like the rocks timer heap from this PR and 
integrate them with the keyed backends myself so that those features can also 
be supported by making timers part of the backends checkpoint data.

> Implement Internal Timer Service in RocksDB
> -------------------------------------------
>                 Key: FLINK-5544
>                 URL: https://issues.apache.org/jira/browse/FLINK-5544
>             Project: Flink
>          Issue Type: New Feature
>          Components: State Backends, Checkpointing
>            Reporter: Xiaogang Shi
>            Assignee: Xiaogang Shi
>            Priority: Major
> Now the only implementation of internal timer service is 
> HeapInternalTimerService which stores all timers in memory. In the cases 
> where the number of keys is very large, the timer service will cost too much 
> memory. A implementation which stores timers in RocksDB seems good to deal 
> with these cases.
> It might be a little challenging to implement a RocksDB timer service because 
> the timers are accessed in different ways. When timers are triggered, we need 
> to access timers in the order of timestamp. But when performing checkpoints, 
> we must have a method to obtain all timers of a given key group.
> A good implementation, as suggested by [~StephanEwen], follows the idea of 
> merge sorting. We can store timers in RocksDB with the format 
> {{KEY_GROUP#TIMER#KEY}}. In this way, the timers under a key group are put 
> together and are sorted. 
> Then we can deploy an in-memory heap which keeps the first timer of each key 
> group to get the next timer to trigger. When a key group's first timer is 
> updated, we can efficiently update the heap.

This message was sent by Atlassian JIRA

Reply via email to