[FLINK-9182] async checkpoints for timer service

• Type: Improvement
• Status: Closed
• Priority: Minor
• Resolution: Duplicate
• Affects Version/s: 1.4.2, 1.5.0
• Component/s: Runtime / State Backends
• Assignee: makeyang
• Reporter: makeyang
• Created: 16/Apr/18 11:14
• Updated: 06/Aug/18 08:48
• Resolved: 06/Aug/18 08:48

Description

1. problem description:
   1. with the increase in the number of  'InternalTimer' object the checkpoint more and more slowly
2. improvement desgin
   1. maintain a stateTableVersion, which is exactly the same thing as CopyOnWriteStateTable and snapshotVersions which is exactly the same thing as CopyOnWriteStateTable in InternalTimeServiceManager. one more thing: a readwrite lock, which is used to protect snapshotVersions and stateTableVersion
   2. for each InternalTimer, add 2 more properties: create version and delete version beside 3 existing properties: timestamp, key and namespace. each time a Timer is registered in timerservice, it is created with stateTableVersion as its create version while delete version is -1. each time when timer is deleted in timerservice, it is marked delete for giving it a delete verison equals to stateTableVersion without physically delete it from timerservice.
   3. each time when try to snapshot timers, InternalTimeServiceManager increase its stateTableVersion and add this stateTableVersion in snapshotVersions. these 2 operators are protected by write lock of InternalTimeServiceManager. that current stateTableVersion take as snapshot version of this snapshot
   4. shallow copy <String,HeapInternalTimerService> tuples
   5. then use a another thread asynchronous snapshot whole things: keyserialized, namespaceserializer and timers. for timers which is not deleted(delete version is -1) and create version less than snapshot version, serialized it. for timers whose delete version is not -1 and is bigger than or equals snapshot version, serialized it. otherwise, it will not be serialized by this snapshot.
   6. when everything is serialized, remove snapshot version in snapshotVersions, which is still in another thread and this action is guarded by write lock.
   7. last thing: timer physical deletion. 2 places to physically delete timers: each time when timer is deleted in timerservice, it is marked delete for giving it a delete verison equals to stateTableVersion without physically delete it from timerservice. after this, check if snapshotVersions size is 0 (which means there is no running snapshot) and if true, delete timer .the other place to delete is in snapshot timer's iterat: when timer's delete version is less than min value of snapshotVersions, which means the timer is deleted and no running snapshot should keep it.
   8. some more additions: processingTimeTimers and eventTimeTimers for each group used to be hashset and now it is changed to concurrenthashmap with key+namesapce+timestamp as its hash key.
3. related mail list thread
   1. http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Slow-flink-checkpoint-td18946.html
4. github pull request
   1. //coming soon

Issue Links

• is superceded by: FLINK-9485 Improving Flink’s timer management for large state
• links to: GitHub Pull Request #5908
• links to: GitHub Pull Request #6019