worker: fix subtask initialized state (pingcap#240)

Author: amyangfei

dm/worker/subtask.go

@@ -109,6 +109,7 @@ func (st *SubTask) Init() error {
 
 	st.DDLInfo = make(chan *pb.DDLInfo, 1)
 
+	initializeUnitSuccess := true
 	// when error occurred, initialized units should be closed
 	// when continue sub task from loader / syncer, ahead units should be closed
 	var needCloseUnits []unit.Unit
@@ -117,7 +118,7 @@ func (st *SubTask) Init() error {
 			u.Close()
 		}
 
-		st.initialized.Set(len(needCloseUnits) == 0)
+		st.initialized.Set(initializeUnitSuccess)
 	}()
 
 	// every unit does base initialization in `Init`, and this must pass before start running the sub task
@@ -126,6 +127,7 @@ func (st *SubTask) Init() error {
 	for i, u := range st.units {
 		err := u.Init()
 		if err != nil {
+			initializeUnitSuccess = false
 			// when init fail, other units initialized before should be closed
 			for j := 0; j < i; j++ {
 				needCloseUnits = append(needCloseUnits, st.units[j])
@@ -140,6 +142,7 @@ func (st *SubTask) Init() error {
 		u := st.units[i]
 		isFresh, err := u.IsFreshTask()
 		if err != nil {
+			initializeUnitSuccess = false
 			return terror.Annotatef(err, "fail to get fresh status of subtask %s %s", st.cfg.Name, u.Type())
 		} else if !isFresh {
 			skipIdx = i

tests/initial_unit/run.sh

@@ -75,7 +75,14 @@ function run() {
 
         check_sync_diff $WORK_DIR $cur/conf/diff_config.toml
 
+        run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+            "resume-task test" \
+            "\"result\": true" 1 \
+            "\"result\": false" 1 \
+            "current stage is not paused not valid" 1
+
         cleanup_process
+        run_sql "drop database if exists initial_unit" $TIDB_PORT
         run_sql "drop database if exists dm_meta" $TIDB_PORT
     done
 }