Merge branch 'master' into 4.23.04.x

Author: ffffwh

api/handler/v2/job.go

@@ -11,21 +11,16 @@ import (
 	"strings"
 	"time"
 
-	mysql "github.com/actiontech/dtle/driver/mysql"
-
+	nomadApi "github.com/hashicorp/nomad/api"
 	"github.com/hashicorp/nomad/nomad/structs"
+	"github.com/labstack/echo/v4"
 
-	"github.com/actiontech/dtle/driver/kafka"
-
+	"github.com/actiontech/dtle/api/handler"
+	"github.com/actiontech/dtle/api/models"
 	"github.com/actiontech/dtle/driver/common"
-
+	"github.com/actiontech/dtle/driver/kafka"
+	mysql "github.com/actiontech/dtle/driver/mysql"
 	"github.com/actiontech/dtle/g"
-
-	"github.com/actiontech/dtle/api/models"
-
-	"github.com/actiontech/dtle/api/handler"
-	nomadApi "github.com/hashicorp/nomad/api"
-	"github.com/labstack/echo/v4"
 )
 
 // @Id MigrationJobListV2
@@ -1955,6 +1950,11 @@ func ReverseJobV2(c echo.Context, filterJobType DtleJobType) error {
 			return c.JSON(http.StatusInternalServerError, models.BuildBaseResp(err))
 		}
 
+		if originalJob.BasicTaskProfile.Configuration.SrcConfig.MysqlSrcTaskConfig.TwoWaySync {
+			return c.JSON(http.StatusConflict, models.BuildBaseResp(fmt.Errorf(
+				"job_id=%v; job can't be reversed with TwoWaySync = true", reqParam.JobId)))
+		}
+
 		reverseJobParam := new(models.CreateOrUpdateMysqlToMysqlJobParamV2)
 		reverseJobParam.JobId = fmt.Sprintf("%s-%s", "reverse", consulJobItem.JobId)
 		reverseJobParam.TaskStepName = mysql.JobIncrCopy
@@ -1982,6 +1982,15 @@ func ReverseJobV2(c echo.Context, filterJobType DtleJobType) error {
 				TwoWaySyncGtid:      originalJob.BasicTaskProfile.Configuration.SrcConfig.MysqlSrcTaskConfig.TwoWaySyncGtid,
 			},
 		}
+		for _, dbItem := range reverseJobParam.SrcTask.ReplicateDoDb {
+			for _, tbItem := range dbItem.Tables {
+				if len(tbItem.ColumnMapFrom) > 0 && len(tbItem.ColumnMapTo) == 0 {
+					return c.JSON(http.StatusConflict, models.BuildBaseResp(fmt.Errorf(
+						"job_id=%v; job can't be reversed with ColumnMapFrom not matching ColumnMapTo", reqParam.JobId)))
+				}
+				tbItem.ColumnMapFrom, tbItem.ColumnMapTo = tbItem.ColumnMapTo, tbItem.ColumnMapFrom
+			}
+		}
 		reverseJobParam.DestTask = &models.DestTaskConfig{
 			TaskName:         common.TaskTypeDest,
 			ConnectionConfig: &originalJob.BasicTaskProfile.ConnectionInfo.SrcDataBase,

driver/mysql/applier.go

@@ -10,6 +10,7 @@ import (
 	"bytes"
 	gosql "database/sql"
 	"fmt"
+	mysqldriver "github.com/go-sql-driver/mysql"
 	"math"
 	"strconv"
 	"strings"
@@ -304,9 +305,23 @@ func (a *Applier) Run() {
 			Subject:  a.subject + "_dtrev",
 			StateDir: a.stateDir,
 		}
-		var cfg2 = *a.mysqlContext
+		// cfg2 will be a deepcopy of a.mysqlContext
+		cfg2, err := a.storeManager.GetConfig(a.subject)
+		if err != nil {
+			a.onError(common.TaskStateDead, errors.Wrap(err, "GetConfig"))
+			return
+		}
 		cfg2.SrcConnectionConfig = a.mysqlContext.DestConnectionConfig
 		cfg2.DestConnectionConfig = a.mysqlContext.SrcConnectionConfig
+		for _, dbItem := range cfg2.ReplicateDoDb {
+			for _, tbItem := range dbItem.Tables {
+				if len(tbItem.ColumnMapFrom) > 0 && len(tbItem.ColumnMapTo) == 0 {
+					a.onError(common.TaskStateDead, errors.Wrap(err, "GetConfig"))
+					return
+				}
+				tbItem.ColumnMapFrom, tbItem.ColumnMapTo = tbItem.ColumnMapTo, tbItem.ColumnMapFrom
+			}
+		}
 
 		if strings.ToLower(a.mysqlContext.TwoWaySyncGtid) == "auto" {
 			cfg2.AutoGtid = true
@@ -318,7 +333,7 @@ func (a *Applier) Run() {
 		cfg2.TwoWaySync = false
 		cfg2.TwoWaySyncGtid = ""
 
-		a.revExtractor, err = NewExtractor(execCtx2, &cfg2, a.logger, a.storeManager, a.waitCh, a.ctx)
+		a.revExtractor, err = NewExtractor(execCtx2, cfg2, a.logger, a.storeManager, a.waitCh, a.ctx)
 		if err != nil {
 			a.onError(common.TaskStateDead, errors.Wrap(err, "reversed Extractor"))
 			return
@@ -422,7 +437,7 @@ func (a *Applier) doFullCopy() {
 				return
 			case copyRows := <-a.dumpEntryQueue:
 				//time.Sleep(20 * time.Second) // #348 stub
-				if err = a.ApplyEventQueries(a.db, copyRows); err != nil {
+				if err = a.ApplyEventQueries(copyRows); err != nil {
 					return
 				}
 				atomic.AddInt64(a.memory1, -int64(copyRows.Size()))
@@ -825,7 +840,7 @@ func (a *Applier) ValidateGrants() error {
 	return fmt.Errorf("user has insufficient privileges for applier. Needed:ALTER, CREATE, DROP, INDEX, REFERENCES, INSERT, DELETE, UPDATE, SELECT, TRIGGER ON *.*")
 }
 
-func (a *Applier) ApplyEventQueries(db *gosql.DB, entry *common.DumpEntry) (err error) {
+func (a *Applier) ApplyEventQueries(entry *common.DumpEntry) (err error) {
 	a.logger.Debug("ApplyEventQueries", "schema", entry.TableSchema, "table", entry.TableName,
 		"rows", len(entry.ValuesX))
 
@@ -879,29 +894,18 @@ func (a *Applier) ApplyEventQueries(db *gosql.DB, entry *common.DumpEntry) (err
 
 	queries = append(queries, entry.SqlMode, entry.DbSQL)
 	queries = append(queries, entry.TbSQL...)
-	tx, err := db.BeginTx(a.ctx, &gosql.TxOptions{})
-	if err != nil {
-		return err
-	}
+
+	conn := a.dbs[0].Db
 	nRows := int64(len(entry.ValuesX))
-	defer func() {
-		if err != nil {
-			return
-		}
-		err = tx.Commit()
-		if err == nil {
-			atomic.AddInt64(&a.TotalRowsReplayed, nRows)
-		}
-	}()
-	if _, err := tx.ExecContext(a.ctx, querySetFKChecksOff); err != nil {
+	if _, err := conn.ExecContext(a.ctx, querySetFKChecksOff); err != nil {
 		return err
 	}
 	execQuery := func(query string) error {
 		a.logger.Debug("ApplyEventQueries. exec", "query", g.StrLim(query, 256))
-		_, err := tx.ExecContext(a.ctx, query)
+		_, err := conn.ExecContext(a.ctx, query)
 		if err != nil {
 			queryStart := g.StrLim(query, 10) // avoid printing sensitive information
-			errCtx := errors.Wrapf(err, "tx.Exec. queryStart %v seq", queryStart)
+			errCtx := errors.Wrapf(err, "tx.Exec. queryStart %v", queryStart)
 			if !sql.IgnoreError(err) {
 				a.logger.Error("ApplyEventQueries. exec error", "err", errCtx)
 				return errCtx
@@ -958,13 +962,22 @@ func (a *Applier) ApplyEventQueries(db *gosql.DB, entry *common.DumpEntry) (err
 		// last rows or sql too large
 
 		if needInsert {
-			err := execQuery(buf.String())
-			buf.Reset()
-			if err != nil {
-				return err
+			for iTry := 0; ; iTry++ {
+				err := execQuery(buf.String())
+				if errIsMysqlDeadlock(err) && iTry < a.mysqlContext.RetryTxLimit {
+					a.logger.Info("found deadlock. will retry tx", "schema", entry.TableSchema, "table", entry.Table,
+						"iTry", iTry)
+					time.Sleep(retryTxDelay)
+					continue
+				} else if err != nil {
+					return err
+				}
+				break
 			}
+			buf.Reset()
 		}
 	}
+	atomic.AddInt64(&a.TotalRowsReplayed, nRows)
 
 	return nil
 }
@@ -1214,3 +1227,15 @@ func (a *Applier) updateDumpProgressLoop() {
 		time.Sleep(time.Duration(interval) * time.Second)
 	}
 }
+func errIsMysqlDeadlock(err error) bool {
+	if err != nil {
+		merr, isME := err.(*mysqldriver.MySQLError)
+		if isME {
+			return merr.Number ==sql.ErrLockDeadlock
+		} else {
+			return false
+		}
+	} else {
+		return false
+	}
+}

driver/mysql/applier_gtid_executed.go

@@ -246,11 +246,9 @@ func (a *GtidExecutedCreater) createTableGtidExecutedV4() error {
 }
 
 func (a *ApplierIncr) cleanGtidExecuted(sid uuid.UUID, txSid string) error {
-	a.logger.Debug("incr. cleanup before WaitForExecution")
-	if !a.mtsManager.WaitForAllCommitted() {
+	if !a.mtsManager.WaitForAllCommitted(a.logger.With("txSid", txSid)) {
 		return nil // shutdown
 	}
-	a.logger.Debug("incr. cleanup after WaitForExecution")
 
 	intervalStr := func() string {
 		a.gtidSetLock.RLock()

driver/mysql/applier_incr.go

@@ -18,7 +18,6 @@ import (
 	gomysql "github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/pkg/errors"
 	uuid "github.com/satori/go.uuid"
-	mysqldriver "github.com/go-sql-driver/mysql"
 )
 
 const (
@@ -236,19 +235,16 @@ func (a *ApplierIncr) MtsWorker(workerIndex int) {
 			logger.Debug("a binlogEntry MTS dequeue", "gno", entryContext.Entry.Coordinates.GetGNO())
 			for iTry := 0; ; iTry++ {
 				err := a.ApplyBinlogEvent(workerIndex, entryContext)
-				if err != nil {
-					if merr, isME := err.(*mysqldriver.MySQLError); isME {
-						if merr.Number == sql.ErrLockDeadlock && iTry < a.mysqlContext.RetryTxLimit {
-							logger.Info("found deadlock. will retry tx", "gno", entryContext.Entry.Coordinates.GetGNO(),
-								"iTry", iTry)
-							time.Sleep(retryTxDelay)
-							continue
-						}
-					}
+				if errIsMysqlDeadlock(err) && iTry < a.mysqlContext.RetryTxLimit {
+					logger.Info("found deadlock. will retry tx", "gno", entryContext.Entry.Coordinates.GetGNO(),
+						"iTry", iTry)
+					time.Sleep(retryTxDelay)
+					continue
+				} else if err != nil {
 					a.OnError(common.TaskStateDead, err) // TODO coordinate with other goroutine
 					keepLoop = false
 				}
-				break;
+				break
 			}
 			logger.Debug("after ApplyBinlogEvent.", "gno", entryContext.Entry.Coordinates.GetGNO())
 		case <-t.C:
@@ -351,11 +347,9 @@ func (a *ApplierIncr) handleEntry(entryCtx *common.EntryContext) (err error) {
 	} else {
 		if binlogEntry.Index == 0 {
 			if rotated {
-				a.logger.Debug("binlog rotated. WaitForAllCommitted before", "file", a.replayingBinlogFile)
-				if !a.mtsManager.WaitForAllCommitted() {
+				if !a.mtsManager.WaitForAllCommitted(a.logger.With("rotate", a.replayingBinlogFile)) {
 					return nil // TODO shutdown
 				}
-				a.logger.Debug("binlog rotated. WaitForAllCommitted after", "file", a.replayingBinlogFile)
 				a.mtsManager.lastCommitted = 0
 				a.mtsManager.lastEnqueue = 0
 				a.wsManager.resetCommonParent(0)
@@ -380,12 +374,11 @@ func (a *ApplierIncr) handleEntry(entryCtx *common.EntryContext) (err error) {
 			hasDDL := binlogEntry.HasDDL()
 			inMiddleDDL := hasDDL || a.prevDDL // DDL must be executed separatedly
 			if inMiddleDDL || isBig {
-				a.logger.Info("WaitForAllCommitted",
-					"gno", txGno, "seq", binlogEntry.Coordinates.GetSequenceNumber(),
-					"lc", binlogEntry.Coordinates.GetLastCommit(), "leq", a.mtsManager.lastEnqueue,
+				if !a.mtsManager.WaitForAllCommitted(a.logger.With("gno", txGno,
+					"seq", binlogEntry.Coordinates.GetSequenceNumber(),
+					"lc", binlogEntry.Coordinates.GetLastCommit(),
 					"hasDDL", hasDDL, "prevDDL", a.prevDDL,
-					"bigtx", isBig, "index", binlogEntry.Index)
-				if !a.mtsManager.WaitForAllCommitted() {
+					"bigtx", isBig, "index", binlogEntry.Index)) {
 					return nil // shutdown
 				}
 			}

driver/mysql/applier_incr_test.go

@@ -1,2 +1,44 @@
 package mysql
 
+import (
+	"github.com/hashicorp/go-hclog"
+	"testing"
+	"time"
+)
+
+func TestMtsManager(t *testing.T) {
+	logger := hclog.Default()
+	shutdownCh := make(chan struct{})
+	defer close(shutdownCh)
+
+	mm := NewMtsManager(shutdownCh, logger)
+	go mm.LcUpdater()
+
+	doneCh := make(chan struct{})
+	go func() {
+		mm.WaitForExecution0(1, 0)
+		logger.Info("wait 1 0")
+		mm.Executed0(1)
+		mm.WaitForExecution0(2, 1)
+		logger.Info("wait 2 1")
+		mm.Executed0(2)
+		mm.WaitForExecution0(3, 2)
+		logger.Info("wait 3 2")
+		mm.Executed0(3)
+
+		// BinlogEntry was resent
+		mm.WaitForExecution0(2, 1)
+		mm.Executed0(2)
+
+		mm.WaitForAllCommitted(logger)
+		close(doneCh)
+	}()
+
+	tm := time.NewTimer(10 * time.Second)
+	select {
+	case <-doneCh:
+		t.Log("case finished")
+	case <-tm.C:
+		t.Fatal("might be stuck")
+	}
+}

driver/mysql/applier_mts.go

@@ -4,6 +4,7 @@ import (
 	"container/heap"
 	"hash/fnv"
 	"sync/atomic"
+	"time"
 
 	"github.com/actiontech/dtle/driver/common"
 	"github.com/actiontech/dtle/g"
@@ -56,14 +57,18 @@ func NewMtsManager(shutdownCh chan struct{}, logger g.LoggerType) *MtsManager {
 }
 
 //  This function must be called sequentially.
-func (mm *MtsManager) WaitForAllCommitted() bool {
-	g.Logger.Debug("WaitForAllCommitted", "lc", mm.lastCommitted, "le", mm.lastEnqueue)
+func (mm *MtsManager) WaitForAllCommitted(logger g.LoggerType) bool {
+	t := time.NewTimer(30 * time.Second)
+	defer t.Stop()
+
 	for {
-		if mm.lastCommitted == mm.lastEnqueue {
+		if atomic.LoadInt64(&mm.lastCommitted) >= mm.lastEnqueue {
 			return true
 		}
 
 		select {
+		case <-t.C: // this will only be triggered once
+			logger.Warn("WaitForAllCommitted has been stuck for 30s")
 		case <-mm.shutdownCh:
 			return false
 		case <-mm.updated:
@@ -75,15 +80,21 @@ func (mm *MtsManager) WaitForAllCommitted() bool {
 // block for waiting. return true for can_execute, false for abortion.
 //  This function must be called sequentially.
 func (mm *MtsManager) WaitForExecution(binlogEntry *common.DataEntry) bool {
-	mm.lastEnqueue = binlogEntry.Coordinates.GetSequenceNumber()
+	return mm.WaitForExecution0(
+		binlogEntry.Coordinates.GetSequenceNumber(),
+		binlogEntry.Coordinates.(*common.MySQLCoordinateTx).LastCommitted)
+}
+
+func (mm *MtsManager) WaitForExecution0(seq int64, lc int64) bool {
+	mm.lastEnqueue = seq
 
 	if mm.forceMts {
 		return true
 	}
 
 	for {
 		currentLC := atomic.LoadInt64(&mm.lastCommitted)
-		if currentLC >= binlogEntry.Coordinates.(*common.MySQLCoordinateTx).LastCommitted {
+		if currentLC >= lc {
 			return true
 		}
 
@@ -130,10 +141,14 @@ func (mm *MtsManager) LcUpdater() {
 }
 
 func (mm *MtsManager) Executed(binlogEntry *common.DataEntry) {
+	mm.Executed0(binlogEntry.Coordinates.GetSequenceNumber())
+}
+
+func (mm *MtsManager) Executed0(seq int64) {
 	select {
 	case <-mm.shutdownCh:
 		return
-	case mm.chExecuted <- binlogEntry.Coordinates.GetSequenceNumber():
+	case mm.chExecuted <- seq:
 	}
 }