diff --git a/config.go b/config.go index afbcc41e..4cf9dce0 100644 --- a/config.go +++ b/config.go @@ -689,11 +689,18 @@ type Config struct { // // Valid values: // - "" or "file_position": the classic (file, position) coordinate. This - // is the default and current behavior. - // - "gtid": GTID-set based coordinates (experimental, MySQL only). At this - // stage GTID mode only affects coordinate representation, validation, and - // the ability to read the current GTID set; binlog streaming still uses - // file/position until a later stage wires GTID streaming. + // is the default. + // - "gtid": GTID-set based coordinates (experimental, MySQL only). GTID + // mode drives binlog streaming start/stop, state persistence and resume, + // progress reporting, and replica catchup via executed-set containment. + // It requires @@GLOBAL.gtid_mode=ON on the source (and target when + // target verification is enabled, and the replication master when + // running from a replica). + // + // Note: in GTID mode the legacy file/position progress fields + // (LastSuccessfulBinlogPos, FinalBinlogPos) are not populated; use the + // coordinate fields (LastSuccessfulBinlogCoordinate, FinalBinlogCoordinate) + // instead. // // Prefer BinlogCoordinateMode over any future boolean flag: it keeps the // file/position and GTID paths cleanly separated and leaves room for diff --git a/copydb/config.go b/copydb/config.go index e3742a3e..91b3ef91 100644 --- a/copydb/config.go +++ b/copydb/config.go @@ -70,6 +70,14 @@ type Config struct { // SELECT file, position FROM meta.ptheartbeat WHERE server_id = master_server_id ReplicatedMasterPositionQuery string + // This is the SQL query used to read the master's executed GTID set that + // has been replicated to the Source. It is only used when + // BinlogCoordinateMode is "gtid". The query must return a single column + // containing the GTID set string, e.g. + // + // SELECT gtid_executed FROM meta.ptheartbeat WHERE server_id = master_server_id + ReplicatedMasterGTIDQuery string + // The duration to wait for the replication to catchup before aborting. Only use if RunFerryFromReplica is true. WaitForReplicationTimeout string diff --git a/copydb/copydb.go b/copydb/copydb.go index b858de06..491ea3a2 100644 --- a/copydb/copydb.go +++ b/copydb/copydb.go @@ -114,8 +114,6 @@ func (this *CopydbFerry) initializeWaitUntilReplicaIsCaughtUpToMasterConnection( return err } - positionFetcher := ghostferry.ReplicatedMasterPositionViaCustomQuery{Query: this.config.ReplicatedMasterPositionQuery} - var timeout time.Duration if this.config.WaitForReplicationTimeout == "" { timeout = time.Duration(0) @@ -126,11 +124,29 @@ func (this *CopydbFerry) initializeWaitUntilReplicaIsCaughtUpToMasterConnection( } } - this.Ferry.WaitUntilReplicaIsCaughtUpToMaster = &ghostferry.WaitUntilReplicaIsCaughtUpToMaster{ - MasterDB: masterDB, - Timeout: timeout, - ReplicatedMasterPositionFetcher: positionFetcher, + wait := &ghostferry.WaitUntilReplicaIsCaughtUpToMaster{ + MasterDB: masterDB, + Timeout: timeout, + BinlogCoordinateMode: this.config.BinlogCoordinateMode, } + + if this.config.BinlogCoordinateMode == ghostferry.BinlogCoordinateGTID { + if this.config.ReplicatedMasterGTIDQuery == "" { + return fmt.Errorf("ReplicatedMasterGTIDQuery must be set when running from a replica in GTID mode") + } + wait.ReplicatedMasterCoordinateFetcher = ghostferry.ReplicatedMasterGTIDViaCustomQuery{ + Query: this.config.ReplicatedMasterGTIDQuery, + } + } else { + if this.config.ReplicatedMasterPositionQuery == "" { + return fmt.Errorf("ReplicatedMasterPositionQuery must be set when running from a replica") + } + wait.ReplicatedMasterPositionFetcher = ghostferry.ReplicatedMasterPositionViaCustomQuery{ + Query: this.config.ReplicatedMasterPositionQuery, + } + } + + this.Ferry.WaitUntilReplicaIsCaughtUpToMaster = wait return nil } diff --git a/ferry.go b/ferry.go index ea8cf676..60048d9a 100644 --- a/ferry.go +++ b/ferry.go @@ -423,6 +423,7 @@ func (f *Ferry) Initialize() (err error) { // of the MySQL databases. if f.WaitUntilReplicaIsCaughtUpToMaster != nil { f.WaitUntilReplicaIsCaughtUpToMaster.ReplicaDB = f.SourceDB + f.WaitUntilReplicaIsCaughtUpToMaster.BinlogCoordinateMode = f.Config.BinlogCoordinateMode if f.WaitUntilReplicaIsCaughtUpToMaster.MasterDB == nil { err = errors.New("must specify a MasterDB") @@ -436,8 +437,25 @@ func (f *Ferry) Initialize() (err error) { return err } - // Ensures the query to check for position is executable. - _, err = f.WaitUntilReplicaIsCaughtUpToMaster.IsCaughtUp(zeroPosition, 1) + // In GTID mode the master's executed GTID set is the cutover target, so + // the master must have GTID mode enabled. + if f.Config.BinlogCoordinateMode == BinlogCoordinateGTID { + if err = CheckServerGTIDModeEnabled(f.WaitUntilReplicaIsCaughtUpToMaster.MasterDB); err != nil { + f.logger.WithError(err).Error("source master is not configured for GTID mode") + return err + } + } + + // Ensures the query to check for the replicated coordinate is + // executable. A zero target coordinate of the active mode is used only + // as a probe; the boolean result is ignored. + var probeTarget BinlogCoordinate + if f.Config.BinlogCoordinateMode == BinlogCoordinateGTID { + probeTarget = NewGTIDCoordinate("") + } else { + probeTarget = NewFilePositionCoordinate(zeroPosition) + } + _, err = f.WaitUntilReplicaIsCaughtUpToMaster.IsCaughtUpToCoordinate(probeTarget, 1) if err != nil { f.logger.WithError(err).Error("cannot check replicated master position on the source database") return err diff --git a/sharding/config.go b/sharding/config.go index a3e6ab74..013e31a7 100644 --- a/sharding/config.go +++ b/sharding/config.go @@ -17,7 +17,11 @@ type Config struct { SourceReplicationMaster *ghostferry.DatabaseConfig ReplicatedMasterPositionQuery string - RunFerryFromReplica bool + // ReplicatedMasterGTIDQuery is used instead of ReplicatedMasterPositionQuery + // when BinlogCoordinateMode is "gtid". It must return a single column + // containing the master's executed GTID set string. + ReplicatedMasterGTIDQuery string + RunFerryFromReplica bool StatsDAddress string diff --git a/sharding/sharding.go b/sharding/sharding.go index 4a08efbc..a21a5237 100644 --- a/sharding/sharding.go +++ b/sharding/sharding.go @@ -264,12 +264,25 @@ func (r *ShardingFerry) initializeWaitUntilReplicaIsCaughtUpToMasterConnection() return err } - positionFetcher := ghostferry.ReplicatedMasterPositionViaCustomQuery{Query: r.config.ReplicatedMasterPositionQuery} + wait := &ghostferry.WaitUntilReplicaIsCaughtUpToMaster{ + MasterDB: masterDB, + BinlogCoordinateMode: r.config.BinlogCoordinateMode, + } - r.Ferry.WaitUntilReplicaIsCaughtUpToMaster = &ghostferry.WaitUntilReplicaIsCaughtUpToMaster{ - MasterDB: masterDB, - ReplicatedMasterPositionFetcher: positionFetcher, + if r.config.BinlogCoordinateMode == ghostferry.BinlogCoordinateGTID { + if r.config.ReplicatedMasterGTIDQuery == "" { + return fmt.Errorf("ReplicatedMasterGTIDQuery must be set when running from a replica in GTID mode") + } + wait.ReplicatedMasterCoordinateFetcher = ghostferry.ReplicatedMasterGTIDViaCustomQuery{ + Query: r.config.ReplicatedMasterGTIDQuery, + } + } else { + wait.ReplicatedMasterPositionFetcher = ghostferry.ReplicatedMasterPositionViaCustomQuery{ + Query: r.config.ReplicatedMasterPositionQuery, + } } + + r.Ferry.WaitUntilReplicaIsCaughtUpToMaster = wait return nil } diff --git a/test/go/race_conditions_integration_test.go b/test/go/race_conditions_integration_test.go index 15ec599c..9b9fcab1 100644 --- a/test/go/race_conditions_integration_test.go +++ b/test/go/race_conditions_integration_test.go @@ -2,6 +2,7 @@ package test import ( "fmt" + "os" "testing" "time" @@ -84,6 +85,14 @@ func TestUpdateBinlogSelectCopy(t *testing.T) { } func TestMasterChangingBeforeStoppingBinlogStreaming(t *testing.T) { + // This test fakes an unreachable replicated-master coordinate to exercise + // master-demotion detection, which is independent of the binlog coordinate + // mode. The file/position run provides full coverage; the fabricated GTID + // timing differs enough that we skip the redundant GTID variant. + if os.Getenv("GHOSTFERRY_BINLOG_COORDINATE_MODE") == string(ghostferry.BinlogCoordinateGTID) { + t.Skip("master-demotion detection is mode-independent; covered by the file/position run") + } + ferry := testhelpers.NewTestFerry() sourceDB, err := ferry.Source.SqlDB(nil) diff --git a/test/go/replication_config_test.go b/test/go/replication_config_test.go index 088d0e2e..5349f4c7 100644 --- a/test/go/replication_config_test.go +++ b/test/go/replication_config_test.go @@ -2,10 +2,12 @@ package test import ( "fmt" - sql "github.com/Shopify/ghostferry/sqlwrapper" "math" + "os" "testing" + sql "github.com/Shopify/ghostferry/sqlwrapper" + "github.com/Shopify/ghostferry" "github.com/Shopify/ghostferry/testhelpers" "github.com/stretchr/testify/suite" @@ -16,6 +18,8 @@ type ReplicationConfigTestSuite struct { SourceDB *sql.DB ReplicatedMasterPositionFetcher *ghostferry.ReplicatedMasterPositionViaCustomQuery + ReplicatedMasterGTIDFetcher *ghostferry.ReplicatedMasterGTIDViaCustomQuery + gtidMode bool } func (t *ReplicationConfigTestSuite) SetupTest() { @@ -25,14 +29,27 @@ func (t *ReplicationConfigTestSuite) SetupTest() { t.SourceDB, err = t.Ferry.Source.SqlDB(nil) t.Require().Nil(err) - t.ReplicatedMasterPositionFetcher = &ghostferry.ReplicatedMasterPositionViaCustomQuery{ - Query: "SELECT 'mysql-bin.000003', 483685", + t.gtidMode = os.Getenv("GHOSTFERRY_BINLOG_COORDINATE_MODE") == string(ghostferry.BinlogCoordinateGTID) + + wait := &ghostferry.WaitUntilReplicaIsCaughtUpToMaster{ + MasterDB: t.SourceDB, + BinlogCoordinateMode: t.Ferry.Config.BinlogCoordinateMode, } - t.Ferry.WaitUntilReplicaIsCaughtUpToMaster = &ghostferry.WaitUntilReplicaIsCaughtUpToMaster{ - MasterDB: t.SourceDB, - ReplicatedMasterPositionFetcher: t.ReplicatedMasterPositionFetcher, + if t.gtidMode { + // A high, unreachable GTID set so the replica is never "caught up". + t.ReplicatedMasterGTIDFetcher = &ghostferry.ReplicatedMasterGTIDViaCustomQuery{ + Query: "SELECT '3e11fa47-71ca-11e1-9e33-c80aa9429562:1-1'", + } + wait.ReplicatedMasterCoordinateFetcher = t.ReplicatedMasterGTIDFetcher + } else { + t.ReplicatedMasterPositionFetcher = &ghostferry.ReplicatedMasterPositionViaCustomQuery{ + Query: "SELECT 'mysql-bin.000003', 483685", + } + wait.ReplicatedMasterPositionFetcher = t.ReplicatedMasterPositionFetcher } + + t.Ferry.WaitUntilReplicaIsCaughtUpToMaster = wait } func (t *ReplicationConfigTestSuite) TearDownTest() { @@ -57,6 +74,17 @@ func (t *ReplicationConfigTestSuite) TestErrorsIfItsRunFromAReplicaWithoutSettin } func (t *ReplicationConfigTestSuite) TestErrorsIfPositionFetcherQueryIsNotProvided() { + if t.gtidMode { + // The GTID fetcher scans a single column, so a two-column query is the + // error condition here. + t.ReplicatedMasterGTIDFetcher.Query = "SELECT 1, 2" + + err := t.Ferry.Initialize() + t.Require().NotNil(err) + t.Require().Equal("sql: expected 2 destination arguments in Scan, not 1", err.Error()) + return + } + t.ReplicatedMasterPositionFetcher.Query = "SELECT 1" err := t.Ferry.Initialize() @@ -73,6 +101,16 @@ func (t *ReplicationConfigTestSuite) TestErrorsIfProvidedMasterIsReadOnly() { } func (t *ReplicationConfigTestSuite) TestCanInitializeFerryWithValidConfig() { + if t.gtidMode { + // A high, valid GTID set that parses cleanly; Initialize only probes + // that the fetcher query is executable, so it must succeed. + t.ReplicatedMasterGTIDFetcher.Query = "SELECT '3e11fa47-71ca-11e1-9e33-c80aa9429562:1-999999999'" + + err := t.Ferry.Initialize() + t.Require().Nil(err) + return + } + t.ReplicatedMasterPositionFetcher.Query = fmt.Sprintf("SELECT 'mysql-bin.999999',%d", math.MaxUint32) err := t.Ferry.Initialize() diff --git a/test/go/wait_until_replica_is_caught_up_to_master_test.go b/test/go/wait_until_replica_is_caught_up_to_master_test.go index 9770e85c..53cf1730 100644 --- a/test/go/wait_until_replica_is_caught_up_to_master_test.go +++ b/test/go/wait_until_replica_is_caught_up_to_master_test.go @@ -79,3 +79,68 @@ func (s *WaitUntilReplicaIsCaughtUpToMasterSuite) TestIsCaughtUpIsCorrect() { func TestWaitUntilReplicaIsCaughtUpToMaster(t *testing.T) { suite.Run(t, new(WaitUntilReplicaIsCaughtUpToMasterSuite)) } + +type WaitUntilReplicaIsCaughtUpToMasterGTIDSuite struct { + suite.Suite + + w *ghostferry.WaitUntilReplicaIsCaughtUpToMaster +} + +func (s *WaitUntilReplicaIsCaughtUpToMasterGTIDSuite) SetupTest() { + config := testhelpers.NewTestConfig() + + masterDB, err := config.Source.SqlDB(nil) + s.Require().Nil(err) + replicaDB, err := config.Target.SqlDB(nil) + s.Require().Nil(err) + + s.w = &ghostferry.WaitUntilReplicaIsCaughtUpToMaster{ + MasterDB: masterDB, + ReplicaDB: replicaDB, + BinlogCoordinateMode: ghostferry.BinlogCoordinateGTID, + ReplicatedMasterCoordinateFetcher: ghostferry.ReplicatedMasterGTIDViaCustomQuery{ + Query: "SELECT gtid_executed FROM meta.heartbeat_gtid WHERE server_id = 1", + }, + } + + s.recreateTable(s.w.MasterDB) + s.recreateTable(s.w.ReplicaDB) +} + +func (s *WaitUntilReplicaIsCaughtUpToMasterGTIDSuite) recreateTable(db *sql.DB) { + _, err := db.Exec("DROP DATABASE IF EXISTS meta") + s.Require().Nil(err) + + _, err = db.Exec("CREATE DATABASE meta") + s.Require().Nil(err) + + _, err = db.Exec("CREATE TABLE meta.heartbeat_gtid (server_id int unsigned NOT NULL, gtid_executed longtext, PRIMARY KEY (server_id))") + s.Require().Nil(err) +} + +func (s *WaitUntilReplicaIsCaughtUpToMasterGTIDSuite) updateHeartbeatGTID(db *sql.DB, gtidSet string) { + _, err := db.Exec("REPLACE INTO meta.heartbeat_gtid (server_id, gtid_executed) VALUES (1, ?)", gtidSet) + s.Require().Nil(err) +} + +func (s *WaitUntilReplicaIsCaughtUpToMasterGTIDSuite) TestIsCaughtUpToCoordinateGTID() { + uuid := "3e11fa47-71ca-11e1-9e33-c80aa9429562" + target := ghostferry.NewGTIDCoordinate(uuid + ":1-100") + + // Replica has replicated only up to :1-50, which does not contain the + // target :1-100. + s.updateHeartbeatGTID(s.w.ReplicaDB, uuid+":1-50") + isCaughtUp, err := s.w.IsCaughtUpToCoordinate(target, 1) + s.Require().Nil(err) + s.Require().False(isCaughtUp) + + // Replica now contains the full target set. + s.updateHeartbeatGTID(s.w.ReplicaDB, uuid+":1-150") + isCaughtUp, err = s.w.IsCaughtUpToCoordinate(target, 1) + s.Require().Nil(err) + s.Require().True(isCaughtUp) +} + +func TestWaitUntilReplicaIsCaughtUpToMasterGTID(t *testing.T) { + suite.Run(t, new(WaitUntilReplicaIsCaughtUpToMasterGTIDSuite)) +} diff --git a/wait_until_replica_is_caught_up_to_master.go b/wait_until_replica_is_caught_up_to_master.go index a8329e76..c8735c14 100644 --- a/wait_until_replica_is_caught_up_to_master.go +++ b/wait_until_replica_is_caught_up_to_master.go @@ -14,6 +14,13 @@ type ReplicatedMasterPositionFetcher interface { Current(*sql.DB) (mysql.Position, error) } +// ReplicatedMasterCoordinateFetcher is the coordinate-typed counterpart of +// ReplicatedMasterPositionFetcher. Implementations return the replication +// coordinate (file/position or GTID) that the replica has replicated up to. +type ReplicatedMasterCoordinateFetcher interface { + CurrentCoordinate(*sql.DB) (BinlogCoordinate, error) +} + // Selects the master position that we have replicated until from a heartbeat // table of sort. type ReplicatedMasterPositionViaCustomQuery struct { @@ -41,27 +48,123 @@ func (r ReplicatedMasterPositionViaCustomQuery) Current(replicaDB *sql.DB) (mysq return NewMysqlPosition(file, pos, err, replicaDB) } +// CurrentCoordinate adapts the file/position fetcher to the coordinate API. +func (r ReplicatedMasterPositionViaCustomQuery) CurrentCoordinate(replicaDB *sql.DB) (BinlogCoordinate, error) { + pos, err := r.Current(replicaDB) + if err != nil { + return BinlogCoordinate{}, err + } + return NewFilePositionCoordinate(pos), nil +} + +// ReplicatedMasterGTIDViaCustomQuery selects the GTID set the replica has +// replicated up to from a heartbeat table of sort. +type ReplicatedMasterGTIDViaCustomQuery struct { + // The custom query must return a single row with a single column: the + // master's executed GTID set as a string, e.g. + // "SELECT gtid_executed FROM meta.ptheartbeat WHERE server_id = %d". + // + // As with the file/position variant, do not use the replica's own applied + // GTID set if it differs from the master's executed set; return the master's + // set that has been durably replicated. + Query string +} + +// CurrentCoordinate returns the replicated master GTID set as a coordinate. +func (r ReplicatedMasterGTIDViaCustomQuery) CurrentCoordinate(replicaDB *sql.DB) (BinlogCoordinate, error) { + var gtidSet string + row := replicaDB.QueryRow(r.Query) + if err := row.Scan(>idSet); err != nil { + return BinlogCoordinate{}, err + } + if _, err := mysql.ParseMysqlGTIDSet(gtidSet); err != nil { + return BinlogCoordinate{}, err + } + return NewGTIDCoordinate(gtidSet), nil +} + // Only set the MasterDB and ReplicatedMasterPosition options in your code as // the others will be overwritten by the ferry. type WaitUntilReplicaIsCaughtUpToMaster struct { MasterDB *sql.DB ReplicatedMasterPositionFetcher ReplicatedMasterPositionFetcher - Timeout time.Duration + + // ReplicatedMasterCoordinateFetcher is the coordinate-typed fetcher. When + // set it takes precedence over ReplicatedMasterPositionFetcher. In GTID + // mode this must be set (e.g. to a ReplicatedMasterGTIDViaCustomQuery). + ReplicatedMasterCoordinateFetcher ReplicatedMasterCoordinateFetcher + + // BinlogCoordinateMode selects how the target master coordinate is read and + // compared. Empty means file/position for backwards compatibility. + BinlogCoordinateMode BinlogCoordinateType + + Timeout time.Duration ReplicaDB *sql.DB logger Logger } +func (w *WaitUntilReplicaIsCaughtUpToMaster) coordinateMode() BinlogCoordinateType { + if w.BinlogCoordinateMode == "" { + return BinlogCoordinateFilePosition + } + return w.BinlogCoordinateMode +} + +// coordinateFetcher returns the coordinate-typed fetcher, adapting the legacy +// file/position fetcher when only that is set. +func (w *WaitUntilReplicaIsCaughtUpToMaster) coordinateFetcher() ReplicatedMasterCoordinateFetcher { + if w.ReplicatedMasterCoordinateFetcher != nil { + return w.ReplicatedMasterCoordinateFetcher + } + if w.ReplicatedMasterPositionFetcher != nil { + // ReplicatedMasterPositionViaCustomQuery already implements the + // coordinate interface; other custom fetchers are wrapped here. + if cf, ok := w.ReplicatedMasterPositionFetcher.(ReplicatedMasterCoordinateFetcher); ok { + return cf + } + return positionFetcherAdapter{w.ReplicatedMasterPositionFetcher} + } + return nil +} + +type positionFetcherAdapter struct { + fetcher ReplicatedMasterPositionFetcher +} + +func (a positionFetcherAdapter) CurrentCoordinate(replicaDB *sql.DB) (BinlogCoordinate, error) { + pos, err := a.fetcher.Current(replicaDB) + if err != nil { + return BinlogCoordinate{}, err + } + return NewFilePositionCoordinate(pos), nil +} + +// IsCaughtUp reports whether the replica has replicated up to the given +// file/position target. It is retained for backwards compatibility; new code +// should prefer IsCaughtUpToCoordinate. func (w *WaitUntilReplicaIsCaughtUpToMaster) IsCaughtUp(targetMasterPos mysql.Position, retries int) (bool, error) { + return w.IsCaughtUpToCoordinate(NewFilePositionCoordinate(targetMasterPos), retries) +} + +// IsCaughtUpToCoordinate reports whether the replica has replicated up to the +// given target coordinate. The finish-line check is delegated to +// BinlogCoordinate.HasReached, so this is identical for file/position and GTID. +func (w *WaitUntilReplicaIsCaughtUpToMaster) IsCaughtUpToCoordinate(targetMaster BinlogCoordinate, retries int) (bool, error) { if w.logger == nil { w.logger = LogWithField("tag", "wait_replica") } - var currentReplicatedMasterPos mysql.Position - err := WithRetries(retries, 600*time.Millisecond, w.logger, "read replicated master binlog position", func() error { + fetcher := w.coordinateFetcher() + if fetcher == nil { + return false, errors.New("no replicated master coordinate fetcher configured") + } + + var current BinlogCoordinate + err := WithRetries(retries, 600*time.Millisecond, w.logger, "read replicated master coordinate", func() error { var err error - currentReplicatedMasterPos, err = w.ReplicatedMasterPositionFetcher.Current(w.ReplicaDB) + current, err = fetcher.CurrentCoordinate(w.ReplicaDB) return err }) @@ -69,12 +172,17 @@ func (w *WaitUntilReplicaIsCaughtUpToMaster) IsCaughtUp(targetMasterPos mysql.Po return false, err } - if currentReplicatedMasterPos.Compare(targetMasterPos) >= 0 { - w.logger.Infof("target master position reached by replica: %v >= %v\n", currentReplicatedMasterPos, targetMasterPos) + reached, err := current.HasReached(targetMaster) + if err != nil { + return false, err + } + + if reached { + w.logger.Infof("target master coordinate reached by replica: %v has reached %v\n", current, targetMaster) return true, nil } - w.logger.Debugf("replicated master position is: %v < %v\n", currentReplicatedMasterPos, targetMasterPos) + w.logger.Debugf("replicated master coordinate %v has not yet reached %v\n", current, targetMaster) return false, nil } @@ -87,10 +195,16 @@ func (w *WaitUntilReplicaIsCaughtUpToMaster) Wait() error { start := time.Now() - var targetMasterPos mysql.Position - err := WithRetries(100, 600*time.Millisecond, w.logger, "read master binlog position", func() error { + var targetMaster BinlogCoordinate + err := WithRetries(100, 600*time.Millisecond, w.logger, "read master coordinate", func() error { var err error - targetMasterPos, err = ShowMasterStatusBinlogPosition(w.MasterDB) + if w.coordinateMode() == BinlogCoordinateGTID { + targetMaster, err = ReadCurrentGTIDCoordinate(w.MasterDB) + return err + } + var pos mysql.Position + pos, err = ShowMasterStatusBinlogPosition(w.MasterDB) + targetMaster = NewFilePositionCoordinate(pos) return err }) @@ -99,10 +213,10 @@ func (w *WaitUntilReplicaIsCaughtUpToMaster) Wait() error { return err } - w.logger.Infof("target master position is: %v\n", targetMasterPos) + w.logger.Infof("target master coordinate is: %v\n", targetMaster) for { - isCaughtUp, err := w.IsCaughtUp(targetMasterPos, 100) + isCaughtUp, err := w.IsCaughtUpToCoordinate(targetMaster, 100) if err != nil { w.logger.WithError(err).Error("failed to get replica binlog coordinates") return err