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

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

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

internal/datacoord/compaction.go

@@ -194,12 +194,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
 		}
 

internal/datacoord/compaction_test.go

@@ -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()
 		})
 	}
 }