mysql (ticdc): only set write source when downstream is supported (#9175

) (#9236)

close #9180

cdc/sinkv2/eventsink/txn/mysql/mysql.go

@@ -103,6 +103,11 @@ func NewMySQLBackends(
 		return nil, err
 	}
 
+	cfg.IsWriteSourceExisted, err = pmysql.CheckIfBDRModeIsSupported(ctx, db)
+	if err != nil {
+		return nil, err
+	}
+
 	db.SetMaxIdleConns(cfg.WorkerCount)
 	db.SetMaxOpenConns(cfg.WorkerCount)
 
@@ -712,7 +717,7 @@ func (s *mysqlBackend) execDMLWithMaxRetries(pctx context.Context, dmls *prepare
 				}
 			}
 
-			// we set write source for each txn,
+			// we try to set write source for each txn,
 			// so we can use it to trace the data source
 			if err = s.setWriteSource(pctx, tx); err != nil {
 				err := logDMLTxnErr(
@@ -805,8 +810,8 @@ func (s *mysqlBackend) setDMLMaxRetry(maxRetry uint64) {
 
 // setWriteSource sets write source for the transaction.
 func (s *mysqlBackend) setWriteSource(ctx context.Context, txn *sql.Tx) error {
-	// we only set write source when donwstream is TiDB
-	if !s.cfg.IsTiDB {
+	// we only set write source when donwstream is TiDB and write source is existed.
+	if !s.cfg.IsWriteSourceExisted {
 		return nil
 	}
 	// downstream is TiDB, set system variables.

cdc/sinkv2/eventsink/txn/mysql/mysql_test.go

@@ -83,6 +83,10 @@ func newTestMockDB(t *testing.T) (db *sql.DB, mock sqlmock.Sqlmock) {
 		Number:  1305,
 		Message: "FUNCTION test.tidb_version does not exist",
 	})
+	mock.ExpectQuery("select tidb_version()").WillReturnError(&dmysql.MySQLError{
+		Number:  1305,
+		Message: "FUNCTION test.tidb_version does not exist",
+	})
 	require.Nil(t, err)
 	return
 }

pkg/applier/redo_test.go

@@ -236,7 +236,10 @@ func getMockDB(t *testing.T) *sql.DB {
 		Number:  1305,
 		Message: "FUNCTION test.tidb_version does not exist",
 	})
-
+	mock.ExpectQuery("select tidb_version()").WillReturnError(&mysql.MySQLError{
+		Number:  1305,
+		Message: "FUNCTION test.tidb_version does not exist",
+	})
 	mock.ExpectBegin()
 	mock.ExpectExec("USE `test`;").WillReturnResult(sqlmock.NewResult(1, 1))
 	mock.ExpectExec("create table checkpoint(id int)").WillReturnResult(sqlmock.NewResult(1, 1))

pkg/sink/mysql/config.go

@@ -96,7 +96,11 @@ type Config struct {
 	ForceReplicate         bool
 	EnableOldValue         bool
 
-	IsTiDB          bool // IsTiDB is true if the downstream is TiDB
+	IsTiDB bool // IsTiDB is true if the downstream is TiDB
+	// IsBDRModeSupported is true if the downstream is TiDB and write source is existed.
+	// write source exists when the downstream is TiDB and version is greater than or equal to v6.5.0.
+	IsWriteSourceExisted bool
+
 	SourceID        uint64
 	BatchDMLEnable  bool
 	MultiStmtEnable bool