Merge branch 'master' into rustin-patch-sink-flush-worker

pingcap · Mar 15, 2022 · e8287f5 · e8287f5
2 parents bca7121 + cf8f76b
commit e8287f5
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 2 deletions.
diff --git a/dm/syncer/dml_worker.go b/dm/syncer/dml_worker.go
@@ -249,7 +249,9 @@ func (w *DMLWorker) executeBatchJobs(queueID int, jobs []*job) {
 		time.Sleep(time.Duration(t) * time.Second)
 	})
 	// use background context to execute sqls as much as possible
-	ctx, cancel := w.syncCtx.WithTimeout(maxDMLExecutionDuration)
+	// set timeout to maxDMLConnectionDuration to make sure dmls can be replicated to downstream event if the latency is high
+	// if users need to quit this asap, we can support pause-task/stop-task --force in the future
+	ctx, cancel := w.syncCtx.WithTimeout(maxDMLConnectionDuration)
 	defer cancel()
 	affect, err = db.ExecuteSQL(ctx, queries, args...)
 	failpoint.Inject("SafeModeExit", func(val failpoint.Value) {

diff --git a/dm/syncer/syncer.go b/dm/syncer/syncer.go
@@ -86,7 +86,6 @@ var (
 	maxDDLConnectionTimeout = fmt.Sprintf("%dm", MaxDDLConnectionTimeoutMinute)
 
 	maxDMLConnectionDuration, _ = time.ParseDuration(maxDMLConnectionTimeout)
-	maxDMLExecutionDuration     = 30 * time.Second
 
 	defaultMaxPauseOrStopWaitTime = 10 * time.Second