fix Canal.startSyncer and position comparison (go-mysql-org#341)

Author: bytewatch

canal/dump.go

@@ -159,7 +159,7 @@ func (c *Canal) dump() error {
 return errors.Trace(err)
 }
 
-pos := mysql.Position{h.name, uint32(h.pos)}
+pos := mysql.Position{Name: h.name, Pos: uint32(h.pos)}
 c.master.Update(pos)
 c.eventHandler.OnPosSynced(pos, true)
 var startPos fmt.Stringer = pos

canal/master.go

@@ -42,5 +42,8 @@ func (m *masterInfo) GTIDSet() mysql.GTIDSet {
 m.RLock()
 defer m.RUnlock()
 
-return m.gset
+if m.gset == nil {
+return nil
+}
+return m.gset.Clone()
 }

canal/sync.go

@@ -229,13 +229,13 @@ func (c *Canal) WaitUntilPos(pos mysql.Position, timeout time.Duration) error {
 func (c *Canal) GetMasterPos() (mysql.Position, error) {
 rr, err := c.Execute("SHOW MASTER STATUS")
 if err != nil {
-return mysql.Position{"", 0}, errors.Trace(err)
+return mysql.Position{}, errors.Trace(err)
 }
 
 name, _ := rr.GetString(0, 0)
 pos, _ := rr.GetInt(0, 1)
 
-return mysql.Position{name, uint32(pos)}, nil
+return mysql.Position{Name: name, Pos: uint32(pos)}, nil
 }
 
 func (c *Canal) GetMasterGTIDSet() (mysql.GTIDSet, error) {

cmd/go-mysqlbinlog/main.go

@@ -47,7 +47,7 @@ func main() {
 
 b := replication.NewBinlogSyncer(cfg)
 
-pos := mysql.Position{*file, uint32(*pos)}
+pos := mysql.Position{Name: *file, Pos: uint32(*pos)}
 if len(*backupPath) > 0 {
 // Backup will always use RawMode.
 err := b.StartBackup(*backupPath, pos, 0)

failover/mariadb_gtid_handler.go

@@ -118,7 +118,7 @@ func (h *MariadbGTIDHandler) WaitRelayLogDone(s *Server) error {
 fname, _ := r.GetStringByName(0, "Master_Log_File")
 pos, _ := r.GetIntByName(0, "Read_Master_Log_Pos")
 
-return s.MasterPosWait(Position{fname, uint32(pos)}, 0)
+return s.MasterPosWait(Position{Name: fname, Pos: uint32(pos)}, 0)
 }
 
 func (h *MariadbGTIDHandler) WaitCatchMaster(s *Server, m *Server) error {

failover/server.go

@@ -152,7 +152,7 @@ func (s *Server) FetchSlaveReadPos() (Position, error) {
 fname, _ := r.GetStringByName(0, "Master_Log_File")
 pos, _ := r.GetIntByName(0, "Read_Master_Log_Pos")
 
-return Position{fname, uint32(pos)}, nil
+return Position{Name: fname, Pos: uint32(pos)}, nil
 }
 
 // Get current executed binlog filename and position from master
@@ -165,7 +165,7 @@ func (s *Server) FetchSlaveExecutePos() (Position, error) {
 fname, _ := r.GetStringByName(0, "Relay_Master_Log_File")
 pos, _ := r.GetIntByName(0, "Exec_Master_Log_Pos")
 
-return Position{fname, uint32(pos)}, nil
+return Position{Name: fname, Pos: uint32(pos)}, nil
 }
 
 func (s *Server) MasterPosWait(pos Position, timeout int) error {

replication/backup_test.go

@@ -1,41 +1,41 @@
 package replication
 
 import (
-"sync"
-"os"
-"time"
-. "github.com/pingcap/check"
-"github.com/siddontang/go-mysql/mysql"
 "context"
 "github.com/juju/errors"
+. "github.com/pingcap/check"
+"github.com/siddontang/go-mysql/mysql"
+"os"
+"sync"
+"time"
 )
 
 func (t *testSyncerSuite) TestStartBackupEndInGivenTime(c *C) {
 t.setupTest(c, mysql.MySQLFlavor)
-
+
 t.testExecute(c, "RESET MASTER")
-
+
 var wg sync.WaitGroup
 wg.Add(1)
 defer wg.Wait()
-
+
 go func() {
 defer wg.Done()
-
+
 t.testSync(c, nil)
-
+
 t.testExecute(c, "FLUSH LOGS")
-
+
 t.testSync(c, nil)
 }()
-
+
 os.RemoveAll("./var")
 timeout := 2 * time.Second
-
+
 done := make(chan bool)
-
+
 go func() {
-err := t.b.StartBackup("./var", mysql.Position{"", uint32(0)}, timeout)
+err := t.b.StartBackup("./var", mysql.Position{Name: "", Pos: uint32(0)}, timeout)
 c.Assert(err, IsNil)
 done <- true
 }()

replication/binlogsyncer.go

@@ -416,7 +416,7 @@ func (b *BinlogSyncer) writeBinlogDumpCommand(p Position) error {
 }
 
 func (b *BinlogSyncer) writeBinlogDumpMysqlGTIDCommand(gset GTIDSet) error {
-p := Position{"", 4}
+p := Position{Name: "", Pos: 4}
 gtidData := gset.Encode()
 
 b.c.ResetSequence()
@@ -472,7 +472,7 @@ func (b *BinlogSyncer) writeBinlogDumpMariadbGTIDCommand(gset GTIDSet) error {
 }
 
 // Since we use @slave_connect_state, the file and position here are ignored.
-return b.writeBinlogDumpCommand(Position{"", 0})
+return b.writeBinlogDumpCommand(Position{Name: "", Pos: 0})
 }
 
 // localHostname returns the hostname that register slave would register as.
@@ -759,20 +759,10 @@ func (b *BinlogSyncer) parseEvent(s *BinlogStreamer, data []byte) error {
 }
 
 func (b *BinlogSyncer) getGtidSet() GTIDSet {
-var gtidSet GTIDSet
-
 if b.gset == nil {
 return nil
 }
-
-switch b.cfg.Flavor {
-case MariaDBFlavor:
-gtidSet, _ = ParseGTIDSet(MariaDBFlavor, b.gset.String())
-default:
-gtidSet, _ = ParseGTIDSet(MySQLFlavor, b.gset.String())
-}
-
-return gtidSet
+return b.gset.Clone()
 }
 
 // LastConnectionID returns last connectionID.

replication/replication_test.go

@@ -304,7 +304,7 @@ func (t *testSyncerSuite) testPositionSync(c *C) {
 binFile, _ := r.GetString(0, 0)
 binPos, _ := r.GetInt(0, 1)
 
-s, err := t.b.StartSync(mysql.Position{binFile, uint32(binPos)})
+s, err := t.b.StartSync(mysql.Position{Name: binFile, Pos: uint32(binPos)})
 c.Assert(err, IsNil)
 
 // Test re-sync.
@@ -400,7 +400,7 @@ func (t *testSyncerSuite) TestMysqlBinlogCodec(c *C) {
 
 os.RemoveAll(binlogDir)
 
-err := t.b.StartBackup(binlogDir, mysql.Position{"", uint32(0)}, 2*time.Second)
+err := t.b.StartBackup(binlogDir, mysql.Position{Name: "", Pos: uint32(0)}, 2*time.Second)
 c.Assert(err, IsNil)
 
 p := NewBinlogParser()