milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[skip ci]Add rocksmq_retention comment (#8016) 

Signed-off-by: fishpenguin <kun.yu@zilliz.com>
pull/7855/head
yukun 2021-09-15 23:29:48 +08:00 committed by GitHub
parent 0aa65f559a
commit e0761271dd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 9 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
internal/util/rocksmq/server/rocksmq/rocksmq_retention.go
View File

@ -107,6 +107,8 @@ func initRetentionInfo(kv *rocksdbkv.RocksdbKV, db *gorocksdb.DB) (*retentionInf
return ri, nil
}
// Before do retention, load retention info from rocksdb to retention info structure in goroutines.
// Because loadRetentionInfo may need some time, so do this asynchronously. Finally start retention goroutine.
func (ri *retentionInfo) startRetentionInfo() error {
var wg sync.WaitGroup
ri.kv.ResetPrefixLength(FixedChannelNameLen)
@ -123,6 +125,7 @@ func (ri *retentionInfo) startRetentionInfo() error {
return nil
}
// Read retention infos from rocksdb so that retention check can be done based on memory data
func (ri *retentionInfo) loadRetentionInfo(topic string, wg *sync.WaitGroup) {
// TODO(yukun): If there needs to add lock
// ll, ok := topicMu.Load(topic)
@ -248,6 +251,7 @@ func (ri *retentionInfo) loadRetentionInfo(topic string, wg *sync.WaitGroup) {
func (ri *retentionInfo) retention() error {
log.Debug("Rocksmq retention goroutine start!")
// Do retention check every 6s
ticker := time.NewTicker(time.Duration(TickerTimeInMinutes * int64(time.Minute) / 10))
for {
@ -271,6 +275,11 @@ func (ri *retentionInfo) retention() error {
}
}
// 1. Obtain pageAckedInfo and do time expired check, get the expired page scope;
// 2. Do iteration in the page after the last page in step 1 and get the last time expired message id;
// 3. Do size expired check in next page, and get the last size expired message id;
// 4. Do delete by range of [start_msg_id, end_msg_id) in rocksdb
// 5. Delete corresponding data in retentionInfo
func (ri *retentionInfo) expiredCleanUp(topic string) error {
// log.Debug("Timeticker triggers an expiredCleanUp task for topic: " + topic)
var ackedInfo *topicAckedInfo