flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-5544) Implement Internal Timer Service in RocksDB
Date Wed, 29 Nov 2017 10:11:00 GMT

    [ https://issues.apache.org/jira/browse/FLINK-5544?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16270535#comment-16270535

ASF GitHub Bot commented on FLINK-5544:

Github user StefanRRichter commented on the issue:

    @shixiaogang The changes sound helpful, considering that this PR was implemented before
incremental checkpoints. For a final opinion I would have to review the code. One concern
that I have is a wild grow of more and more reported state types that we introduce. I am currently
working on a feature that introduces recovery from task-local state, which also looks like
it will add to the states that can be produced in a snapshot. So we might need some form of
consolidation in the future to avoid a "wild-grow" of more and more state types that make
everything more complex and harder to maintain. But overall, I think the general idea is very

> 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
> 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

View raw message