sink/mysql: add db read/write timeout

cdc/sink/mysql.go

@@ -61,6 +61,8 @@ const (
 	defaultFlushInterval       = time.Millisecond * 50
 	defaultBatchReplaceEnabled = true
 	defaultBatchReplaceSize    = 20
+	defaultReadTimeout         = "2m"
+	defaultWriteTimeout        = "2m"
 )
 
 var (
@@ -270,6 +272,8 @@ type sinkParams struct {
 	captureAddr         string
 	batchReplaceEnabled bool
 	batchReplaceSize    int
+	readTimeout         string
+	writeTimeout        string
 }
 
 func (s *sinkParams) Clone() *sinkParams {
@@ -283,47 +287,58 @@ var defaultParams = &sinkParams{
 	tidbTxnMode:         defaultTiDBTxnMode,
 	batchReplaceEnabled: defaultBatchReplaceEnabled,
 	batchReplaceSize:    defaultBatchReplaceSize,
+	readTimeout:         defaultReadTimeout,
+	writeTimeout:        defaultWriteTimeout,
 }
 
-func configureSinkURI(ctx context.Context, dsnCfg *dmysql.Config, tz *time.Location, params *sinkParams) (string, error) {
+func checkTiDBVariable(ctx context.Context, db *sql.DB, variableName, defaultValue string) (string, error) {
+	var name string
+	var value string
+	querySQL := fmt.Sprintf("show session variables like '%s';", variableName)
+	err := db.QueryRowContext(ctx, querySQL).Scan(&name, &value)
+	if err != nil && err != sql.ErrNoRows {
+		errMsg := "fail to query session variable " + variableName
+		return "", errors.Annotate(cerror.WrapError(cerror.ErrMySQLQueryError, err), errMsg)
+	}
+	// session variable works, use given default value
+	if err == nil {
+		return defaultValue, nil
+	}
+	// session variable not exists, return "" to ignore it
+	return "", nil
+}
+
+func configureSinkURI(
+	ctx context.Context,
+	dsnCfg *dmysql.Config,
+	tz *time.Location,
+	params *sinkParams,
+	testDB *sql.DB,
+) (string, error) {
 	if dsnCfg.Params == nil {
 		dsnCfg.Params = make(map[string]string, 1)
 	}
 	dsnCfg.DBName = ""
 	dsnCfg.InterpolateParams = true
 	dsnCfg.MultiStatements = true
 	dsnCfg.Params["time_zone"] = fmt.Sprintf(`"%s"`, tz.String())
+	dsnCfg.Params["readTimeout"] = params.readTimeout
+	dsnCfg.Params["writeTimeout"] = params.writeTimeout
 
-	testDB, err := sql.Open("mysql", dsnCfg.FormatDSN())
+	autoRandom, err := checkTiDBVariable(ctx, testDB, "allow_auto_random_explicit_insert", "1")
 	if err != nil {
-		return "", errors.Annotate(
-			cerror.WrapError(cerror.ErrMySQLConnectionError, err), "fail to open MySQL connection when configuring sink")
-	}
-	defer testDB.Close()
-	log.Debug("Opened connection to configure some tidb special parameters")
-
-	var variableName string
-	var autoRandomInsertEnabled string
-	queryStr := "show session variables like 'allow_auto_random_explicit_insert';"
-	err = testDB.QueryRowContext(ctx, queryStr).Scan(&variableName, &autoRandomInsertEnabled)
-	if err != nil && err != sql.ErrNoRows {
-		return "", errors.Annotate(
-			cerror.WrapError(cerror.ErrMySQLQueryError, err), "fail to query sink for support of auto-random")
+		return "", err
 	}
-	if err == nil && (autoRandomInsertEnabled == "off" || autoRandomInsertEnabled == "0") {
-		dsnCfg.Params["allow_auto_random_explicit_insert"] = "1"
-		log.Debug("Set allow_auto_random_explicit_insert to 1")
+	if autoRandom != "" {
+		dsnCfg.Params["allow_auto_random_explicit_insert"] = autoRandom
 	}
 
-	var txnMode string
-	queryStr = "show session variables like 'tidb_txn_mode';"
-	err = testDB.QueryRowContext(ctx, queryStr).Scan(&variableName, &txnMode)
-	if err != nil && err != sql.ErrNoRows {
-		return "", errors.Annotate(
-			cerror.WrapError(cerror.ErrMySQLQueryError, err), "fail to query sink for txn mode")
+	txnMode, err := checkTiDBVariable(ctx, testDB, "tidb_txn_mode", params.tidbTxnMode)
+	if err != nil {
+		return "", err
 	}
-	if err == nil {
-		dsnCfg.Params["tidb_txn_mode"] = params.tidbTxnMode
+	if txnMode != "" {
+		dsnCfg.Params["tidb_txn_mode"] = txnMode
 	}
 
 	dsnClone := dsnCfg.Clone()
@@ -432,7 +447,20 @@ func newMySQLSink(ctx context.Context, changefeedID model.ChangeFeedID, sinkURI
 	if err != nil {
 		return nil, cerror.WrapError(cerror.ErrMySQLInvalidConfig, err)
 	}
-	dsnStr, err = configureSinkURI(ctx, dsn, tz, params)
+
+	// create test db used for parameter detection
+	if dsn.Params == nil {
+		dsn.Params = make(map[string]string, 1)
+	}
+	dsn.Params["time_zone"] = fmt.Sprintf(`"%s"`, tz.String())
+	testDB, err := sql.Open("mysql", dsn.FormatDSN())
+	if err != nil {
+		return nil, errors.Annotate(
+			cerror.WrapError(cerror.ErrMySQLConnectionError, err), "fail to open MySQL connection when configuring sink")
+	}
+	defer testDB.Close()
+
+	dsnStr, err = configureSinkURI(ctx, dsn, tz, params, testDB)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}

cdc/sink/mysql_test.go

@@ -15,12 +15,16 @@ package sink
 
 import (
 	"context"
+	"database/sql"
 	"fmt"
 	"sort"
+	"strings"
 	"testing"
 	"time"
 
+	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/davecgh/go-spew/spew"
+	dmysql "github.com/go-sql-driver/mysql"
 	"github.com/pingcap/check"
 	"github.com/pingcap/parser/mysql"
 	"github.com/pingcap/ticdc/cdc/model"
@@ -608,6 +612,8 @@ func (s MySQLSinkSuite) TestSinkParamsClone(c *check.C) {
 		tidbTxnMode:         defaultTiDBTxnMode,
 		batchReplaceEnabled: defaultBatchReplaceEnabled,
 		batchReplaceSize:    defaultBatchReplaceSize,
+		readTimeout:         defaultReadTimeout,
+		writeTimeout:        defaultWriteTimeout,
 	})
 	c.Assert(param2, check.DeepEquals, &sinkParams{
 		changefeedID:        "123",
@@ -616,9 +622,59 @@ func (s MySQLSinkSuite) TestSinkParamsClone(c *check.C) {
 		tidbTxnMode:         defaultTiDBTxnMode,
 		batchReplaceEnabled: false,
 		batchReplaceSize:    defaultBatchReplaceSize,
+		readTimeout:         defaultReadTimeout,
+		writeTimeout:        defaultWriteTimeout,
 	})
 }
 
+func (s MySQLSinkSuite) TestConfigureSinkURI(c *check.C) {
+	db, mock, err := sqlmock.New()
+	c.Assert(err, check.IsNil)
+	columns := []string{"Variable_name", "Value"}
+	mock.ExpectQuery("show session variables like 'allow_auto_random_explicit_insert';").WillReturnRows(
+		sqlmock.NewRows(columns).AddRow("allow_auto_random_explicit_insert", "0"),
+	)
+	mock.ExpectQuery("show session variables like 'tidb_txn_mode';").WillReturnRows(
+		sqlmock.NewRows(columns).AddRow("tidb_txn_mode", "pessimistic"),
+	)
+
+	dsn, err := dmysql.ParseDSN("root:123456@tcp(127.0.0.1:4000)/")
+	c.Assert(err, check.IsNil)
+	dsnStr, err := configureSinkURI(context.TODO(), dsn, time.Local, defaultParams.Clone(), db)
+	c.Assert(err, check.IsNil)
+	expectedParams := []string{
+		"tidb_txn_mode=optimistic",
+		"readTimeout=2m",
+		"writeTimeout=2m",
+		"allow_auto_random_explicit_insert=1",
+	}
+	for _, param := range expectedParams {
+		c.Assert(strings.Contains(dsnStr, param), check.IsTrue)
+	}
+}
+
+func (s MySQLSinkSuite) TestCheckTiDBVariable(c *check.C) {
+	db, mock, err := sqlmock.New()
+	c.Assert(err, check.IsNil)
+	columns := []string{"Variable_name", "Value"}
+
+	mock.ExpectQuery("show session variables like 'allow_auto_random_explicit_insert';").WillReturnRows(
+		sqlmock.NewRows(columns).AddRow("allow_auto_random_explicit_insert", "0"),
+	)
+	val, err := checkTiDBVariable(context.TODO(), db, "allow_auto_random_explicit_insert", "1")
+	c.Assert(err, check.IsNil)
+	c.Assert(val, check.Equals, "1")
+
+	mock.ExpectQuery("show session variables like 'no_exist_variable';").WillReturnError(sql.ErrNoRows)
+	val, err = checkTiDBVariable(context.TODO(), db, "no_exist_variable", "0")
+	c.Assert(err, check.IsNil)
+	c.Assert(val, check.Equals, "")
+
+	mock.ExpectQuery("show session variables like 'version';").WillReturnError(sql.ErrConnDone)
+	_, err = checkTiDBVariable(context.TODO(), db, "version", "5.7.25-TiDB-v4.0.0")
+	c.Assert(err, check.ErrorMatches, ".*"+sql.ErrConnDone.Error())
+}
+
 /*
    import (
    	"context"

go.mod

@@ -4,7 +4,7 @@ go 1.13
 
 require (
 	github.com/BurntSushi/toml v0.3.1
-	github.com/DATA-DOG/go-sqlmock v1.3.3 // indirect
+	github.com/DATA-DOG/go-sqlmock v1.3.3
 	github.com/Shopify/sarama v1.26.1
 	github.com/apache/pulsar-client-go v0.1.1
 	github.com/cenkalti/backoff v2.2.1+incompatible