Set MySQL sink default isolation level to read-committed (#4138)

close #3589

cdc/sink/mysql.go

@@ -604,6 +604,7 @@ func (s *mysqlSink) prepareDMLs(rows []*model.RowChangedEvent, replicaID uint64,
 	values := make([][]interface{}, 0, len(rows))
 	replaces := make(map[string][][]interface{})
 	rowCount := 0
+	// translateToInsert control the update and insert behavior
 	translateToInsert := s.params.enableOldValue && !s.params.safeMode
 
 	// flush cached batch replace or insert, to keep the sequence of DMLs

cdc/sink/mysql_params.go

@@ -46,6 +46,7 @@ const (
 	defaultWriteTimeout        = "2m"
 	defaultDialTimeout         = "2m"
 	defaultSafeMode            = true
+	defaultTxnIsolationRC      = "READ-COMMITTED"
 )
 
 var defaultParams = &sinkParams{
@@ -227,6 +228,8 @@ func generateDSNByParams(
 	dsnCfg.Params["readTimeout"] = params.readTimeout
 	dsnCfg.Params["writeTimeout"] = params.writeTimeout
 	dsnCfg.Params["timeout"] = params.dialTimeout
+	// Since we don't need select, just set default isolation level to read-committed
+	dsnCfg.Params["transaction_isolation"] = fmt.Sprintf(`"%s"`, defaultTxnIsolationRC)
 
 	autoRandom, err := checkTiDBVariable(ctx, testDB, "allow_auto_random_explicit_insert", "1")
 	if err != nil {

cdc/sink/mysql_params_test.go

@@ -27,7 +27,6 @@ import (
 )
 
 func TestSinkParamsClone(t *testing.T) {
-	defer testleak.AfterTestT(t)()
 	param1 := defaultParams.Clone()
 	param2 := param1.Clone()
 	param2.changefeedID = "123"
@@ -59,8 +58,6 @@ func TestSinkParamsClone(t *testing.T) {
 }
 
 func TestGenerateDSNByParams(t *testing.T) {
-	defer testleak.AfterTestT(t)()
-
 	testDefaultParams := func() {
 		db, err := mockTestDB(false)
 		require.Nil(t, err)
@@ -76,6 +73,7 @@ func TestGenerateDSNByParams(t *testing.T) {
 			"readTimeout=2m",
 			"writeTimeout=2m",
 			"allow_auto_random_explicit_insert=1",
+			"transaction_isolation=%22READ-COMMITTED%22",
 		}
 		for _, param := range expectedParams {
 			require.True(t, strings.Contains(dsnStr, param))