enhance: Increase task capacity and clean illegal task (#37896)

1. taskQueueCapacity 256 is too small for production when we want to
re-write the entire collection

2. tasks should be cleaned when unable to recover, or the meta will
remain in etcd forever later.

Signed-off-by: yangxuan <xuan.yang@zilliz.com>

configs/milvus.yaml

@@ -562,7 +562,7 @@ dataCoord:
     # level is prioritized by level: L0 compactions first, then mix compactions, then clustering compactions.
     # mix is prioritized by level: mix compactions first, then L0 compactions, then clustering compactions.
     taskPrioritizer: default
-    taskQueueCapacity: 256 # compaction task queue size
+    taskQueueCapacity: 100000 # compaction task queue size
     rpcTimeout: 10
     maxParallelTaskNum: 10
     dropTolerance: 86400 # Compaction task will be cleaned after finish longer than this time(in seconds)

internal/datacoord/compaction.go

@@ -337,15 +337,16 @@ func (c *compactionPlanHandler) loadMeta() {
 					zap.String("state", task.GetState().String()))
 				continue
 			} else {
-				// TODO: how to deal with the create failed tasks, leave it in meta forever?
 				t, err := c.createCompactTask(task)
 				if err != nil {
-					log.Warn("compactionPlanHandler loadMeta create compactionTask failed",
+					log.Info("compactionPlanHandler loadMeta create compactionTask failed, try to clean it",
 						zap.Int64("planID", task.GetPlanID()),
 						zap.String("type", task.GetType().String()),
 						zap.String("state", task.GetState().String()),
 						zap.Error(err),
 					)
+					// ignore the drop error
+					c.meta.DropCompactionTask(task)
 					continue
 				}
 				if t.NeedReAssignNodeID() {

pkg/util/paramtable/component_param.go

@@ -3530,7 +3530,7 @@ mix is prioritized by level: mix compactions first, then L0 compactions, then cl
 	p.CompactionTaskQueueCapacity = ParamItem{
 		Key:          "dataCoord.compaction.taskQueueCapacity",
 		Version:      "2.5.0",
-		DefaultValue: "256",
+		DefaultValue: "100000",
 		Doc:          `compaction task queue size`,
 		Export:       true,
 	}