Repository: milvus-io/milvus
File: internal/datacoord/import_checker.go (Go)

Code before:
		return
	}
	for _, t := range newTasks {
		err = c.imeta.AddTask(t)
		if err != nil {
			log.Warn("add preimport task failed", WrapTaskLog(t, zap.Error(err))...)
			return
		}
		log.Info("add new preimport task", WrapTaskLog(t)...)
	}
	err = c.imeta.UpdateJob(job.GetJobID(), UpdateJobState(internalpb.ImportJobState_PreImporting))
	if err != nil {
		log.Warn("failed to update job state to PreImporting", zap.Int64("jobID", job.GetJobID()), zap.Error(err))
	}
}

func (c *importChecker) checkPreImportingJob(job ImportJob) {
	lacks := c.getLackFilesForImports(job)
	if len(lacks) == 0 {
		return
	}

	err := CheckDiskQuota(job, c.meta, c.imeta)
	if err != nil {
		log.Warn("import failed, disk quota exceeded", zap.Int64("jobID", job.GetJobID()), zap.Error(err))
		err = c.imeta.UpdateJob(job.GetJobID(), UpdateJobState(internalpb.ImportJobState_Failed), UpdateJobReason(err.Error()))
		if err != nil {
			log.Warn("failed to update job state to Failed", zap.Int64("jobID", job.GetJobID()), zap.Error(err))
		}
		return
	}

	groups := RegroupImportFiles(job, lacks)
	newTasks, err := NewImportTasks(groups, job, c.sm, c.alloc)
	if err != nil {
		log.Warn("new import tasks failed", zap.Error(err))
		return
	}
	for _, t := range newTasks {
		err = c.imeta.AddTask(t)
		if err != nil {
			log.Warn("add new import task failed", WrapTaskLog(t, zap.Error(err))...)
			return
		}
		log.Info("add new import task", WrapTaskLog(t)...)
	}
	err = c.imeta.UpdateJob(job.GetJobID(), UpdateJobState(internalpb.ImportJobState_Importing))
	if err != nil {
		log.Warn("failed to update job state to Importing", zap.Int64("jobID", job.GetJobID()), zap.Error(err))
	}
}

Review comment (suggestion):
I suggest refraining from immediately throwing an error. Instead, it could remain in an "InProgress" state, but with the reason clearly stated as insufficient quota.

Code after:
		return
	}
	for _, t := range newTasks {
		err = c.imeta.AddTask(t)
		if err != nil {
			log.Warn("add preimport task failed", WrapTaskLog(t, zap.Error(err))...)
			return
		}
		log.Info("add new preimport task", WrapTaskLog(t)...)
	}
	err = c.imeta.UpdateJob(job.GetJobID(), UpdateJobState(internalpb.ImportJobState_PreImporting))
	if err != nil {
		log.Warn("failed to update job state to PreImporting", zap.Int64("jobID", job.GetJobID()), zap.Error(err))
	}
}

func (c *importChecker) checkPreImportingJob(job ImportJob) {
	lacks := c.getLackFilesForImports(job)
	if len(lacks) == 0 {
		return
	}

	requestSize, err := CheckDiskQuota(job, c.meta, c.imeta)
	if err != nil {
		log.Warn("import failed, disk quota exceeded", zap.Int64("jobID", job.GetJobID()), zap.Error(err))
		err = c.imeta.UpdateJob(job.GetJobID(), UpdateJobState(internalpb.ImportJobState_Failed), UpdateJobReason(err.Error()))
		if err != nil {
			log.Warn("failed to update job state to Failed", zap.Int64("jobID", job.GetJobID()), zap.Error(err))
		}
		return
	}

	groups := RegroupImportFiles(job, lacks)
	newTasks, err := NewImportTasks(groups, job, c.sm, c.alloc)
	if err != nil {
		log.Warn("new import tasks failed", zap.Error(err))
		return
	}
	for _, t := range newTasks {
		err = c.imeta.AddTask(t)
		if err != nil {
			log.Warn("add new import task failed", WrapTaskLog(t, zap.Error(err))...)
			return
		}
		log.Info("add new import task", WrapTaskLog(t)...)
	}
	err = c.imeta.UpdateJob(job.GetJobID(), UpdateJobState(internalpb.ImportJobState_Importing), UpdateRequestedDiskSize(requestSize))
	if err != nil {
		log.Warn("failed to update job state to Importing", zap.Int64("jobID", job.GetJobID()), zap.Error(err))
	}
}