fix: [2.5]Fix target segment marked dropped for save stats result twice (#45480)

issue: #45477 

master pr: #45478

---------

Signed-off-by: Cai Zhang <cai.zhang@zilliz.com>

ci/jenkins/PR.groovy

@@ -89,7 +89,7 @@ pipeline {
                 axes {
                     axis {
                         name 'milvus_deployment_option'
-                        values 'standalone', 'distributed', 'standalone-kafka-mmap'
+                        values 'standalone', 'distributed'
                     }
                 }
                 stages {

internal/datacoord/compaction_trigger.go

@@ -360,7 +360,10 @@ func (t *compactionTrigger) handleSignal(signal *compactionSignal) error {
 		coll, err := t.getCollection(group.collectionID)
 		if err != nil {
 			log.Warn("get collection info failed, skip handling compaction", zap.Error(err))
-			return err
+			if signal.collectionID != 0 {
+				return err
+			}
+			continue
 		}
 
 		if !signal.isForce && !isCollectionAutoCompactionEnabled(coll) {

internal/datacoord/meta.go

@@ -2117,8 +2117,8 @@ func (m *meta) SaveStatsResultSegment(oldSegmentID int64, result *workerpb.Stats
 	metricMutation := &segMetricMutation{stateChange: make(map[string]map[string]map[string]int)}
 
 	oldSegment := m.segments.GetSegment(oldSegmentID)
-	if oldSegment == nil {
+	if oldSegment == nil || !isSegmentHealthy(oldSegment) {
-		log.Warn("old segment is not found with stats task")
+		log.Warn("old segment is not found or not healthy with stats task")
 		return nil, merr.WrapErrSegmentNotFound(oldSegmentID)
 	}
 
@@ -2135,16 +2135,17 @@ func (m *meta) SaveStatsResultSegment(oldSegmentID int64, result *workerpb.Stats
 	}
 
 	segmentInfo := &datapb.SegmentInfo{
-		CollectionID:              oldSegment.GetCollectionID(),
+		CollectionID:   oldSegment.GetCollectionID(),
-		PartitionID:               oldSegment.GetPartitionID(),
+		PartitionID:    oldSegment.GetPartitionID(),
-		InsertChannel:             oldSegment.GetInsertChannel(),
+		InsertChannel:  oldSegment.GetInsertChannel(),
-		MaxRowNum:                 oldSegment.GetMaxRowNum(),
+		MaxRowNum:      oldSegment.GetMaxRowNum(),
-		LastExpireTime:            oldSegment.GetLastExpireTime(),
+		LastExpireTime: oldSegment.GetLastExpireTime(),
-		StartPosition:             oldSegment.GetStartPosition(),
+		StartPosition:  oldSegment.GetStartPosition(),
-		DmlPosition:               oldSegment.GetDmlPosition(),
+		DmlPosition:    oldSegment.GetDmlPosition(),
-		IsImporting:               oldSegment.GetIsImporting(),
+		IsImporting:    oldSegment.GetIsImporting(),
-		StorageVersion:            oldSegment.GetStorageVersion(),
+		StorageVersion: oldSegment.GetStorageVersion(),
-		State:                     oldSegment.GetState(),
+		// only flushed segment can do sort stats
+		State:                     commonpb.SegmentState_Flushed,
 		Level:                     oldSegment.GetLevel(),
 		LastLevel:                 oldSegment.GetLastLevel(),
 		PartitionStatsVersion:     oldSegment.GetPartitionStatsVersion(),

internal/datacoord/task_scheduler.go

@@ -581,7 +581,9 @@ func (s *taskScheduler) processFinished(task Task) bool {
 	client, exist := s.nodeManager.GetClientByID(task.GetNodeID())
 	if exist {
 		if !task.DropTaskOnWorker(s.ctx, client) {
-			return false
+			log.Ctx(s.ctx).Warn("drop task on worker failed, but ignore it",
+				zap.Int64("taskID", task.GetTaskID()), zap.Int64("nodeID", task.GetNodeID()))
+			return true
 		}
 	}
 	log.Ctx(s.ctx).Info("task has been finished", zap.Int64("taskID", task.GetTaskID()),

internal/datacoord/task_scheduler_test.go

@@ -1301,16 +1301,11 @@ func (s *taskSchedulerSuite) Test_analyzeTaskFailCase() {
 		// set job info failed --> state: Finished
 		catalog.EXPECT().SaveAnalyzeTask(mock.Anything, mock.Anything).Return(errors.New("set job info failed")).Once()
 
-		// set job success, drop job on task failed --> state: Finished
+		// set job success, drop job on task failed --> state: Finished (skip drop error, task success)
 		catalog.EXPECT().SaveAnalyzeTask(mock.Anything, mock.Anything).Return(nil).Once()
 		workerManager.EXPECT().GetClientByID(mock.Anything).Return(in, true).Once()
 		in.EXPECT().DropJobsV2(mock.Anything, mock.Anything).Return(merr.Status(errors.New("drop job failed")), nil).Once()
 
-		// drop job success --> no task
-		catalog.EXPECT().SaveAnalyzeTask(mock.Anything, mock.Anything).Return(nil).Once()
-		workerManager.EXPECT().GetClientByID(mock.Anything).Return(in, true).Once()
-		in.EXPECT().DropJobsV2(mock.Anything, mock.Anything).Return(merr.Success(), nil).Once()
-
 		for {
 			if scheduler.pendingTasks.TaskCount() == 0 {
 				taskNum := scheduler.runningTasks.Len()