enhance: Add filter for remove compaction command (#327)

Add task state & collection, partition id filter for remove compaction
command.

Signed-off-by: Congqi Xia <[email protected]>

states/etcd/remove/compaction_task.go

@@ -17,6 +17,9 @@ type CompactionTaskParam struct {
 	CompactionType      string `name:"type" default:"ClusteringCompaction" desc:"compaction type to remove"`
 	JobID               string `name:"jobID" default:"" desc:"jobID also known as triggerID"`
 	TaskID              string `name:"taskID" default:"" desc:"taskID also known as planID"`
+	State               string `name:"state" default:"" desc:"task state"`
+	CollectionID        int64  `name:"collectionID" default:"0" desc:"collection id to filter"`
+	PartitionID         int64  `name:"partitionID" default:"0" desc:"partitionID id to filter"`
 	Run                 bool   `name:"run" default:"false" desc:"flag to control actually run or dry"`
 }
 
@@ -32,6 +35,15 @@ func (c *ComponentRemove) RemoveCompactionTaskCommand(ctx context.Context, p *Co
 		if p.TaskID != "" && fmt.Sprint(task.GetPlanID()) != p.TaskID {
 			return false
 		}
+		if p.State != "" && task.State.String() != p.State {
+			return false
+		}
+		if p.CollectionID > 0 && task.GetCollectionID() != p.CollectionID {
+			return false
+		}
+		if p.PartitionID > 0 && task.GetPartitionID() != p.PartitionID {
+			return false
+		}
 		return true
 	})
 	if err != nil {