Repository: milvus-io/milvus
File: internal/querycoordv2/task/executor.go (Go)

Code before:

	// Get shard leader for the given replica and segment
	channel := mergeTask.req.GetInfos()[0].GetInsertChannel()
	leader, ok := getShardLeader(ex.meta.ReplicaManager, ex.dist, task.CollectionID(), action.Node(), channel)
	if !ok {
		msg := "no shard leader for the segment to execute loading"
		task.SetErr(utils.WrapError(msg, ErrTaskStale))
		log.Warn(msg, zap.String("shard", channel))
		return
	}

	log.Info("load segments...")
	status, err := ex.cluster.LoadSegments(task.Context(), leader, mergeTask.req)
	if err != nil {
		log.Warn("failed to load segment, it may be a false failure", zap.Error(err))
		return
	}
	if status.ErrorCode != commonpb.ErrorCode_Success {
		log.Warn("failed to load segment", zap.String("reason", status.GetReason()))
		return
	}
	elapsed := time.Since(startTs)
	log.Info("load segments done", zap.Int64("taskID", task.ID()), zap.Duration("timeTaken", elapsed))
}

func (ex *Executor) removeTask(task Task, step int) {
	if task.Err() != nil {
		log.Info("excute action done, remove it",
			zap.Int64("taskID", task.ID()),
			zap.Int("step", step),
			zap.Error(task.Err()))
	}

	ex.executingTasks.Delete(task.ID())
	ex.executingTaskNum.Dec()
}

func (ex *Executor) executeSegmentAction(task *SegmentTask, step int) {
	switch task.Actions()[step].Type() {
	case ActionTypeGrow:
		ex.loadSegment(task, step)

	case ActionTypeReduce:
		ex.releaseSegment(task, step)
	}
}

// loadSegment commits the request to merger,
// not really executes the request
func (ex *Executor) loadSegment(task *SegmentTask, step int) error {
	action := task.Actions()[step].(*SegmentAction)

Review comment (suggestion):
Here `mergeTask` contains all merged tasks, `task` is only the first one, need to set error and cancel all tasks in `mergeTask`

Code after:
		zap.String("shard", task.Shard()),
		zap.Int64s("segmentIDs", segments),
		zap.Int64("nodeID", action.Node()),
		zap.Int64("source", task.SourceID()),
	)

	// Get shard leader for the given replica and segment
	channel := mergeTask.req.GetInfos()[0].GetInsertChannel()
	leader, ok := getShardLeader(ex.meta.ReplicaManager, ex.dist, task.CollectionID(), action.Node(), channel)
	if !ok {
		msg := "no shard leader for the segment to execute loading"
		task.SetErr(utils.WrapError(msg, ErrTaskStale))
		log.Warn(msg, zap.String("shard", channel))
		return
	}

	log.Info("load segments...")
	status, err := ex.cluster.LoadSegments(task.Context(), leader, mergeTask.req)
	if err != nil {
		log.Warn("failed to load segment, it may be a false failure", zap.Error(err))
		return
	}
	if status.ErrorCode == commonpb.ErrorCode_InsufficientMemoryToLoad {
		log.Warn("insufficient memory to load segment", zap.String("err", status.GetReason()))
		task.SetErr(fmt.Errorf("%w, err:%s", ErrInsufficientMemory, status.GetReason()))
		task.Cancel()
		return
	}
	if status.ErrorCode != commonpb.ErrorCode_Success {
		log.Warn("failed to load segment", zap.String("reason", status.GetReason()))
		return
	}
	elapsed := time.Since(startTs)
	log.Info("load segments done", zap.Int64("taskID", task.ID()), zap.Duration("timeTaken", elapsed))
}

func (ex *Executor) removeTask(task Task, step int) {
	if task.Err() != nil {
		log.Info("excute action done, remove it",
			zap.Int64("taskID", task.ID()),
			zap.Int("step", step),
			zap.Error(task.Err()))
	}

	ex.executingTasks.Delete(task.ID())
	ex.executingTaskNum.Dec()
}

func (ex *Executor) executeSegmentAction(task *SegmentTask, step int) {
	switch task.Actions()[step].Type() {
	case ActionTypeGrow: