admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2025-12-28 22:45:26 +08:00
Code Issues Packages Projects Releases Wiki Activity

Move topic lock to the front of final delete in retention expired cleanup (#11076)

Browse Source 
Signed-off-by: fishpenguin <kun.yu@zilliz.com>
This commit is contained in:
yukun 2021-11-02 17:28:31 +08:00 committed by GitHub
parent a8722797fe
commit ea6cd7d27c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 19 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -155,17 +155,6 @@ func (ri *retentionInfo) Stop() {
func (ri *retentionInfo) newExpiredCleanUp(topic string) error {
log.Debug("Timeticker triggers an expiredCleanUp task for topic: " + topic)
ll, ok := topicMu.Load(topic)
if !ok {
return fmt.Errorf("topic name = %s not exist", topic)
}
lock, ok := ll.(*sync.Mutex)
if !ok {
return fmt.Errorf("get mutex failed, topic name = %s", topic)
}
lock.Lock()
defer lock.Unlock()
var deletedAckedSize int64 = 0
var startID UniqueID
var endID UniqueID
@ -346,7 +335,25 @@ func (ri *retentionInfo) newExpiredCleanUp(topic string) error {
writeBatch.DeleteRange([]byte(ackedStartIDKey), []byte(ackedEndIDKey))
}
newAckedSize := totalAckedSize - deletedAckedSize
ll, ok := topicMu.Load(topic)
if !ok {
return fmt.Errorf("topic name = %s not exist", topic)
}
lock, ok := ll.(*sync.Mutex)
if !ok {
return fmt.Errorf("get mutex failed, topic name = %s", topic)
}
lock.Lock()
defer lock.Unlock()
currentAckedSizeVal, err := ri.kv.Load(ackedSizeKey)
if err != nil {
return err
}
currentAckedSize, err := strconv.ParseInt(currentAckedSizeVal, 10, 64)
if err != nil {
return err
}
newAckedSize := currentAckedSize - deletedAckedSize
writeBatch.Put([]byte(ackedSizeKey), []byte(strconv.FormatInt(newAckedSize, 10)))
err = DeleteMessages(ri.db, topic, startID, endID)