shutdown job on src changed in incr stage #680-2

Author: ffffwh

drivers/mysql/mysql/applier_incr.go

@@ -91,8 +91,9 @@ func NewApplierIncr(subject string, mysqlContext *common.MySQLDriverConfig,
 }
 
 func (a *ApplierIncr) Run() (err error) {
-	a.logger.Debug("beging connetion mysql 4 validate  serverid")
-	if err := a.validateServerUUID(); err != nil {
+	a.logger.Debug("Run. GetServerUUID. before")
+	a.MySQLServerUuid, err = sql.GetServerUUID(a.db)
+	if err != nil {
 		return err
 	}
 
@@ -600,11 +601,3 @@ func (a *ApplierIncr) setTableItemForBinlogEntry(binlogEntry *common.BinlogEntry
 	}
 	return nil
 }
-
-func (a *ApplierIncr) validateServerUUID() error {
-	query := `SELECT @@SERVER_UUID`
-	if err := a.db.QueryRow(query).Scan(&a.MySQLServerUuid); err != nil {
-		return err
-	}
-	return nil
-}

drivers/mysql/mysql/binlog/binlog_reader.go

@@ -8,7 +8,9 @@ package binlog
 
 import (
 	"bytes"
+	gosql "database/sql"
 	"github.com/actiontech/dtle/drivers/mysql/common"
+	"github.com/actiontech/dtle/drivers/mysql/mysql/sql"
 	"github.com/pingcap/parser/format"
 	"github.com/pkg/errors"
 	"github.com/shirou/gopsutil/mem"
@@ -91,6 +93,8 @@ type BinlogReader struct {
 	maybeSqleContext *sqle.Context
 	memory           *int64
 	extractedTxCount uint32
+	db               *gosql.DB
+	serverUUID       string
 }
 
 type SqlFilter struct {
@@ -154,9 +158,7 @@ func parseSqlFilter(strs []string) (*SqlFilter, error) {
 	return s, nil
 }
 
-func NewMySQLReader(execCtx *common.ExecContext, cfg *common.MySQLDriverConfig, logger hclog.Logger,
-	replicateDoDb []*common.DataSource, sqleContext *sqle.Context, memory *int64,
-) (binlogReader *BinlogReader, err error) {
+func NewMySQLReader(execCtx *common.ExecContext, cfg *common.MySQLDriverConfig, logger hclog.Logger, replicateDoDb []*common.DataSource, sqleContext *sqle.Context, memory *int64, db *gosql.DB) (binlogReader *BinlogReader, err error) {
 
 	sqlFilter, err := parseSqlFilter(cfg.SqlFilter)
 	if err != nil {
@@ -176,6 +178,12 @@ func NewMySQLReader(execCtx *common.ExecContext, cfg *common.MySQLDriverConfig,
 		sqlFilter:            sqlFilter,
 		maybeSqleContext:     sqleContext,
 		memory:               memory,
+		db:                   db,
+	}
+
+	binlogReader.serverUUID, err = sql.GetServerUUID(db)
+	if err != nil {
+		return nil, err
 	}
 
 	for _, db := range replicateDoDb {
@@ -1041,6 +1049,15 @@ func (b *BinlogReader) DataStreamEvents(entriesChannel chan<- *common.BinlogEntr
 		b.currentCoordMutex.Unlock()
 
 		if ev.Header.EventType == replication.ROTATE_EVENT {
+			serverUUID, err := sql.GetServerUUID(b.db)
+			if err != nil {
+				return errors.Wrap(err, "on rotate_event. GetServerUUID")
+			}
+			if serverUUID != b.serverUUID {
+				return fmt.Errorf("serverUUID changed from %v to %v. job should restart",
+					b.serverUUID, serverUUID)
+			}
+
 			rotateEvent := ev.Event.(*replication.RotateEvent)
 			nextLogName := string(rotateEvent.NextLogName)
 			b.currentCoordMutex.Lock()

drivers/mysql/mysql/extractor.go

@@ -669,7 +669,7 @@ func (e *Extractor) getSchemaTablesAndMeta() error {
 // Cooperate with `initiateStreaming()` using `e.streamerReadyCh`. Any err will be sent thru the chan.
 func (e *Extractor) initBinlogReader(binlogCoordinates *base.BinlogCoordinatesX) {
 	binlogReader, err := binlog.NewMySQLReader(e.execCtx, e.mysqlContext, e.logger.ResetNamed("reader"),
-		e.replicateDoDb, e.context, e.memory2)
+		e.replicateDoDb, e.context, e.memory2, e.db)
 	if err != nil {
 		e.logger.Error("err at initBinlogReader: NewMySQLReader", "err", err)
 		e.streamerReadyCh <- err

drivers/mysql/mysql/sql/sqlutils.go

@@ -242,6 +242,14 @@ type QueryAble interface {
 	QueryRow(query string, args ...interface{}) *gosql.Row
 }
 
+func GetServerUUID(db QueryAble) (result string, err error) {
+	err = db.QueryRow(`SELECT @@SERVER_UUID /*dtle*/`).Scan(&result)
+	if err != nil {
+		return "", err
+	}
+	return result, nil
+}
+
 // queryResultData returns a raw array of rows for a given query, optionally reading and returning column names
 func queryResultData(db *gosql.DB, query string, retrieveColumns bool, args ...interface{}) (ResultData, []string, error) {
 	var err error