cleanup: remove TaskStateRestart

and TaskStateComplete

Author: ffffwh

driver/common/common.go

@@ -28,9 +28,7 @@ const (
 )
 
 const (
-	TaskStateComplete int = iota
-	TaskStateRestart
-	TaskStateDead
+	TaskStateDead = 2
 )
 
 const (

driver/kafka/kafka3.go

@@ -641,9 +641,7 @@ func (kr *KafkaRunner) onError(state int, err error) {
 	}
 
 	switch state {
-	case common.TaskStateComplete:
-		kr.logger.Info("Done migrating")
-	case common.TaskStateRestart, common.TaskStateDead:
+	case common.TaskStateDead:
 		msg := &common.ControlMsg{
 			Msg:  err.Error(),
 			Type: common.ControlMsgError,

driver/mysql/applier.go

@@ -1025,9 +1025,7 @@ func (a *Applier) onError(state int, err error) {
 	}
 
 	switch state {
-	case common.TaskStateComplete:
-		a.logger.Info("Done migrating")
-	case common.TaskStateRestart, common.TaskStateDead:
+	case common.TaskStateDead:
 		msg := &common.ControlMsg{
 			Msg:  err.Error(),
 			Type: common.ControlMsgError,

driver/mysql/extractor.go

@@ -1140,7 +1140,7 @@ func (e *Extractor) sendSysVarAndSqlMode() error {
 		SqlMode:         setSqlMode,
 	}
 	if err := e.encodeAndSendDumpEntry(entry); err != nil {
-		e.onError(common.TaskStateRestart, err)
+		e.onError(common.TaskStateDead, err)
 	}
 
 	return nil
@@ -1318,7 +1318,7 @@ func (e *Extractor) mysqlDump() error {
 				if db.TableSchemaRename != "" {
 					dbSQL, err = base.RenameCreateSchemaAddINE(db.CreateSchemaString, db.TableSchemaRename)
 					if err != nil {
-						e.onError(common.TaskStateRestart, err)
+						e.onError(common.TaskStateDead, err)
 					}
 				} else {
 					dbSQL = db.CreateSchemaString
@@ -1331,7 +1331,7 @@ func (e *Extractor) mysqlDump() error {
 			atomic.AddInt64(&e.mysqlContext.RowsEstimate, 1)
 			atomic.AddInt64(&e.TotalRowsCopied, 1)
 			if err := e.encodeAndSendDumpEntry(entry); err != nil {
-				e.onError(common.TaskStateRestart, err)
+				e.onError(common.TaskStateDead, err)
 			}
 
 			for _, tbCtx := range db.TableMap {
@@ -1377,7 +1377,7 @@ func (e *Extractor) mysqlDump() error {
 				atomic.AddInt64(&e.mysqlContext.RowsEstimate, 1)
 				atomic.AddInt64(&e.TotalRowsCopied, 1)
 				if err := e.encodeAndSendDumpEntry(entry); err != nil {
-					e.onError(common.TaskStateRestart, err)
+					e.onError(common.TaskStateDead, err)
 				}
 			}
 			e.tableCount += len(db.TableMap)
@@ -1427,7 +1427,7 @@ func (e *Extractor) mysqlDump() error {
 						}
 					}
 					if err = e.encodeAndSendDumpEntry(entry); err != nil {
-						e.onError(common.TaskStateRestart, err)
+						e.onError(common.TaskStateDead, err)
 					}
 					atomic.AddInt64(&e.TotalRowsCopied, int64(len(entry.ValuesX)))
 					atomic.AddInt64(d.Memory, -memSize)

driver/oracle/extractor/extractor_oracle.go

@@ -909,7 +909,7 @@ func (e *ExtractorOracle) oracleDump() error {
 				DbSQL: dbSQL,
 			}
 			if err := e.encodeAndSendDumpEntry(entry); err != nil {
-				e.onError(common.TaskStateRestart, err)
+				e.onError(common.TaskStateDead, err)
 			}
 
 			for _, tb := range db.Tables {
@@ -938,7 +938,7 @@ func (e *ExtractorOracle) oracleDump() error {
 					// TotalCount: tb.Counter,
 				}
 				if err := e.encodeAndSendDumpEntry(entry); err != nil {
-					e.onError(common.TaskStateRestart, err)
+					e.onError(common.TaskStateDead, err)
 				}
 			}
 		}
@@ -966,7 +966,7 @@ func (e *ExtractorOracle) oracleDump() error {
 					e.onError(common.TaskStateDead, fmt.Errorf(entry.Err))
 				} else {
 					if err := e.encodeAndSendDumpEntry(entry); err != nil {
-						e.onError(common.TaskStateRestart, err)
+						e.onError(common.TaskStateDead, err)
 					}
 				}
 			}