perf(runner): close fs service client when task finished

interfaces/task_runner.go

@@ -12,4 +12,5 @@ type TaskRunner interface {
 	Dispose() (err error)
 	SetSubscribeTimeout(timeout time.Duration)
 	GetTaskId() (id primitive.ObjectID)
+	CleanUp() (err error)
 }

task/handler/runner.go

@@ -199,6 +199,19 @@ func (r *Runner) Dispose() (err error) {
 	}, backoff.NewExponentialBackOff())
 }
 
+// CleanUp clean up task runner
+func (r *Runner) CleanUp() (err error) {
+	// close fs service
+	fsSvc := r.fsSvc.GetFsService().GetFs()
+	if fsSvc == nil {
+		return
+	}
+	if err = fsSvc.Close(); err != nil {
+		return
+	}
+	return
+}
+
 func (r *Runner) SetSubscribeTimeout(timeout time.Duration) {
 	r.subscribeTimeout = timeout
 }

task/handler/service.go

@@ -376,7 +376,12 @@ func (svc *Service) run(taskId primitive.ObjectID) (err error) {
 	go func() {
 		// delete runner from pool
 		defer svc.deleteRunner(r.GetTaskId())
-
+		defer func(r interfaces.TaskRunner) {
+			err := r.CleanUp()
+			if err != nil {
+				log.Errorf("task[%s] clean up error: %v", r.GetTaskId().Hex(), err)
+			}
+		}(r)
 		// run task process (blocking)
 		// error or finish after task runner ends
 		if err := r.Run(); err != nil {