Repository: pingcap/tiflow
Pull request #6126 (2022-06-30 04:55:32)

upstream(tcdc): cancel upstream wait group 
<!--
Thank you for contributing to TiFlow! 
Please read MD's [CONTRIBUTING](https://github.com/pingcap/tiflow/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve?
<!--
Please create an issue first to describe the problem.

There MUST be one line starting with "Issue Number:  " and 
linking the relevant issues via the "close" or "ref".

For more info, check https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue.
 -->

Issue Number: close #xxx

### What is changed and how it works?
 cancel upstream wait group , so upstream can be closed


### Check List <!--REMOVE the items that are not applicable-->

#### Tests <!-- At least one of them must be included. -->

 - Unit test
 - Integration test
 - Manual test (add detailed scripts or steps below)
 - No code

#### Questions <!-- Authors should answer these questions and reviewers should consider these questions. -->

##### Will it cause performance regression or break compatibility?

##### Do you need to update user documentation, design documentation or monitoring documentation?

### Release note <!-- bugfixes or new feature need a release note -->

```release-note
Please refer to [Release Notes Language Style Guide](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/release-notes-style-guide.html) to write a quality release note.

If you don't think this PR needs a release note then fill it with `None`.
```



Patch:
diff --git a/cdc/api/v2/api_helpers_mock.go b/cdc/api/v2/api_helpers_mock.go
index 7b69b16dd..a6d18a9d9 100644
--- a/cdc/api/v2/api_helpers_mock.go
+++ b/cdc/api/v2/api_helpers_mock.go
@@ -1,5 +1,5 @@
 // Code generated by MockGen. DO NOT EDIT.
-// Source: api_helpers.go
+// Source: cdc/api/v2/api_helpers.go
 
 // Package v2 is a generated GoMock package.
 package v2
diff --git a/cdc/owner/changefeed.go b/cdc/owner/changefeed.go
index deec8ddf1..7aec1e3b4 100644
--- a/cdc/owner/changefeed.go
+++ b/cdc/owner/changefeed.go
@@ -156,6 +156,12 @@ func (c *changefeed) Tick(ctx cdcContext.Context, state *orchestrator.Changefeed
 		c.handleErr(ctx, err)
 		return
 	}
+	if c.upstream.IsClosed() {
+		log.Panic("upstream is closed",
+			zap.Uint64("upstreamID", c.upstream.ID),
+			zap.String("namespace", c.id.Namespace),
+			zap.String("changefeed", c.id.ID))
+	}
 	// skip this tick
 	if !c.upstream.IsNormal() {
 		return
diff --git a/cdc/processor/processor.go b/cdc/processor/processor.go
index fa29acd85..e8062243f 100644
--- a/cdc/processor/processor.go
+++ b/cdc/processor/processor.go
@@ -475,6 +475,12 @@ func (p *processor) Tick(ctx cdcContext.Context, state *orchestrator.ChangefeedR
 	if err := p.upstream.Error(); err != nil {
 		return p.handleErr(ctx, state, err)
 	}
+	if p.upstream.IsClosed() {
+		log.Panic("upstream is closed",
+			zap.Uint64("upstreamID", p.upstream.ID),
+			zap.String("namespace", p.changefeedID.Namespace),
+			zap.String("changefeed", p.changefeedID.ID))
+	}
 	// skip this tick
 	if !p.upstream.IsNormal() {
 		log.Warn("upstream is not ready, skip",
diff --git a/dm/checker/checker.go b/dm/checker/checker.go
index 6ce515241..3b5bee703 100644
--- a/dm/checker/checker.go
+++ b/dm/checker/checker.go
@@ -216,7 +216,7 @@ func (c *Checker) Init(ctx context.Context) (err error) {
 				c.checkList = append(c.checkList, checker.NewOnlineDDLChecker(instance.sourceDB.DB, checkSchemas, c.onlineDDL, instance.baList))
 			}
 			if _, ok := c.checkingItems[config.BinlogDBChecking]; ok {
-				c.checkList = append(c.checkList, checker.NewBinlogDBChecker(instance.sourceDB.DB, instance.sourceDBinfo, checkSchemas, instance.cfg.CaseSensitive))
+				c.checkList = append(c.checkList, checker.NewBinlogDBChecker(instance.sourceDB, instance.sourceDBinfo, checkSchemas, instance.cfg.CaseSensitive))
 			}
 		}
 	}
diff --git a/dm/dm/config/subtask.go b/dm/dm/config/subtask.go
index 54390d1ea..6c8e37145 100644
--- a/dm/dm/config/subtask.go
+++ b/dm/dm/config/subtask.go
@@ -15,6 +15,8 @@ package config
 
 import (
 	"bytes"
+	"context"
+	"database/sql"
 	_ "embed"
 	"encoding/json"
 	"flag"
@@ -26,6 +28,7 @@ import (
 	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb-tools/pkg/column-mapping"
 	extstorage "github.com/pingcap/tidb/br/pkg/storage"
+	"github.com/pingcap/tidb/util/dbutil"
 	"github.com/pingcap/tidb/util/filter"
 	regexprrouter "github.com/pingcap/tidb/util/regexpr-router"
 	router "github.com/pingcap/tidb/util/table-router"
@@ -143,6 +146,16 @@ func (db *DBConfig) AdjustWithTimeZone(timeZone string) {
 	db.Adjust()
 }
 
+// FetchTimeZoneSetting fetch target db global time_zone setting.
+// TODO: move GetTimeZoneOffset and FormatTimeZoneOffset from TiDB to tiflow.
+func FetchTimeZoneSetting(ctx context.Context, db *sql.DB) (string, error) {
+	dur, err := dbutil.GetTimeZoneOffset(ctx, db)
+	if err != nil {
+		return "", err
+	}
+	return dbutil.FormatTimeZoneOffset(dur), nil
+}
+
 // Clone returns a deep copy of DBConfig. This function only fixes data race when adjusting Session.
 func (db *DBConfig) Clone() *DBConfig {
 	if db == nil {
diff --git a/dm/dm/worker/source_worker.go b/dm/dm/worker/source_worker.go
index 40838ede3..556dd0cbd 100644
--- a/dm/dm/worker/source_worker.go
+++ b/dm/dm/worker/source_worker.go
@@ -32,6 +32,7 @@ import (
 	"github.com/pingcap/tiflow/dm/dm/pb"
 	"github.com/pingcap/tiflow/dm/pkg/binlog"
 	"github.com/pingcap/tiflow/dm/pkg/conn"
+	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
 	"github.com/pingcap/tiflow/dm/pkg/etcdutil"
 	"github.com/pingcap/tiflow/dm/pkg/ha"
 	"github.com/pingcap/tiflow/dm/pkg/log"
@@ -268,10 +269,11 @@ func (w *SourceWorker) updateSourceStatus(ctx context.Context, needLock bool) er
 	}
 	w.sourceDBMu.Unlock()
 
+	// TODO: write NewSourceStatusFromDB, so engine can easily to use
 	var status binlog.SourceStatus
 	ctx, cancel := context.WithTimeout(ctx, utils.DefaultDBTimeout)
 	defer cancel()
-	pos, gtidSet, err := utils.GetPosAndGs(ctx, w.sourceDB.DB, cfg.Flavor)
+	pos, gtidSet, err := conn.GetPosAndGs(tcontext.NewContext(ctx, log.L()), w.sourceDB, cfg.Flavor)
 	if err != nil {
 		return err
 	}
diff --git a/dm/dumpling/dumpling.go b/dm/dumpling/dumpling.go
index 1bf2e9693..7ea069ea8 100644
--- a/dm/dumpling/dumpling.go
+++ b/dm/dumpling/dumpling.go
@@ -329,8 +329,13 @@ func (m *Dumpling) constructArgs(ctx context.Context) (*export.Config, error) {
 	tz := m.cfg.Timezone
 	if len(tz) == 0 {
 		// use target db time_zone as default
+		baseDB, err2 := conn.DefaultDBProvider.Apply(&m.cfg.To)
+		if err2 != nil {
+			return nil, err2
+		}
+		defer baseDB.Close()
 		var err1 error
-		tz, err1 = conn.FetchTimeZoneSetting(ctx, &m.cfg.To)
+		tz, err1 = config.FetchTimeZoneSetting(ctx, baseDB.DB)
 		if err1 != nil {
 			return nil, err1
 		}
diff --git a/dm/loader/lightning.go b/dm/loader/lightning.go
index f32a7278e..6922f23c1 100644
--- a/dm/loader/lightning.go
+++ b/dm/loader/lightning.go
@@ -149,8 +149,13 @@ func (l *LightningLoader) Init(ctx context.Context) (err error) {
 
 	timeZone := l.cfg.Timezone
 	if len(timeZone) == 0 {
+		baseDB, err2 := conn.DefaultDBProvider.Apply(&l.cfg.To)
+		if err2 != nil {
+			return err2
+		}
+		defer baseDB.Close()
 		var err1 error
-		timeZone, err1 = conn.FetchTimeZoneSetting(ctx, &l.cfg.To)
+		timeZone, err1 = config.FetchTimeZoneSetting(ctx, baseDB.DB)
 		if err1 != nil {
 			return err1
 		}
diff --git a/dm/loader/loader.go b/dm/loader/loader.go
index 671e2f04b..da37b8a42 100644
--- a/dm/loader/loader.go
+++ b/dm/loader/loader.go
@@ -540,8 +540,13 @@ func (l *Loader) Init(ctx context.Context) (err error) {
 	}
 	timeZone := l.cfg.Timezone
 	if len(timeZone) == 0 {
+		baseDB, err2 := conn.DefaultDBProvider.Apply(&l.cfg.To)
+		if err2 != nil {
+			return err2
+		}
+		defer baseDB.Close()
 		var err1 error
-		timeZone, err1 = conn.FetchTimeZoneSetting(ctx, &lcfg.To)
+		timeZone, err1 = config.FetchTimeZoneSetting(ctx, baseDB.DB)
 		if err1 != nil {
 			return err1
 		}
diff --git a/dm/pkg/checker/binlog.go b/dm/pkg/checker/binlog.go
index 23f505e80..ef9797e3b 100644
--- a/dm/pkg/checker/binlog.go
+++ b/dm/pkg/checker/binlog.go
@@ -20,6 +20,10 @@ import (
 	"strings"
 
 	"github.com/pingcap/tidb/util/dbutil"
+
+	"github.com/pingcap/tiflow/dm/pkg/conn"
+	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
+	"github.com/pingcap/tiflow/dm/pkg/log"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
 )
 
@@ -183,14 +187,14 @@ func (pc *MySQLBinlogRowImageChecker) Name() string {
 
 // BinlogDBChecker checks if migrated dbs are in binlog_do_db or binlog_ignore_db.
 type BinlogDBChecker struct {
-	db            *sql.DB
+	db            *conn.BaseDB
 	dbinfo        *dbutil.DBConfig
 	schemas       map[string]struct{}
 	caseSensitive bool
 }
 
 // NewBinlogDBChecker returns a RealChecker.
-func NewBinlogDBChecker(db *sql.DB, dbinfo *dbutil.DBConfig, schemas map[string]struct{}, caseSensitive bool) RealChecker {
+func NewBinlogDBChecker(db *conn.BaseDB, dbinfo *dbutil.DBConfig, schemas map[string]struct{}, caseSensitive bool) RealChecker {
 	newSchemas := make(map[string]struct{}, len(schemas))
 	for schema := range schemas {
 		newSchemas[schema] = struct{}{}
@@ -207,12 +211,13 @@ func (c *BinlogDBChecker) Check(ctx context.Context) *Result {
 		Extra: fmt.Sprintf("address of db instance - %s:%d", c.dbinfo.Host, c.dbinfo.Port),
 	}
 
-	flavor, err := utils.GetFlavor(ctx, c.db)
+	flavor, err := utils.GetFlavor(ctx, c.db.DB)
 	if err != nil {
 		markCheckError(result, err)
 		return result
 	}
-	binlogDoDB, binlogIgnoreDB, err := utils.GetBinlogDB(ctx, c.db, flavor)
+	tctx := tcontext.NewContext(ctx, log.L())
+	binlogDoDB, binlogIgnoreDB, err := conn.GetBinlogDB(tctx, c.db, flavor)
 	if err != nil {
 		markCheckError(result, err)
 		return result
diff --git a/dm/pkg/conn/basedb.go b/dm/pkg/conn/basedb.go
index 97e108703..07e47e695 100644
--- a/dm/pkg/conn/basedb.go
+++ b/dm/pkg/conn/basedb.go
@@ -170,6 +170,7 @@ func (d *BaseDB) GetBaseConn(ctx context.Context) (*BaseConn, error) {
 	return baseConn, nil
 }
 
+// TODO: retry can be done inside the BaseDB.
 func (d *BaseDB) ExecContext(tctx *tcontext.Context, query string, args ...interface{}) (sql.Result, error) {
 	if tctx.L().Core().Enabled(zap.DebugLevel) {
 		tctx.L().Debug("exec context",
@@ -179,6 +180,7 @@ func (d *BaseDB) ExecContext(tctx *tcontext.Context, query string, args ...inter
 	return d.DB.ExecContext(tctx.Ctx, query, args...)
 }
 
+// TODO: retry can be done inside the BaseDB.
 func (d *BaseDB) QueryContext(tctx *tcontext.Context, query string, args ...interface{}) (*sql.Rows, error) {
 	if tctx.L().Core().Enabled(zap.DebugLevel) {
 		tctx.L().Debug("query context",
diff --git a/dm/pkg/conn/utils.go b/dm/pkg/conn/utils.go
index 63e865367..7a1d3f877 100644
--- a/dm/pkg/conn/utils.go
+++ b/dm/pkg/conn/utils.go
@@ -14,24 +14,189 @@
 package conn
 
 import (
-	"context"
+	"fmt"
+	"strconv"
+	"strings"
+	"time"
 
-	"github.com/pingcap/errors"
-	"github.com/pingcap/tidb/util/dbutil"
+	gmysql "github.com/go-mysql-org/go-mysql/mysql"
+	"github.com/pingcap/failpoint"
+	tmysql "github.com/pingcap/tidb/parser/mysql"
+	"go.uber.org/zap"
 
-	"github.com/pingcap/tiflow/dm/dm/config"
+	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
+	"github.com/pingcap/tiflow/dm/pkg/gtid"
+	"github.com/pingcap/tiflow/dm/pkg/terror"
 )
 
-// FetchTimeZoneSetting fetch target db global time_zone setting.
-func FetchTimeZoneSetting(ctx context.Context, cfg *config.DBConfig) (string, error) {
-	db, err := DefaultDBProvider.Apply(cfg)
+const (
+	// DefaultDBTimeout represents a DB operation timeout for common usages.
+	DefaultDBTimeout = 30 * time.Second
+)
+
+// GetGlobalVariable gets server's global variable.
+func GetGlobalVariable(ctx *tcontext.Context, db *BaseDB, variable string) (value string, err error) {
+	failpoint.Inject("GetGlobalVariableFailed", func(val failpoint.Value) {
+		items := strings.Split(val.(string), ",")
+		if len(items) != 2 {
+			ctx.L().Fatal("failpoint GetGlobalVariableFailed's value is invalid", zap.String("val", val.(string)))
+		}
+		variableName := items[0]
+		errCode, err1 := strconv.ParseUint(items[1], 10, 16)
+		if err1 != nil {
+			ctx.L().Fatal("failpoint GetGlobalVariableFailed's value is invalid", zap.String("val", val.(string)))
+		}
+		if variable == variableName {
+			err = tmysql.NewErr(uint16(errCode))
+			ctx.L().Warn("GetGlobalVariable failed", zap.String("variable", variable), zap.String("failpoint", "GetGlobalVariableFailed"), zap.Error(err))
+			failpoint.Return("", terror.DBErrorAdapt(err, terror.ErrDBDriverError))
+		}
+	})
+
+	conn, err := db.GetBaseConn(ctx.Context())
+	if err != nil {
+		return "", terror.DBErrorAdapt(err, terror.ErrDBDriverError)
+	}
+	defer conn.close()
+	return getVariable(ctx, conn, variable, true)
+}
+
+func getVariable(ctx *tcontext.Context, conn *BaseConn, variable string, isGlobal bool) (value string, err error) {
+	var template string
+	if isGlobal {
+		template = "SHOW GLOBAL VARIABLES LIKE '%s'"
+	} else {
+		template = "SHOW VARIABLES LIKE '%s'"
+	}
+	query := fmt.Sprintf(template, variable)
+	row, err := conn.QuerySQL(ctx, query)
 	if err != nil {
 		return "", err
 	}
-	defer db.Close()
-	dur, err := dbutil.GetTimeZoneOffset(ctx, db.DB)
+	defer row.Close()
+
+	// Show an example.
+	/*
+		mysql> SHOW GLOBAL VARIABLES LIKE "binlog_format";
+		+---------------+-------+
+		| Variable_name | Value |
+		+---------------+-------+
+		| binlog_format | ROW   |
+		+---------------+-------+
+	*/
+
+	if !row.Next() {
+		return "", terror.ErrDBDriverError.Generatef("variable %s not found", variable)
+	}
+
+	err = row.Scan(&variable, &value)
 	if err != nil {
-		return "", errors.Trace(err)
+		return "", terror.DBErrorAdapt(err, terror.ErrDBDriverError)
+	}
+	return value, nil
+}
+
+// GetMasterStatus gets status from master.
+// When the returned error is nil, the gtid must be not nil.
+func GetMasterStatus(ctx *tcontext.Context, db *BaseDB, flavor string) (
+	string, uint32, string, string, string, error,
+) {
+	var (
+		binlogName     string
+		pos            uint32
+		binlogDoDB     string
+		binlogIgnoreDB string
+		gtidStr        string
+		err            error
+	)
+	// need REPLICATION SLAVE privilege
+	rows, err := db.QueryContext(ctx, `SHOW MASTER STATUS`)
+	if err != nil {
+		err = terror.DBErrorAdapt(err, terror.ErrDBDriverError)
+		return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
+	}
+	defer rows.Close()
+
+	if !rows.Next() {
+		err = terror.ErrNoMasterStatus.Generate()
+		return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
 	}
-	return dbutil.FormatTimeZoneOffset(dur), nil
+	// Show an example.
+	/*
+		MySQL [test]> SHOW MASTER STATUS;
+		+-----------+----------+--------------+------------------+--------------------------------------------+
+		| File      | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set                          |
+		+-----------+----------+--------------+------------------+--------------------------------------------+
+		| ON.000001 |     4822 |              |                  | 85ab69d1-b21f-11e6-9c5e-64006a8978d2:1-46
+		+-----------+----------+--------------+------------------+--------------------------------------------+
+	*/
+	/*
+		For MariaDB,SHOW MASTER STATUS:
+		+--------------------+----------+--------------+------------------+
+		| File               | Position | Binlog_Do_DB | Binlog_Ignore_DB |
+		+--------------------+----------+--------------+------------------+
+		| mariadb-bin.000016 |      475 |              |                  |
+		+--------------------+----------+--------------+------------------+
+		SELECT @@global.gtid_binlog_pos;
+		+--------------------------+
+		| @@global.gtid_binlog_pos |
+		+--------------------------+
+		| 0-1-2                    |
+		+--------------------------+
+	*/
+	if flavor == gmysql.MySQLFlavor {
+		err = rows.Scan(&binlogName, &pos, &binlogDoDB, &binlogIgnoreDB, &gtidStr)
+	} else {
+		err = rows.Scan(&binlogName, &pos, &binlogDoDB, &binlogIgnoreDB)
+	}
+	if err != nil {
+		err = terror.DBErrorAdapt(err, terror.ErrDBDriverError)
+		return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
+	}
+	if flavor == gmysql.MariaDBFlavor {
+		gtidStr, err = GetGlobalVariable(ctx, db, "gtid_binlog_pos")
+		if err != nil {
+			return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
+		}
+	}
+
+	if rows.Next() {
+		ctx.L().Warn("SHOW MASTER STATUS returns more than one row, will only use first row")
+	}
+	if rows.Close() != nil {
+		err = terror.DBErrorAdapt(rows.Err(), terror.ErrDBDriverError)
+		return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
+	}
+	if rows.Err() != nil {
+		err = terror.DBErrorAdapt(rows.Err(), terror.ErrDBDriverError)
+		return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
+	}
+
+	return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
+}
+
+// GetPosAndGs get binlog position and gmysql.GTIDSet from `show master status`.
+func GetPosAndGs(ctx *tcontext.Context, db *BaseDB, flavor string) (
+	binlogPos gmysql.Position,
+	gs gmysql.GTIDSet,
+	err error,
+) {
+	binlogName, pos, _, _, gtidStr, err := GetMasterStatus(ctx, db, flavor)
+	if err != nil {
+		return
+	}
+	binlogPos = gmysql.Position{
+		Name: binlogName,
+		Pos:  pos,
+	}
+
+	gs, err = gtid.ParserGTID(flavor, gtidStr)
+	return
+}
+
+// GetBinlogDB get binlog_do_db and binlog_ignore_db from `show master status`.
+func GetBinlogDB(ctx *tcontext.Context, db *BaseDB, flavor string) (string, string, error) {
+	// nolint:dogsled
+	_, _, binlogDoDB, binlogIgnoreDB, _, err := GetMasterStatus(ctx, db, flavor)
+	return binlogDoDB, binlogIgnoreDB, err
 }
diff --git a/dm/pkg/utils/db.go b/dm/pkg/utils/db.go
index 5a89af11b..9ac9af862 100644
--- a/dm/pkg/utils/db.go
+++ b/dm/pkg/utils/db.go
@@ -33,7 +33,6 @@ import (
 	"github.com/pingcap/tidb/util/dbutil"
 	"go.uber.org/zap"
 
-	"github.com/pingcap/tiflow/dm/pkg/gtid"
 	"github.com/pingcap/tiflow/dm/pkg/log"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 )
@@ -168,125 +167,6 @@ func GetSlaveServerID(ctx context.Context, db *sql.DB) (map[uint32]struct{}, err
 	return serverIDs, nil
 }
 
-// GetPosAndGs get binlog position and gmysql.GTIDSet from `show master status`.
-func GetPosAndGs(ctx context.Context, db *sql.DB, flavor string) (
-	binlogPos gmysql.Position,
-	gs gmysql.GTIDSet,
-	err error,
-) {
-	binlogName, pos, _, _, gtidStr, err := GetMasterStatus(ctx, db, flavor)
-	if err != nil {
-		return
-	}
-	binlogPos = gmysql.Position{
-		Name: binlogName,
-		Pos:  pos,
-	}
-
-	gs, err = gtid.ParserGTID(flavor, gtidStr)
-	return
-}
-
-// GetBinlogDB get binlog_do_db and binlog_ignore_db from `show master status`.
-func GetBinlogDB(ctx context.Context, db *sql.DB, flavor string) (string, string, error) {
-	// nolint:dogsled
-	_, _, binlogDoDB, binlogIgnoreDB, _, err := GetMasterStatus(ctx, db, flavor)
-	return binlogDoDB, binlogIgnoreDB, err
-}
-
-// GetMasterStatus gets status from master.
-// When the returned error is nil, the gtid must be not nil.
-func GetMasterStatus(ctx context.Context, db *sql.DB, flavor string) (
-	string, uint32, string, string, string, error,
-) {
-	var (
-		binlogName     string
-		pos            uint32
-		binlogDoDB     string
-		binlogIgnoreDB string
-		gtidStr        string
-		err            error
-	)
-	// need REPLICATION SLAVE privilege
-	rows, err := db.QueryContext(ctx, `SHOW MASTER STATUS`)
-	if err != nil {
-		err = terror.DBErrorAdapt(err, terror.ErrDBDriverError)
-		return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
-	}
-	defer rows.Close()
-
-	if !rows.Next() {
-		err = terror.ErrNoMasterStatus.Generate()
-		return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
-	}
-	// Show an example.
-	/*
-		MySQL [test]> SHOW MASTER STATUS;
-		+-----------+----------+--------------+------------------+--------------------------------------------+
-		| File      | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set                          |
-		+-----------+----------+--------------+------------------+--------------------------------------------+
-		| ON.000001 |     4822 |              |                  | 85ab69d1-b21f-11e6-9c5e-64006a8978d2:1-46
-		+-----------+----------+--------------+------------------+--------------------------------------------+
-	*/
-	/*
-		For MariaDB,SHOW MASTER STATUS:
-		+--------------------+----------+--------------+------------------+
-		| File               | Position | Binlog_Do_DB | Binlog_Ignore_DB |
-		+--------------------+----------+--------------+------------------+
-		| mariadb-bin.000016 |      475 |              |                  |
-		+--------------------+----------+--------------+------------------+
-		SELECT @@global.gtid_binlog_pos;
-		+--------------------------+
-		| @@global.gtid_binlog_pos |
-		+--------------------------+
-		| 0-1-2                    |
-		+--------------------------+
-	*/
-	if flavor == gmysql.MySQLFlavor {
-		err = rows.Scan(&binlogName, &pos, &binlogDoDB, &binlogIgnoreDB, &gtidStr)
-	} else {
-		err = rows.Scan(&binlogName, &pos, &binlogDoDB, &binlogIgnoreDB)
-	}
-	if err != nil {
-		err = terror.DBErrorAdapt(err, terror.ErrDBDriverError)
-		return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
-	}
-	if flavor == gmysql.MariaDBFlavor {
-		gtidStr, err = GetGlobalVariable(ctx, db, "gtid_binlog_pos")
-		if err != nil {
-			return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
-		}
-	}
-
-	if rows.Next() {
-		log.L().Warn("SHOW MASTER STATUS returns more than one row, will only use first row")
-	}
-	if rows.Close() != nil {
-		err = terror.DBErrorAdapt(rows.Err(), terror.ErrDBDriverError)
-		return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
-	}
-	if rows.Err() != nil {
-		err = terror.DBErrorAdapt(rows.Err(), terror.ErrDBDriverError)
-		return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
-	}
-
-	return binlogName, pos, binlogDoDB, binlogIgnoreDB, gtidStr, err
-}
-
-// GetMariaDBGTID gets MariaDB's `gtid_binlog_pos`
-// it can not get by `SHOW MASTER STATUS`.
-func GetMariaDBGTID(ctx context.Context, db *sql.DB) (gmysql.GTIDSet, error) {
-	gtidStr, err := GetGlobalVariable(ctx, db, "gtid_binlog_pos")
-	if err != nil {
-		return nil, err
-	}
-	gs, err := gtid.ParserGTID(gmysql.MariaDBFlavor, gtidStr)
-	if err != nil {
-		return nil, err
-	}
-	return gs, nil
-}
-
 // GetGlobalVariable gets server's global variable.
 func GetGlobalVariable(ctx context.Context, db *sql.DB, variable string) (value string, err error) {
 	failpoint.Inject("GetGlobalVariableFailed", func(val failpoint.Value) {
diff --git a/dm/relay/relay.go b/dm/relay/relay.go
index d88c35a0a..2a15bc683 100644
--- a/dm/relay/relay.go
+++ b/dm/relay/relay.go
@@ -39,6 +39,7 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/binlog/common"
 	binlogReader "github.com/pingcap/tiflow/dm/pkg/binlog/reader"
 	"github.com/pingcap/tiflow/dm/pkg/conn"
+	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
 	"github.com/pingcap/tiflow/dm/pkg/gtid"
 	"github.com/pingcap/tiflow/dm/pkg/log"
 	parserpkg "github.com/pingcap/tiflow/dm/pkg/parser"
@@ -596,7 +597,8 @@ func (r *Relay) handleEvents(
 					r.logger.Error("the requested binlog files have purged in the master server or the master server have switched, currently DM do no support to handle this error",
 						zap.String("db host", cfg.Host), zap.Int("db port", cfg.Port), zap.Stringer("last pos", lastPos), log.ShortError(err))
 					// log the status for debug
-					pos, gs, err2 := utils.GetPosAndGs(ctx, r.db.DB, r.cfg.Flavor)
+					tctx := tcontext.NewContext(ctx, r.logger)
+					pos, gs, err2 := conn.GetPosAndGs(tctx, r.db, r.cfg.Flavor)
 					if err2 == nil {
 						r.logger.Info("current master status", zap.Stringer("position", pos), log.WrapStringerField("GTID sets", gs))
 					}
@@ -782,7 +784,8 @@ func (r *Relay) reSetupMeta(ctx context.Context) error {
 
 	var latestPosName, latestGTIDStr string
 	if (r.cfg.EnableGTID && len(r.cfg.BinlogGTID) == 0) || (!r.cfg.EnableGTID && len(r.cfg.BinLogName) == 0) {
-		latestPos, latestGTID, err2 := utils.GetPosAndGs(ctx, r.db.DB, r.cfg.Flavor)
+		tctx := tcontext.NewContext(ctx, r.logger)
+		latestPos, latestGTID, err2 := conn.GetPosAndGs(tctx, r.db, r.cfg.Flavor)
 		if err2 != nil {
 			return err2
 		}
@@ -862,7 +865,8 @@ func (r *Relay) doIntervalOps(ctx context.Context) {
 				return
 			}
 			ctx2, cancel2 := context.WithTimeout(ctx, utils.DefaultDBTimeout)
-			pos, _, err := utils.GetPosAndGs(ctx2, r.db.DB, r.cfg.Flavor)
+			tctx := tcontext.NewContext(ctx2, r.logger)
+			pos, _, err := conn.GetPosAndGs(tctx, r.db, r.cfg.Flavor)
 			cancel2()
 			if err != nil {
 				r.logger.Warn("get master status", zap.Error(err))
diff --git a/dm/syncer/dbconn/db.go b/dm/syncer/dbconn/db.go
index c2d847f0e..674ed9529 100644
--- a/dm/syncer/dbconn/db.go
+++ b/dm/syncer/dbconn/db.go
@@ -20,17 +20,17 @@ import (
 
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/util/dbutil"
-	"github.com/pingcap/tiflow/dm/pkg/metricsproxy"
-	"github.com/pingcap/tiflow/dm/syncer/metrics"
 	"go.uber.org/zap"
 
 	"github.com/pingcap/tiflow/dm/dm/config"
 	"github.com/pingcap/tiflow/dm/pkg/conn"
 	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
 	"github.com/pingcap/tiflow/dm/pkg/log"
+	"github.com/pingcap/tiflow/dm/pkg/metricsproxy"
 	"github.com/pingcap/tiflow/dm/pkg/retry"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
+	"github.com/pingcap/tiflow/dm/syncer/metrics"
 )
 
 var retryTimeout = 3 * time.Second
diff --git a/dm/syncer/dbconn/upstream_db.go b/dm/syncer/dbconn/upstream_db.go
index cdb48dc8f..cd61c022f 100644
--- a/dm/syncer/dbconn/upstream_db.go
+++ b/dm/syncer/dbconn/upstream_db.go
@@ -50,8 +50,8 @@ func NewUpStreamConn(dbCfg *config.DBConfig) (*UpStreamConn, error) {
 }
 
 // GetMasterStatus returns binlog location that extracted from SHOW MASTER STATUS.
-func (conn *UpStreamConn) GetMasterStatus(ctx context.Context, flavor string) (mysql.Position, mysql.GTIDSet, error) {
-	pos, gtidSet, err := utils.GetPosAndGs(ctx, conn.BaseDB.DB, flavor)
+func (c *UpStreamConn) GetMasterStatus(ctx *tcontext.Context, flavor string) (mysql.Position, mysql.GTIDSet, error) {
+	pos, gtidSet, err := conn.GetPosAndGs(ctx, c.BaseDB, flavor)
 
 	failpoint.Inject("GetMasterStatusFailed", func(val failpoint.Value) {
 		err = tmysql.NewErr(uint16(val.(int)))
@@ -62,13 +62,13 @@ func (conn *UpStreamConn) GetMasterStatus(ctx context.Context, flavor string) (m
 }
 
 // GetServerUUID returns upstream server UUID.
-func (conn *UpStreamConn) GetServerUUID(ctx context.Context, flavor string) (string, error) {
-	return utils.GetServerUUID(ctx, conn.BaseDB.DB, flavor)
+func (c *UpStreamConn) GetServerUUID(ctx context.Context, flavor string) (string, error) {
+	return utils.GetServerUUID(ctx, c.BaseDB.DB, flavor)
 }
 
 // GetServerUnixTS returns the result of current timestamp in upstream.
-func (conn *UpStreamConn) GetServerUnixTS(ctx context.Context) (int64, error) {
-	return utils.GetServerUnixTS(ctx, conn.BaseDB.DB)
+func (c *UpStreamConn) GetServerUnixTS(ctx context.Context) (int64, error) {
+	return utils.GetServerUnixTS(ctx, c.BaseDB.DB)
 }
 
 // GetCharsetAndCollationInfo returns charset and collation info.
@@ -117,18 +117,18 @@ func GetCharsetAndCollationInfo(tctx *tcontext.Context, conn *DBConn) (map[strin
 }
 
 // GetParser returns the parser with correct flag for upstream.
-func (conn *UpStreamConn) GetParser(ctx context.Context) (*parser.Parser, error) {
-	return utils.GetParser(ctx, conn.BaseDB.DB)
+func (c *UpStreamConn) GetParser(ctx context.Context) (*parser.Parser, error) {
+	return utils.GetParser(ctx, c.BaseDB.DB)
 }
 
 // KillConn kills a connection in upstream.
-func (conn *UpStreamConn) KillConn(ctx context.Context, connID uint32) error {
-	return utils.KillConn(ctx, conn.BaseDB.DB, connID)
+func (c *UpStreamConn) KillConn(ctx context.Context, connID uint32) error {
+	return utils.KillConn(ctx, c.BaseDB.DB, connID)
 }
 
 // FetchAllDoTables returns tables matches allow-list.
-func (conn *UpStreamConn) FetchAllDoTables(ctx context.Context, bw *filter.Filter) (map[string][]string, error) {
-	return utils.FetchAllDoTables(ctx, conn.BaseDB.DB, bw)
+func (c *UpStreamConn) FetchAllDoTables(ctx context.Context, bw *filter.Filter) (map[string][]string, error) {
+	return utils.FetchAllDoTables(ctx, c.BaseDB.DB, bw)
 }
 
 // CloseUpstreamConn closes the UpStreamConn.
diff --git a/dm/syncer/relay.go b/dm/syncer/relay.go
index 5325dd628..a96f845c5 100644
--- a/dm/syncer/relay.go
+++ b/dm/syncer/relay.go
@@ -63,7 +63,7 @@ func (s *Syncer) setInitActiveRelayLog(ctx context.Context) error {
 		}
 	default:
 		// start from dumper or loader, get current pos from master
-		pos, _, err = s.fromDB.GetMasterStatus(ctx, s.cfg.Flavor)
+		pos, _, err = s.fromDB.GetMasterStatus(s.tctx.WithContext(ctx), s.cfg.Flavor)
 		if err != nil {
 			return terror.Annotatef(err, "get master status")
 		}
diff --git a/dm/syncer/syncer.go b/dm/syncer/syncer.go
index 2eae19f51..2d55c3410 100644
--- a/dm/syncer/syncer.go
+++ b/dm/syncer/syncer.go
@@ -1939,7 +1939,7 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
 		failpoint.Inject("FakeRedirect", func(val failpoint.Value) {
 			if len(shardingReSyncCh) == 0 && shardingReSync == nil {
 				if strVal, ok := val.(string); ok {
-					pos, gtidSet, err2 := s.fromDB.GetMasterStatus(ctx, s.cfg.Flavor)
+					pos, gtidSet, err2 := s.fromDB.GetMasterStatus(s.tctx.WithContext(ctx), s.cfg.Flavor)
 					if err2 != nil {
 						s.tctx.L().Error("fail to get master status in failpoint FakeRedirect", zap.Error(err2))
 						os.Exit(1)
diff --git a/dm/syncer/util.go b/dm/syncer/util.go
index 0871aa772..279196696 100644
--- a/dm/syncer/util.go
+++ b/dm/syncer/util.go
@@ -109,7 +109,12 @@ func printServerVersion(tctx *tcontext.Context, db *conn.BaseDB, scope string) {
 func str2TimezoneOrFromDB(tctx *tcontext.Context, tzStr string, dbCfg *config.DBConfig) (*time.Location, error) {
 	var err error
 	if len(tzStr) == 0 {
-		tzStr, err = conn.FetchTimeZoneSetting(tctx.Ctx, dbCfg)
+		baseDB, err2 := conn.DefaultDBProvider.Apply(dbCfg)
+		if err2 != nil {
+			return nil, err2
+		}
+		defer baseDB.Close()
+		tzStr, err = config.FetchTimeZoneSetting(tctx.Ctx, baseDB.DB)
 		if err != nil {
 			return nil, err
 		}
diff --git a/engine/pkg/rpcutil/client.go b/engine/pkg/rpcutil/client.go
index 39fc83a9d..9fd57f365 100644
--- a/engine/pkg/rpcutil/client.go
+++ b/engine/pkg/rpcutil/client.go
@@ -20,10 +20,13 @@ import (
 
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/pkg/errors"
+	"github.com/pingcap/tiflow/pkg/retry"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
 )
 
+const defaultDialRetry = 3
+
 // CloseableConnIface defines an interface that supports Close(release resource)
 type CloseableConnIface interface {
 	Close() error
@@ -98,11 +101,20 @@ func (c *FailoverRPCClients[T]) UpdateClients(ctx context.Context, urls []string
 
 		if _, ok := c.clients[addr]; !ok {
 			log.L().Info("add new server master client", zap.String("addr", addr))
-			cli, conn, err := c.dialer(ctx, addr)
+
+			var (
+				cli  T
+				conn CloseableConnIface
+			)
+			err := retry.Do(ctx, func() (err error) {
+				cli, conn, err = c.dialer(ctx, addr)
+				return err
+			}, retry.WithMaxTries(defaultDialRetry))
 			if err != nil {
 				log.L().Warn("dial to server master failed", zap.String("addr", addr), zap.Error(err))
 				continue
 			}
+
 			c.clients[addr] = &clientHolder[T]{
 				conn:   conn,
 				client: cli,
diff --git a/pkg/upstream/manager.go b/pkg/upstream/manager.go
index cba6ee6be..4b3f970a9 100644
--- a/pkg/upstream/manager.go
+++ b/pkg/upstream/manager.go
@@ -75,6 +75,7 @@ func NewManager4Test(pdClient pd.Client) *Manager {
 		ups: new(sync.Map), ctx: context.Background(),
 		gcServiceID:     etcd.GcServiceIDForTest(),
 		defaultUpstream: up,
+		cancel:          func() {},
 	}
 	up.isDefaultUpstream = true
 	res.ups.Store(testUpstreamID, up)
@@ -162,6 +163,7 @@ func (m *Manager) Close() {
 	m.cancel()
 	m.ups.Range(func(k, v interface{}) bool {
 		v.(*Upstream).Close()
+		m.ups.Delete(k)
 		return true
 	})
 }
@@ -214,14 +216,9 @@ func (m *Manager) Tick(ctx context.Context,
 		log.Info("no active changefeed found, try to close upstream",
 			zap.Uint64("id", up.ID))
 		if up.shouldClose() {
-			log.Info("upstream should be closed",
+			log.Info("upstream should be closed ,remove it from manager",
 				zap.Uint64("id", up.ID))
 			go up.Close()
-		}
-
-		if up.IsClosed() {
-			log.Info("upstream is closed, remove it from manager",
-				zap.Uint64("id", up.ID))
 			m.ups.Delete(id)
 		}
 		return true
diff --git a/pkg/upstream/upstream.go b/pkg/upstream/upstream.go
index a337ec09e..312db3535 100644
--- a/pkg/upstream/upstream.go
+++ b/pkg/upstream/upstream.go
@@ -66,7 +66,9 @@ type Upstream struct {
 	RegionCache *tikv.RegionCache
 	PDClock     pdutil.Clock
 	GCManager   gc.Manager
-	mu          sync.Mutex
+	// Only use in Close().
+	cancel func()
+	mu     sync.Mutex
 	// record the time when Upstream.hc becomes zero.
 	idleTime time.Time
 	// use clock to facilitate unit test
@@ -102,6 +104,7 @@ func NewUpstream4Test(pdClient pd.Client) *Upstream {
 		wg:             new(sync.WaitGroup),
 		clock:          clock.New(),
 		SecurityConfig: &config.SecurityConfig{},
+		cancel:         func() {},
 	}
 
 	return res
@@ -110,6 +113,8 @@ func NewUpstream4Test(pdClient pd.Client) *Upstream {
 // init initializes the upstream
 func initUpstream(ctx context.Context, up *Upstream, gcServiceID string) error {
 	log.Info("upstream is initializing", zap.Uint64("upstreamID", up.ID))
+	ctx, cancel := context.WithCancel(ctx)
+	up.cancel = cancel
 	grpcTLSOption, err := up.SecurityConfig.ToGRPCDialOption()
 	if err != nil {
 		up.err.Store(err)
@@ -207,7 +212,7 @@ func initUpstream(ctx context.Context, up *Upstream, gcServiceID string) error {
 func (up *Upstream) Close() {
 	up.mu.Lock()
 	up.mu.Unlock()
-
+	up.cancel()
 	if atomic.LoadInt32(&up.status) == closed ||
 		atomic.LoadInt32(&up.status) == closing {
 		return


Test patch:
diff --git a/dm/dm/config/subtask_test.go b/dm/dm/config/subtask_test.go
index a18cd4206..fc51edcad 100644
--- a/dm/dm/config/subtask_test.go
+++ b/dm/dm/config/subtask_test.go
@@ -14,11 +14,12 @@
 package config
 
 import (
+	"context"
 	"reflect"
 
+	"github.com/DATA-DOG/go-sqlmock"
 	. "github.com/pingcap/check"
 	"github.com/pingcap/tidb/util/filter"
-
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 )
 
@@ -316,3 +317,14 @@ func (t *testConfig) TestDBConfigClone(c *C) {
 	c.Assert(a, DeepEquals, b)
 	c.Assert(a.Security, Not(Equals), b.Security)
 }
+
+func (t *testConfig) TestFetchTZSetting(c *C) {
+	db, mock, err := sqlmock.New()
+	c.Assert(err, IsNil)
+
+	mock.ExpectQuery("SELECT cast\\(TIMEDIFF\\(NOW\\(6\\), UTC_TIMESTAMP\\(6\\)\\) as time\\);").
+		WillReturnRows(mock.NewRows([]string{""}).AddRow("01:00:00"))
+	tz, err := FetchTimeZoneSetting(context.Background(), db)
+	c.Assert(err, IsNil)
+	c.Assert(tz, Equals, "+01:00")
+}
diff --git a/dm/dm/worker/subtask_test.go b/dm/dm/worker/subtask_test.go
index ccaaf8b72..9b2b4b1ab 100644
--- a/dm/dm/worker/subtask_test.go
+++ b/dm/dm/worker/subtask_test.go
@@ -20,6 +20,11 @@ import (
 	"time"
 
 	"github.com/go-mysql-org/go-mysql/mysql"
+	. "github.com/pingcap/check"
+	"github.com/pingcap/errors"
+	"github.com/stretchr/testify/require"
+	clientv3 "go.etcd.io/etcd/client/v3"
+
 	"github.com/pingcap/tiflow/dm/dm/config"
 	"github.com/pingcap/tiflow/dm/dm/pb"
 	"github.com/pingcap/tiflow/dm/dm/unit"
@@ -30,11 +35,6 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/utils"
 	"github.com/pingcap/tiflow/dm/relay"
 	"github.com/pingcap/tiflow/dm/syncer"
-	"github.com/stretchr/testify/require"
-
-	. "github.com/pingcap/check"
-	"github.com/pingcap/errors"
-	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
 const (
diff --git a/dm/pkg/checker/binlog_test.go b/dm/pkg/checker/binlog_test.go
index 6a7475237..35186e4d0 100644
--- a/dm/pkg/checker/binlog_test.go
+++ b/dm/pkg/checker/binlog_test.go
@@ -19,6 +19,8 @@ import (
 	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/pingcap/tidb/util/dbutil"
 	"github.com/stretchr/testify/require"
+
+	"github.com/pingcap/tiflow/dm/pkg/conn"
 )
 
 func TestBinlogDB(t *testing.T) {
@@ -115,7 +117,7 @@ func TestBinlogDB(t *testing.T) {
 	}
 
 	for _, cs := range cases {
-		binlogDBChecker := NewBinlogDBChecker(db, &dbutil.DBConfig{}, cs.schemas, cs.caseSensitive)
+		binlogDBChecker := NewBinlogDBChecker(conn.NewBaseDB(db), &dbutil.DBConfig{}, cs.schemas, cs.caseSensitive)
 		versionRow := sqlmock.NewRows([]string{"Variable_name", "Value"}).AddRow("version", "mysql")
 		masterStatusRow := sqlmock.NewRows([]string{"File", "Position", "Binlog_Do_DB", "Binlog_Ignore_DB", "Executed_Gtid_Set"}).
 			AddRow("", 0, cs.doDB, cs.ignoreDB, "")
diff --git a/dm/pkg/conn/utils_test.go b/dm/pkg/conn/utils_test.go
index b366fb251..e031deda1 100644
--- a/dm/pkg/conn/utils_test.go
+++ b/dm/pkg/conn/utils_test.go
@@ -15,22 +15,48 @@ package conn
 
 import (
 	"context"
+	"testing"
 
-	. "github.com/pingcap/check"
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/stretchr/testify/require"
 
-	"github.com/pingcap/tiflow/dm/dm/config"
+	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
+	"github.com/pingcap/tiflow/dm/pkg/log"
 )
 
-var _ = Suite(testUtilSuite{})
-
-type testUtilSuite struct{}
-
-func (s testUtilSuite) TestFetchTZSetting(c *C) {
-	m := InitMockDB(c)
-
-	m.ExpectQuery("SELECT cast\\(TIMEDIFF\\(NOW\\(6\\), UTC_TIMESTAMP\\(6\\)\\) as time\\);").
-		WillReturnRows(m.NewRows([]string{""}).AddRow("01:00:00"))
-	tz, err := FetchTimeZoneSetting(context.Background(), &config.DBConfig{})
-	c.Assert(err, IsNil)
-	c.Assert(tz, Equals, "+01:00")
+func TestGetBinlogDB(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), DefaultDBTimeout)
+	defer cancel()
+
+	tctx := tcontext.NewContext(ctx, log.L())
+
+	db, mock, err := sqlmock.New()
+	require.NoError(t, err)
+	baseDB := NewBaseDB(db)
+
+	// 5 columns for MySQL
+	rows := mock.NewRows([]string{"File", "Position", "Binlog_Do_DB", "Binlog_Ignore_DB", "Executed_Gtid_Set"}).AddRow(
+		"mysql-bin.000009", 11232, "do_db", "ignore_db", "074be7f4-f0f1-11ea-95bd-0242ac120002:1-699",
+	)
+	mock.ExpectQuery(`SHOW MASTER STATUS`).WillReturnRows(rows)
+
+	binlogDoDB, binlogIgnoreDB, err := GetBinlogDB(tctx, baseDB, "mysql")
+	require.Nil(t, err)
+	require.Equal(t, binlogDoDB, "do_db")
+	require.Equal(t, binlogIgnoreDB, "ignore_db")
+	require.Nil(t, mock.ExpectationsWereMet())
+
+	// 4 columns for MariaDB
+	rows = mock.NewRows([]string{"File", "Position", "Binlog_Do_DB", "Binlog_Ignore_DB"}).AddRow(
+		"mysql-bin.000009", 11232, "do_db", "ignore_db",
+	)
+	mock.ExpectQuery(`SHOW MASTER STATUS`).WillReturnRows(rows)
+	rows = mock.NewRows([]string{"Variable_name", "Value"}).AddRow("gtid_binlog_pos", "1-2-100")
+	mock.ExpectQuery(`SHOW GLOBAL VARIABLES LIKE 'gtid_binlog_pos'`).WillReturnRows(rows)
+
+	binlogDoDB, binlogIgnoreDB, err = GetBinlogDB(tctx, baseDB, "mariadb")
+	require.Nil(t, err)
+	require.Equal(t, binlogDoDB, "do_db")
+	require.Equal(t, binlogIgnoreDB, "ignore_db")
+	require.Nil(t, mock.ExpectationsWereMet())
 }
diff --git a/dm/pkg/utils/db_test.go b/dm/pkg/utils/db_test.go
index a5a1aaf06..1bfce9f32 100644
--- a/dm/pkg/utils/db_test.go
+++ b/dm/pkg/utils/db_test.go
@@ -75,88 +75,6 @@ func (t *testDBSuite) TestGetRandomServerID(c *C) {
 	c.Assert(serverID, Not(Equals), 101)
 }
 
-func TestGetPosAndGs(t *testing.T) {
-	ctx, cancel := context.WithTimeout(context.Background(), DefaultDBTimeout)
-	defer cancel()
-
-	db, mock, err := sqlmock.New()
-	require.Nil(t, err)
-
-	// 5 columns for MySQL
-	rows := mock.NewRows([]string{"File", "Position", "Binlog_Do_DB", "Binlog_Ignore_DB", "Executed_Gtid_Set"}).AddRow(
-		"mysql-bin.000009", 11232, "do_db", "ignore_db", "074be7f4-f0f1-11ea-95bd-0242ac120002:1-699",
-	)
-	mock.ExpectQuery(`SHOW MASTER STATUS`).WillReturnRows(rows)
-
-	pos, gs, err := GetPosAndGs(ctx, db, "mysql")
-	require.Nil(t, err)
-	require.Equal(t, pos, gmysql.Position{
-		Name: "mysql-bin.000009",
-		Pos:  11232,
-	})
-	require.Equal(t, gs.String(), "074be7f4-f0f1-11ea-95bd-0242ac120002:1-699")
-	require.Nil(t, mock.ExpectationsWereMet())
-
-	// 4 columns for MariaDB
-	rows = mock.NewRows([]string{"File", "Position", "Binlog_Do_DB", "Binlog_Ignore_DB"}).AddRow(
-		"mysql-bin.000009", 11232, "do_db", "ignore_db",
-	)
-	mock.ExpectQuery(`SHOW MASTER STATUS`).WillReturnRows(rows)
-	rows = mock.NewRows([]string{"Variable_name", "Value"}).AddRow("gtid_binlog_pos", "1-2-100")
-	mock.ExpectQuery(`SHOW GLOBAL VARIABLES LIKE 'gtid_binlog_pos'`).WillReturnRows(rows)
-
-	pos, gs, err = GetPosAndGs(ctx, db, "mariadb")
-	require.Nil(t, err)
-	require.Equal(t, pos, gmysql.Position{
-		Name: "mysql-bin.000009",
-		Pos:  11232,
-	})
-	require.Equal(t, gs.String(), "1-2-100")
-	require.Nil(t, mock.ExpectationsWereMet())
-
-	// some upstream (maybe a polarDB secondary node)
-	rows = mock.NewRows([]string{"File", "Position", "Binlog_Do_DB", "Binlog_Ignore_DB"})
-	mock.ExpectQuery(`SHOW MASTER STATUS`).WillReturnRows(rows)
-
-	_, gs, err = GetPosAndGs(ctx, db, "mysql")
-	require.Nil(t, gs)
-	require.NotNil(t, err)
-}
-
-func TestGetBinlogDB(t *testing.T) {
-	ctx, cancel := context.WithTimeout(context.Background(), DefaultDBTimeout)
-	defer cancel()
-
-	db, mock, err := sqlmock.New()
-	require.Nil(t, err)
-
-	// 5 columns for MySQL
-	rows := mock.NewRows([]string{"File", "Position", "Binlog_Do_DB", "Binlog_Ignore_DB", "Executed_Gtid_Set"}).AddRow(
-		"mysql-bin.000009", 11232, "do_db", "ignore_db", "074be7f4-f0f1-11ea-95bd-0242ac120002:1-699",
-	)
-	mock.ExpectQuery(`SHOW MASTER STATUS`).WillReturnRows(rows)
-
-	binlogDoDB, binlogIgnoreDB, err := GetBinlogDB(ctx, db, "mysql")
-	require.Nil(t, err)
-	require.Equal(t, binlogDoDB, "do_db")
-	require.Equal(t, binlogIgnoreDB, "ignore_db")
-	require.Nil(t, mock.ExpectationsWereMet())
-
-	// 4 columns for MariaDB
-	rows = mock.NewRows([]string{"File", "Position", "Binlog_Do_DB", "Binlog_Ignore_DB"}).AddRow(
-		"mysql-bin.000009", 11232, "do_db", "ignore_db",
-	)
-	mock.ExpectQuery(`SHOW MASTER STATUS`).WillReturnRows(rows)
-	rows = mock.NewRows([]string{"Variable_name", "Value"}).AddRow("gtid_binlog_pos", "1-2-100")
-	mock.ExpectQuery(`SHOW GLOBAL VARIABLES LIKE 'gtid_binlog_pos'`).WillReturnRows(rows)
-
-	binlogDoDB, binlogIgnoreDB, err = GetBinlogDB(ctx, db, "mariadb")
-	require.Nil(t, err)
-	require.Equal(t, binlogDoDB, "do_db")
-	require.Equal(t, binlogIgnoreDB, "ignore_db")
-	require.Nil(t, mock.ExpectationsWereMet())
-}
-
 func (t *testDBSuite) TestGetMariaDBGtidDomainID(c *C) {
 	ctx, cancel := context.WithTimeout(context.Background(), DefaultDBTimeout)
 	defer cancel()
diff --git a/dm/pkg/utils/string_test.go b/dm/pkg/utils/string_test.go
index 12c946cde..df60aba4d 100644
--- a/dm/pkg/utils/string_test.go
+++ b/dm/pkg/utils/string_test.go
@@ -14,14 +14,14 @@
 package utils
 
 import (
-	. "github.com/pingcap/check"
-)
+	"testing"
 
-var _ = Suite(&testStringSuite{})
+	"github.com/stretchr/testify/require"
+)
 
-type testStringSuite struct{}
+func TestTruncateString(t *testing.T) {
+	t.Parallel()
 
-func (t *testStringSuite) TestTruncateString(c *C) {
 	cases := []struct {
 		s      string
 		n      int
@@ -38,11 +38,13 @@ func (t *testStringSuite) TestTruncateString(c *C) {
 	}
 
 	for _, cs := range cases {
-		c.Assert(TruncateString(cs.s, cs.n), Equals, cs.expect)
+		require.Equal(t, cs.expect, TruncateString(cs.s, cs.n))
 	}
 }
 
-func (t *testStringSuite) TestTruncateInterface(c *C) {
+func TestTruncateInterface(t *testing.T) {
+	t.Parallel()
+
 	type compose struct {
 		b bool
 		i int
@@ -82,6 +84,6 @@ func (t *testStringSuite) TestTruncateInterface(c *C) {
 	}
 
 	for _, cs := range cases {
-		c.Assert(TruncateInterface(cs.v, cs.n), Equals, cs.expect)
+		require.Equal(t, cs.expect, TruncateInterface(cs.v, cs.n))
 	}
 }
diff --git a/dm/syncer/dbconn/upstream_db_test.go b/dm/syncer/dbconn/upstream_db_test.go
index 6389ddd0e..b66a0a2dd 100644
--- a/dm/syncer/dbconn/upstream_db_test.go
+++ b/dm/syncer/dbconn/upstream_db_test.go
@@ -24,6 +24,8 @@ import (
 	. "github.com/pingcap/check"
 
 	"github.com/pingcap/tiflow/dm/dm/config"
+	"github.com/pingcap/tiflow/dm/pkg/conn"
+	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
 )
 
@@ -79,7 +81,7 @@ func (s *testDBSuite) resetBinlogSyncer(c *C) {
 		s.syncer.Close()
 	}
 
-	pos, _, err := utils.GetPosAndGs(context.Background(), s.db, "mysql")
+	pos, _, err := conn.GetPosAndGs(tcontext.Background(), conn.NewBaseDB(s.db), "mysql")
 	c.Assert(err, IsNil)
 
 	s.syncer = replication.NewBinlogSyncer(cfg)
diff --git a/dm/tests/full_mode/run.sh b/dm/tests/full_mode/run.sh
index 0f6b530ca..88e02dd6e 100755
--- a/dm/tests/full_mode/run.sh
+++ b/dm/tests/full_mode/run.sh
@@ -150,7 +150,7 @@ function only_route_schema() {
 		"query-status test" \
 		"\"stage\": \"Finished\"" 1
 
-	run_sql_tidb_with_retry "SHOW DATABASES LIKE 'full_mode_test';" "Database: full_mode_test"
+	run_sql_tidb_with_retry "SHOW DATABASES LIKE 'full_mode_test';" ": full_mode_test"
 	cleanup_process $*
 	cleanup_data full_mode
 }
diff --git a/engine/pkg/rpcutil/client_test.go b/engine/pkg/rpcutil/client_test.go
index 7bebe9a4c..47ad1e7a2 100644
--- a/engine/pkg/rpcutil/client_test.go
+++ b/engine/pkg/rpcutil/client_test.go
@@ -124,3 +124,24 @@ func TestFailToDialLeaderAfterwards(t *testing.T) {
 	require.ErrorIs(t, err, derror.ErrNoRPCClient)
 	t.Log(err.Error())
 }
+
+func getMockTimeoutDial(succEventually bool) DialFunc[*mockRPCClient] {
+	retryCount := 0
+	return func(_ context.Context, addr string) (*mockRPCClient, CloseableConnIface, error) {
+		retryCount++
+		if succEventually && retryCount == defaultDialRetry {
+			return &mockRPCClient{addr: addr}, &closer{}, nil
+		}
+		return nil, nil, errors.New("context deadline exceeded")
+	}
+}
+
+func TestTimeout(t *testing.T) {
+	ctx := context.Background()
+	clients, err := NewFailoverRPCClients(ctx, []string{"timeout"}, getMockTimeoutDial(true))
+	require.NoError(t, err)
+	require.Equal(t, "timeout", clients.GetLeaderClient().addr)
+
+	_, err = NewFailoverRPCClients(ctx, []string{"timeout"}, getMockTimeoutDial(false))
+	require.Regexp(t, ".*failed to dial to master.*", err)
+}
diff --git a/pkg/upstream/manager_test.go b/pkg/upstream/manager_test.go
index f5d19417b..b79b116a3 100644
--- a/pkg/upstream/manager_test.go
+++ b/pkg/upstream/manager_test.go
@@ -15,6 +15,7 @@ package upstream
 
 import (
 	"context"
+	"sync"
 	"testing"
 	"time"
 
@@ -139,3 +140,39 @@ func TestAddUpstream(t *testing.T) {
 	}
 	require.NotNil(t, up.Error())
 }
+
+func TestCloseManager(t *testing.T) {
+	ctx := context.TODO()
+	m := &Manager{
+		ups: &sync.Map{},
+	}
+	ctx, cancel := context.WithCancel(ctx)
+	m.ctx = ctx
+	m.cancel = cancel
+	up := &Upstream{
+		wg: new(sync.WaitGroup),
+	}
+	_, cancel = context.WithCancel(ctx)
+	up.cancel = cancel
+	m.ups.Store(uint64(1), up)
+	m.Close()
+	require.Equal(t, closed, up.status)
+	_, ok := m.ups.Load(uint64(1))
+	require.False(t, ok)
+}
+
+func TestRemoveThenAddAgain(t *testing.T) {
+	m := NewManager(context.Background(), "id")
+	m.initUpstreamFunc = func(ctx context.Context,
+		up *Upstream, gcID string,
+	) error {
+		return nil
+	}
+	up := m.AddUpstream(uint64(3), &model.UpstreamInfo{})
+	require.NotNil(t, up)
+	// test Tick
+	_ = m.Tick(context.Background(), &orchestrator.GlobalReactorState{})
+	require.False(t, up.idleTime.IsZero())
+	_ = m.AddUpstream(uint64(3), &model.UpstreamInfo{})
+	require.True(t, up.idleTime.IsZero())
+}