fix canal test (go-mysql-org#118)

Author: siddontang

canal/canal_test.go

@@ -6,6 +6,7 @@ import (
 "fmt"
 "os"
 "testing"
+"time"
 
 "github.com/ngaut/log"
 . "github.com/pingcap/check"
@@ -92,7 +93,7 @@ func (s *canalTestSuite) TestCanal(c *C) {
 s.execute(c, "ALTER TABLE test.canal_test ADD `age` INT(5) NOT NULL AFTER `name`")
 s.execute(c, "INSERT INTO test.canal_test (name,age) VALUES (?,?)", "d", "18")
 
-err := s.c.CatchMasterPos(100)
+err := s.c.CatchMasterPos(10 * time.Second)
 c.Assert(err, IsNil)
 }
 
@@ -101,8 +102,8 @@ func TestAlterTableExp(t *testing.T) {
 "ALTER TABLE `mydb`.`mytable` ADD `field2` DATE NULL AFTER `field1`;",
 "ALTER TABLE `mytable` ADD `field2` DATE NULL AFTER `field1`;",
 "ALTER TABLE mydb.mytable ADD `field2` DATE NULL AFTER `field1`;",
-"ALTER table mytable ADD `field2` DATE NULL AFTER `field1`;",
-"alter TABLE mydb.mytable ADD field2 DATE NULL AFTER `field1`;",
+"ALTER TABLE mytable ADD `field2` DATE NULL AFTER `field1`;",
+"ALTER TABLE mydb.mytable ADD field2 DATE NULL AFTER `field1`;",
 }
 
 table := []byte("mytable")

canal/master.go

@@ -20,7 +20,7 @@ type masterInfo struct {
 
 name string
 
-l sync.Mutex
+l sync.RWMutex
 
 lastSaveTime time.Time
 }
@@ -68,6 +68,7 @@ func (m *masterInfo) Save(force bool) error {
 }
 
 func (m *masterInfo) Update(name string, pos uint32) {
+log.Debugf("update master position (%s, %d)", name, pos)
 m.l.Lock()
 m.Name = name
 m.Position = pos
@@ -76,10 +77,10 @@ func (m *masterInfo) Update(name string, pos uint32) {
 
 func (m *masterInfo) Pos() mysql.Position {
 var pos mysql.Position
-m.l.Lock()
+m.l.RLock()
 pos.Name = m.Name
 pos.Pos = m.Position
-m.l.Unlock()
+m.l.RUnlock()
 
 return pos
 }

canal/sync.go

@@ -10,6 +10,7 @@ import (
 "github.com/ngaut/log"
 "github.com/siddontang/go-mysql/mysql"
 "github.com/siddontang/go-mysql/replication"
+"github.com/siddontang/go-mysql/schema"
 )
 
 var (
@@ -62,7 +63,9 @@ func (c *Canal) startSyncBinlog() error {
 log.Infof("rotate binlog to %v", pos)
 case *replication.RowsEvent:
 // we only focus row based event
-if err = c.handleRowsEvent(ev); err != nil {
+err = c.handleRowsEvent(ev)
+if err != nil && errors.Cause(err) != schema.ErrTableNotExist {
+// We can ignore table not exist error
 log.Errorf("handle rows event error %v", err)
 return errors.Trace(err)
 }
@@ -119,21 +122,18 @@ func (c *Canal) handleRowsEvent(e *replication.BinlogEvent) error {
 return c.travelRowsEventHandler(events)
 }
 
-func (c *Canal) WaitUntilPos(pos mysql.Position, timeout int) error {
-if timeout <= 0 {
-timeout = 60
-}
-
-timer := time.NewTimer(time.Duration(timeout) * time.Second)
+func (c *Canal) WaitUntilPos(pos mysql.Position, timeout time.Duration) error {
+timer := time.NewTimer(timeout)
 for {
 select {
 case <-timer.C:
-return errors.Errorf("wait position %v err", pos)
+return errors.Errorf("wait position %v too long > %s", pos, timeout)
 default:
-curpos := c.master.Pos()
-if curpos.Compare(pos) >= 0 {
+curPos := c.master.Pos()
+if curPos.Compare(pos) >= 0 {
 return nil
 } else {
+log.Debugf("master pos is %v, wait catching %v", curPos, pos)
 time.Sleep(100 * time.Millisecond)
 }
 }
@@ -142,7 +142,7 @@ func (c *Canal) WaitUntilPos(pos mysql.Position, timeout int) error {
 return nil
 }
 
-func (c *Canal) CatchMasterPos(timeout int) error {
+func (c *Canal) CatchMasterPos(timeout time.Duration) error {
 rr, err := c.Execute("SHOW MASTER STATUS")
 if err != nil {
 return errors.Trace(err)

schema/schema.go

@@ -12,6 +12,8 @@ import (
 "github.com/siddontang/go-mysql/mysql"
 )
 
+var ErrTableNotExist = errors.New("table is not exist")
+
 const (
 TYPE_NUMBER = iota + 1 // tinyint, smallint, mediumint, int, bigint, year
 TYPE_FLOAT // float, double
@@ -144,20 +146,39 @@ func (idx *Index) FindColumn(name string) int {
 return -1
 }
 
+func isTableExist(conn mysql.Executer, schema string, name string) (bool, error) {
+query := fmt.Sprintf("SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = '%s' and TABLE_NAME = '%s' LIMIT 1", schema, name)
+r, err := conn.Execute(query)
+if err != nil {
+return false, errors.Trace(err)
+}
+
+return r.RowNumber() == 1, nil
+}
+
 func NewTable(conn mysql.Executer, schema string, name string) (*Table, error) {
+ok, err := isTableExist(conn, schema, name)
+if err != nil {
+return nil, errors.Trace(err)
+}
+
+if !ok {
+return nil, ErrTableNotExist
+}
+
 ta := &Table{
 Schema: schema,
 Name: name,
 Columns: make([]TableColumn, 0, 16),
 Indexes: make([]*Index, 0, 8),
 }
 
-if err := ta.fetchColumns(conn); err != nil {
-return nil, err
+if err = ta.fetchColumns(conn); err != nil {
+return nil, errors.Trace(err)
 }
 
-if err := ta.fetchIndexes(conn); err != nil {
-return nil, err
+if err = ta.fetchIndexes(conn); err != nil {
+return nil, errors.Trace(err)
 }
 
 return ta, nil