sink(ticdc): Add Support for Multiple MySQL-Compatible Downstream Addresses in TiCDC for High Availability

cdc/sink/ddlsink/mysql/async_ddl.go

@@ -152,7 +152,7 @@ func (m *DDLSink) doCheck(ctx context.Context, table model.TableName) (done bool
 		return true
 	}
 
-	ret := m.db.QueryRowContext(ctx, fmt.Sprintf(checkRunningAddIndexSQL, table.Schema, table.Table))
+	ret := m.connector.CurrentDB.QueryRowContext(ctx, fmt.Sprintf(checkRunningAddIndexSQL, table.Schema, table.Table))
 	if ret.Err() != nil {
 		log.Error("check async exec ddl failed",
 			zap.String("namespace", m.id.Namespace),

cdc/sink/ddlsink/mysql/mysql_ddl_sink.go

@@ -15,12 +15,12 @@
 
 import (
 	"context"
-	"database/sql"
 	"fmt"
 	"net/url"
 	"time"
 
 	lru "github.com/hashicorp/golang-lru"
+	perrors "github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/log"
 	timodel "github.com/pingcap/tidb/pkg/parser/model"
@@ -54,10 +54,9 @@
 // DDLSink is a sink that writes DDL events to MySQL.
 type DDLSink struct {
 	// id indicates which processor (changefeed) this sink belongs to.
-	id model.ChangeFeedID
-	// db is the database connection.
-	db  *sql.DB
-	cfg *pmysql.Config
+	id        model.ChangeFeedID
+	connector *pmysql.DBConnector
+	cfg       *pmysql.Config
 	// statistics is the statistics of this sink.
 	// We use it to record the DDL count.
 	statistics *metrics.Statistics
@@ -81,22 +80,17 @@
 		return nil, err
 	}
 
-	dsnStr, err := pmysql.GenerateDSN(ctx, sinkURI, cfg, GetDBConnImpl.CreateTemporaryConnection)
+	connector, err := pmysql.NewDBConnectorWithFactory(ctx, cfg, sinkURI, GetDBConnImpl)
 	if err != nil {
 		return nil, err
 	}
 
-	db, err := GetDBConnImpl.CreateStandardConnection(ctx, dsnStr)
+	cfg.IsTiDB, err = pmysql.CheckIsTiDB(ctx, connector.CurrentDB)
 	if err != nil {
 		return nil, err
 	}
 
-	cfg.IsTiDB, err = pmysql.CheckIsTiDB(ctx, db)
-	if err != nil {
-		return nil, err
-	}
-
-	cfg.IsWriteSourceExisted, err = pmysql.CheckIfBDRModeIsSupported(ctx, db)
+	cfg.IsWriteSourceExisted, err = pmysql.CheckIfBDRModeIsSupported(ctx, connector.CurrentDB)
 	if err != nil {
 		return nil, err
 	}
@@ -107,7 +101,7 @@
 	}
 	m := &DDLSink{
 		id:                         changefeedID,
-		db:                         db,
+		connector:                  connector,
 		cfg:                        cfg,
 		statistics:                 metrics.NewStatistics(ctx, changefeedID, sink.TxnSink),
 		lastExecutedNormalDDLCache: lruCache,
@@ -157,7 +151,9 @@
 			return err
 		}
 		return nil
-	}, retry.WithBackoffBaseDelay(pmysql.BackoffBaseDelay.Milliseconds()),
+	}, retry.WithPreExecutionWhenRetry(func() error {
+		return m.connector.SwitchToAnAvailableDB(ctx)
+	}), retry.WithBackoffBaseDelay(pmysql.BackoffBaseDelay.Milliseconds()),
 		retry.WithBackoffMaxDelay(pmysql.BackoffMaxDelay.Milliseconds()),
 		retry.WithMaxTries(defaultDDLMaxRetry),
 		retry.WithIsRetryableErr(errorutil.IsRetryableDDLError))
@@ -210,7 +206,7 @@
 	start := time.Now()
 	log.Info("Start exec DDL", zap.String("namespace", m.id.Namespace), zap.String("changefeed", m.id.ID),
 		zap.Uint64("commitTs", ddl.CommitTs), zap.String("DDL", ddl.Query))
-	tx, err := m.db.BeginTx(ctx, nil)
+	tx, err := m.connector.CurrentDB.BeginTx(ctx, nil)
 	if err != nil {
 		return err
 	}
@@ -229,7 +225,7 @@
 	// we try to set cdc write source for the ddl
 	if err = pmysql.SetWriteSource(pctx, m.cfg, tx); err != nil {
 		if rbErr := tx.Rollback(); rbErr != nil {
-			if errors.Cause(rbErr) != context.Canceled {
+			if perrors.Cause(rbErr) != context.Canceled {
 				log.Error("Failed to rollback",
 					zap.String("namespace", m.id.Namespace),
 					zap.String("changefeed", m.id.ID), zap.Error(err))
@@ -282,8 +278,8 @@
 	if m.statistics != nil {
 		m.statistics.Close()
 	}
-	if m.db != nil {
-		if err := m.db.Close(); err != nil {
+	if m.connector.CurrentDB != nil {
+		if err := m.connector.CurrentDB.Close(); err != nil {
 			log.Warn("MySQL ddl sink close db wit error",
 				zap.String("namespace", m.id.Namespace),
 				zap.String("changefeed", m.id.ID),

cdc/sink/dmlsink/txn/mysql/mysql.go

@@ -61,7 +61,7 @@
 type mysqlBackend struct {
 	workerID    int
 	changefeed  string
-	db          *sql.DB
+	connector   pmysql.DBConnector
 	cfg         *pmysql.Config
 	dmlMaxRetry uint64
 
@@ -97,47 +97,44 @@
 		return nil, err
 	}
 
-	dsnStr, err := pmysql.GenerateDSN(ctx, sinkURI, cfg, dbConnFactory.CreateTemporaryConnection)
+	connector, err := pmysql.NewDBConnectorWithFactory(ctx, cfg, sinkURI, dbConnFactory)
 	if err != nil {
 		return nil, err
 	}
 
-	db, err := dbConnFactory.CreateStandardConnection(ctx, dsnStr)
+	cfg.IsTiDB, err = pmysql.CheckIsTiDB(ctx, connector.CurrentDB)
 	if err != nil {
 		return nil, err
 	}
 
-	cfg.IsTiDB, err = pmysql.CheckIsTiDB(ctx, db)
+	cfg.IsWriteSourceExisted, err = pmysql.CheckIfBDRModeIsSupported(ctx, connector.CurrentDB)
 	if err != nil {
 		return nil, err
 	}
 
-	cfg.IsWriteSourceExisted, err = pmysql.CheckIfBDRModeIsSupported(ctx, db)
-	if err != nil {
-		return nil, err
-	}
-
-	// By default, cache-prep-stmts=true, an LRU cache is used for prepared statements,
-	// two connections are required to process a transaction.
-	// The first connection is held in the tx variable, which is used to manage the transaction.
-	// The second connection is requested through a call to s.db.Prepare
-	// in case of a cache miss for the statement query.
-	// The connection pool for CDC is configured with a static size, equal to the number of workers.
-	// CDC may hang at the "Get Connection" call is due to the limited size of the connection pool.
-	// When the connection pool is small,
-	// the chance of all connections being active at the same time increases,
-	// leading to exhaustion of available connections and a hang at the "Get Connection" call.
-	// This issue is less likely to occur when the connection pool is larger,
-	// as there are more connections available for use.
-	// Adding an extra connection to the connection pool solves the connection exhaustion issue.
-	db.SetMaxIdleConns(cfg.WorkerCount + 1)
-	db.SetMaxOpenConns(cfg.WorkerCount + 1)
+	connector.ConfigureDBWhenSwitch(func() {
+		// By default, cache-prep-stmts=true, an LRU cache is used for prepared statements,
+		// two connections are required to process a transaction.
+		// The first connection is held in the tx variable, which is used to manage the transaction.
+		// The second connection is requested through a call to s.db.Prepare
+		// in case of a cache miss for the statement query.
+		// The connection pool for CDC is configured with a static size, equal to the number of workers.
+		// CDC may hang at the "Get Connection" call is due to the limited size of the connection pool.
+		// When the connection pool is small,
+		// the chance of all connections being active at the same time increases,
+		// leading to exhaustion of available connections and a hang at the "Get Connection" call.
+		// This issue is less likely to occur when the connection pool is larger,
+		// as there are more connections available for use.
+		// Adding an extra connection to the connection pool solves the connection exhaustion issue.
+		connector.CurrentDB.SetMaxIdleConns(cfg.WorkerCount + 1)
+		connector.CurrentDB.SetMaxOpenConns(cfg.WorkerCount + 1)
+	}, true)
 
 	// Inherit the default value of the prepared statement cache from the SinkURI Options
 	cachePrepStmts := cfg.CachePrepStmts
 	if cachePrepStmts {
 		// query the size of the prepared statement cache on serverside
-		maxPreparedStmtCount, err := pmysql.QueryMaxPreparedStmtCount(ctx, db)
+		maxPreparedStmtCount, err := pmysql.QueryMaxPreparedStmtCount(ctx, connector.CurrentDB)
 		if err != nil {
 			return nil, err
 		}
@@ -167,7 +164,7 @@
 	}
 
 	var maxAllowedPacket int64
-	maxAllowedPacket, err = pmysql.QueryMaxAllowedPacket(ctx, db)
+	maxAllowedPacket, err = pmysql.QueryMaxAllowedPacket(ctx, connector.CurrentDB)
 	if err != nil {
 		log.Warn("failed to query max_allowed_packet, use default value",
 			zap.String("changefeed", changefeed),
@@ -180,7 +177,7 @@
 		backends = append(backends, &mysqlBackend{
 			workerID:    i,
 			changefeed:  changefeed,
-			db:          db,
+			connector:   *connector,
 			cfg:         cfg,
 			dmlMaxRetry: defaultDMLMaxRetry,
 			statistics:  statistics,
@@ -261,9 +258,9 @@
 	if s.stmtCache != nil {
 		s.stmtCache.Purge()
 	}
-	if s.db != nil {
-		err = s.db.Close()
-		s.db = nil
+	if s.connector.CurrentDB != nil {
+		err = s.connector.CurrentDB.Close()
+		s.connector.CurrentDB = nil
 	}
 	return
 }
@@ -673,7 +670,7 @@
 		if s.cachePrepStmts {
 			if stmt, ok := s.stmtCache.Get(query); ok {
 				prepStmt = stmt.(*sql.Stmt)
-			} else if stmt, err := s.db.Prepare(query); err == nil {
+			} else if stmt, err := s.connector.CurrentDB.Prepare(query); err == nil {
 				prepStmt = stmt
 				s.stmtCache.Add(query, stmt)
 			} else {
@@ -741,7 +738,7 @@
 		})
 
 		err := s.statistics.RecordBatchExecution(func() (int, int64, error) {
-			tx, err := s.db.BeginTx(pctx, nil)
+			tx, err := s.connector.CurrentDB.BeginTx(pctx, nil)
 			if err != nil {
 				return 0, 0, logDMLTxnErr(
 					wrapMysqlTxnError(err),
@@ -799,7 +796,9 @@
 			zap.Int("workerID", s.workerID),
 			zap.Int("numOfRows", dmls.rowCount))
 		return nil
-	}, retry.WithBackoffBaseDelay(pmysql.BackoffBaseDelay.Milliseconds()),
+	}, retry.WithPreExecutionWhenRetry(func() error {
+		return s.connector.SwitchToAnAvailableDB(pctx)
+	}), retry.WithBackoffBaseDelay(pmysql.BackoffBaseDelay.Milliseconds()),
 		retry.WithBackoffMaxDelay(pmysql.BackoffMaxDelay.Milliseconds()),
 		retry.WithMaxTries(s.dmlMaxRetry),
 		retry.WithIsRetryableErr(isRetryableDMLError))

cdc/sink/validator/validator.go

@@ -122,11 +122,11 @@
 	if err != nil {
 		return err
 	}
-	dsn, err := pmysql.GenBasicDSN(sinkURI, cfg)
+	dsnList, err := pmysql.GetDSNCfgs(sinkURI, cfg)
 	if err != nil {
 		return err
 	}
-	testDB, err := pmysql.GetTestDB(ctx, dsn, pmysql.CreateMySQLDBConn)
+	testDB, err := pmysql.GetTestDB(ctx, dsnList[0], pmysql.CreateMySQLDBConn)
 	if err != nil {
 		return err
 	}