// OperateSubTask stop/resume/pause  sub task.
OperateSubTask stop/resume/pause  sub task.
// OperateSubTask stop / resume / pause sub task .
func (w *SourceWorker) OperateSubTask(name string, op pb.TaskOp) error {
	w.Lock()
	defer w.Unlock()
	if w.closed.Load() {
		return terror.ErrWorkerAlreadyClosed.Generate()
	}
	st := w.subTaskHolder.findSubTask(name)
	if st == nil {
		return terror.ErrWorkerSubTaskNotFound.Generate(name)
	}
	w.l.Info("OperateSubTask start", zap.Stringer("op", op), zap.String("task", name))
	var err error
	switch op {
	case pb.TaskOp_Delete:
		w.l.Info("delete subtask", zap.String("task", name))
		st.Close()
		w.subTaskHolder.removeSubTask(name)
	case pb.TaskOp_Pause, pb.TaskOp_Stop:
		w.l.Info("pause subtask", zap.String("task", name))
		err = st.Pause()
	case pb.TaskOp_Resume:
		failpoint.Inject("SkipRefreshFromETCDInUT", func(_ failpoint.Value) {
			failpoint.Goto("bypassRefresh")
		})
		if refreshErr := w.tryRefreshSubTaskAndSourceConfig(st); refreshErr != nil {
			w.l.Warn("can not update subtask config now", zap.Error(refreshErr))
		}
		failpoint.Label("bypassRefresh")
		w.l.Info("resume subtask", zap.String("task", name))
		err = st.Resume(w.getRelayWithoutLock())
	case pb.TaskOp_AutoResume:
		w.l.Info("auto_resume subtask", zap.String("task", name))
		err = st.Resume(w.getRelayWithoutLock())
	default:
		err = terror.ErrWorkerUpdateTaskStage.Generatef("invalid operate %s on subtask %v", op, name)
	}
	w.l.Info("OperateSubTask finished", zap.Stringer("op", op), zap.String("task", name))
	return err
}
func ( w * SourceWorker ) OperateSubTask ( name string , op pb . TaskOp ) error { w . Lock ( ) defer w . Unlock ( ) if w . closed . Load ( ) { return terror . ErrWorkerAlreadyClosed . Generate ( ) } st := w . subTaskHolder . findSubTask ( name ) if st == nil { return terror . ErrWorkerSubTaskNotFound . Generate ( name ) } w . l . Info ( " " , zap . Stringer ( " " , op ) , zap . String ( " " , name ) ) var err error switch op { case pb . TaskOp_Delete : w . l . Info ( " " , zap . String ( " " , name ) ) st . Close ( ) w . subTaskHolder . removeSubTask ( name ) case pb . TaskOp_Pause , pb . TaskOp_Stop : w . l . Info ( " " , zap . String ( " " , name ) ) err = st . Pause ( ) case pb . TaskOp_Resume : failpoint . Inject ( " " , func ( _ failpoint . Value ) { failpoint . Goto ( " " ) } ) if refreshErr := w . tryRefreshSubTaskAndSourceConfig ( st ) ; refreshErr != nil { w . l . Warn ( " " , zap . Error ( refreshErr ) ) } failpoint . Label ( " " ) w . l . Info ( " " , zap . String ( " " , name ) ) err = st . Resume ( w . getRelayWithoutLock ( ) ) case pb . TaskOp_AutoResume : w . l . Info ( " " , zap . String ( " " , name ) ) err = st . Resume ( w . getRelayWithoutLock ( ) ) default : err = terror . ErrWorkerUpdateTaskStage . Generatef ( " " , op , name ) } w . l . Info ( " " , zap . Stringer ( " " , op ) , zap . String ( " " , name ) ) return err }
OperateSubTask stop/resume/pause  sub task.
OperateSubTask stop / resume / pause sub task .
// NOTE: for current unit is not syncer unit or is in shard merge. // TODO(ehco) change to auto_restart