enhance: [24]Increase task capacity and clean illegal task (milvus-io…

…#37896) (milvus-io#38095) 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. pr: milvus-io#37896 Signed-off-by: yangxuan <[email protected]>
shaoting-huang · Dec 2, 2024 · c32ad65 · c32ad65
1 parent 6a32b06
commit c32ad65
Show file tree

Hide file tree

Showing 3 changed files with 5 additions and 4 deletions.
diff --git a/configs/milvus.yaml b/configs/milvus.yaml
@@ -550,7 +550,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
     mix:

diff --git a/internal/datacoord/compaction.go b/internal/datacoord/compaction.go
@@ -335,15 +335,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() {

diff --git a/pkg/util/paramtable/component_param.go b/pkg/util/paramtable/component_param.go
@@ -3457,7 +3457,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,
 	}