add job config RetryTxLimit

Author: ffffwh

api/docs/docs.go

@@ -3017,6 +3017,9 @@ var doc = `{
                 "bulk_insert2": {
                     "type": "integer"
                 },
+                "retry_tx_limit": {
+                    "type": "integer"
+                },
                 "dependency_history_size": {
                     "type": "integer"
                 },

api/docs/swagger.json

@@ -3001,6 +3001,9 @@
                 "bulk_insert2": {
                     "type": "integer"
                 },
+                "retry_tx_limit": {
+                    "type": "integer"
+                },
                 "dependency_history_size": {
                     "type": "integer"
                 },
@@ -3756,4 +3759,4 @@
             "in": "header"
         }
     }
-}
+}

api/docs/swagger.yaml

@@ -591,6 +591,8 @@ definitions:
         type: integer
       bulk_insert2:
         type: integer
+      retry_tx_limit:
+        type: integer
       dependency_history_size:
         type: integer
       parallel_workers:

api/handler/v2/job.go

@@ -331,7 +331,7 @@ func createOrUpdateJob(logger g.LoggerType, jobParam *models.CreateOrUpdateMysql
 	if jobParam.SrcTask.GroupTimeout == 0 {
 		jobParam.SrcTask.GroupTimeout = common.DefaultSrcGroupTimeout
 	}
-	logger.Info("MysqlDestTaskConfig", jobParam.DestTask.MysqlDestTaskConfig)
+	logger.Info("MysqlDestTaskConfig", "config", jobParam.DestTask.MysqlDestTaskConfig)
 	if jobParam.DestTask.MysqlDestTaskConfig != nil {
 		if jobParam.DestTask.MysqlDestTaskConfig.ParallelWorkers == 0 {
 			jobParam.DestTask.MysqlDestTaskConfig.ParallelWorkers = common.DefaultNumWorkers
@@ -613,6 +613,7 @@ func buildDatabaseSrcTaskConfigMap(config *models.SrcTaskConfig, destConfig *mod
 		addNotRequiredParamToMap(taskConfigInNomadFormat, destConfig.MysqlDestTaskConfig.DependencyHistorySize, "DependencyHistorySize")
 		addNotRequiredParamToMap(taskConfigInNomadFormat, destConfig.MysqlDestTaskConfig.BulkInsert1, "BulkInsert1")
 		addNotRequiredParamToMap(taskConfigInNomadFormat, destConfig.MysqlDestTaskConfig.BulkInsert2, "BulkInsert2")
+		addNotRequiredParamToMap(taskConfigInNomadFormat, destConfig.MysqlDestTaskConfig.RetryTxLimit, "RetryTxLimit")
 		addNotRequiredParamToMap(taskConfigInNomadFormat, destConfig.MysqlDestTaskConfig.SetGtidNext, "SetGtidNext")
 		taskConfigInNomadFormat["DestConnectionConfig"] = buildMysqlConnectionConfigMap(destConfig.ConnectionConfig)
 	} else if kafkaConfig != nil {
@@ -883,6 +884,7 @@ func buildBasicTaskProfile(logger g.LoggerType, jobId string, srcTaskDetail *mod
 				DependencyHistorySize: destMySqlTaskDetail.TaskConfig.MysqlDestTaskConfig.DependencyHistorySize,
 				BulkInsert1:           destMySqlTaskDetail.TaskConfig.MysqlDestTaskConfig.BulkInsert1,
 				BulkInsert2:           destMySqlTaskDetail.TaskConfig.MysqlDestTaskConfig.BulkInsert2,
+				RetryTxLimit:          destMySqlTaskDetail.TaskConfig.MysqlDestTaskConfig.RetryTxLimit,
 				SetGtidNext:           destMySqlTaskDetail.TaskConfig.MysqlDestTaskConfig.SetGtidNext,
 			}
 			basicTaskProfile.Configuration.DstConfig = models.DstConfig{MysqlDestTaskConfig: mysqlDstConfig}
@@ -1060,6 +1062,7 @@ func buildMysqlDestTaskDetail(taskName string, internalTaskConfig common.DtleTas
 		DependencyHistorySize: internalTaskConfig.DependencyHistorySize,
 		BulkInsert1:           internalTaskConfig.BulkInsert1,
 		BulkInsert2:           internalTaskConfig.BulkInsert2,
+		RetryTxLimit:          internalTaskConfig.RetryTxLimit,
 		SetGtidNext:           internalTaskConfig.SetGtidNext,
 	}
 	destTaskDetail.TaskConfig = models.DestTaskConfig{

api/models/job_v2.go

@@ -182,6 +182,7 @@ type MysqlDestTaskConfig struct {
 	DependencyHistorySize int  `json:"dependency_history_size"`
 	BulkInsert1           int  `json:"bulk_insert1"`
 	BulkInsert2           int  `json:"bulk_insert2"`
+	RetryTxLimit          int  `json:"retry_tx_limit"`
 	SetGtidNext           bool `json:"set_gtid_next"`
 }
 

driver/common/taskconfig.go

@@ -57,6 +57,7 @@ type DtleTaskConfig struct {
 	BulkInsert1          int           `codec:"BulkInsert1"`
 	BulkInsert2          int           `codec:"BulkInsert2"`
 	BulkInsert3          int           `codec:"BulkInsert3"`
+	RetryTxLimit         int           `codec:"RetryTxLimit"`
 	SlaveNetWriteTimeout int           `codec:"SlaveNetWriteTimeout"`
 	BigTxSrcQueue        int32         `codec:"BigTxSrcQueue"`
 	TwoWaySync           bool          `codec:"TwoWaySync"`
@@ -116,7 +117,7 @@ type MySQLDriverConfig struct {
 	RowCopyStartTime time.Time
 	RowCopyEndTime   time.Time
 
-	Stage string
+	Stage        string
 }
 
 // ElapsedRowCopyTime returns time since starting to copy chunks of rows

driver/driver.go

@@ -163,6 +163,8 @@ var (
 			hclspec.NewLiteral(`8`)),
 		"BulkInsert3": hclspec.NewDefault(hclspec.NewAttr("BulkInsert3", "number", false),
 			hclspec.NewLiteral(`128`)),
+		"RetryTxLimit": hclspec.NewDefault(hclspec.NewAttr("RetryTxLimit", "number", false),
+			hclspec.NewLiteral(`3`)),
 		"BigTxSrcQueue": hclspec.NewDefault(hclspec.NewAttr("BigTxSrcQueue", "number", false),
 			hclspec.NewLiteral(`3`)),
 		"KafkaConfig": hclspec.NewBlock("KafkaConfig", false, hclspec.NewObject(map[string]*hclspec.Spec{

driver/mysql/applier_incr.go

@@ -234,12 +234,11 @@ func (a *ApplierIncr) MtsWorker(workerIndex int) {
 		case entryContext := <-a.applyBinlogMtsTxQueue:
 			hasEntry = true
 			logger.Debug("a binlogEntry MTS dequeue", "gno", entryContext.Entry.Coordinates.GetGNO())
-			const deadlockTryLimit = 3
 			for iTry := 0; ; iTry++ {
 				err := a.ApplyBinlogEvent(workerIndex, entryContext)
 				if err != nil {
 					if merr, isME := err.(*mysqldriver.MySQLError); isME {
-						if merr.Number == sql.ErrLockDeadlock && iTry < deadlockTryLimit {
+						if merr.Number == sql.ErrLockDeadlock && iTry < a.mysqlContext.RetryTxLimit {
 							logger.Info("found deadlock. will retry tx", "gno", entryContext.Entry.Coordinates.GetGNO(),
 								"iTry", iTry)
 							continue