admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2026-02-02 01:06:41 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix compaction handler may double release and block forever (#21015) (#21019)

Browse Source 
Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>
This commit is contained in:
congqixia 2022-12-07 14:31:26 +08:00 committed by GitHub
parent 35173aa35c
commit 9692f4ed0e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 10 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
internal/datacoord/compaction.go
View File

@ -191,12 +191,12 @@ func (c *compactionPlanHandler) execCompactionPlan(signal *compactionSignal, pla
err = c.sessions.Compaction(nodeID, plan)
if err != nil {
log.Warn("Try to Compaction but DataNode rejected", zap.Any("TargetNodeId", nodeID), zap.Any("planId", plan.GetPlanID()))
c.mu.Lock()
delete(c.plans, plan.PlanID)
c.executingTaskNum--
c.releaseQueue(nodeID)
c.mu.Unlock()
log.Warn("try to Compaction but DataNode rejected",
zap.Int64("targetNodeID", nodeID),
zap.Int64("planID", plan.GetPlanID()),
)
// do nothing here, prevent double release, see issue#21014
// release queue will be done in `updateCompaction`
return
}

4
internal/datacoord/compaction_test.go
View File

@ -124,6 +124,8 @@ func Test_compactionPlanHandler_execCompactionPlan(t *testing.T) {
parallelCh: make(map[int64]chan struct{}),
allocator: newMockAllocator(),
}
Params.DataCoordCfg.CompactionCheckIntervalInSeconds = 1
c.start()
err := c.execCompactionPlan(tt.args.signal, tt.args.plan)
assert.Equal(t, tt.err, err)
if err == nil {
@ -134,6 +136,7 @@ func Test_compactionPlanHandler_execCompactionPlan(t *testing.T) {
assert.Equal(t, tt.args.signal, task.triggerInfo)
assert.Equal(t, 1, c.executingTaskNum)
} else {
assert.Eventually(t,
func() bool {
c.mu.RLock()
@ -143,6 +146,7 @@ func Test_compactionPlanHandler_execCompactionPlan(t *testing.T) {
5*time.Second, 100*time.Millisecond)
}
}
c.stop()
})
}
}