canal: remove ErrHandleInterrupted

Author: siddontang

canal/handler.go

@@ -1,17 +1,9 @@
 package canal
 
-import (
-"github.com/juju/errors"
-"github.com/ngaut/log"
-"github.com/siddontang/go-mysql/mysql"
-)
-
-var (
-ErrHandleInterrupted = errors.New("do handler error, interrupted")
-)
+import "github.com/juju/errors"
 
 type RowsEventHandler interface {
-// Handle RowsEvent, if return ErrHandleInterrupted, canal will
+// Handle RowsEvent, if return error, canal will
 // stop the sync
 Do(e *RowsEvent) error
 String() string
@@ -29,13 +21,9 @@ func (c *Canal) travelRowsEventHandler(e *RowsEvent) error {
 
 var err error
 for _, h := range c.rsHandlers {
-if err = h.Do(e); err != nil && !mysql.ErrorEqual(err, ErrHandleInterrupted) {
-log.Errorf("handle %v err: %v", h, err)
-} else if mysql.ErrorEqual(err, ErrHandleInterrupted) {
-log.Errorf("handle %v err, interrupted", h)
-return ErrHandleInterrupted
+if err = h.Do(e); err != nil {
+return errors.Trace(err)
 }
-
 }
 return nil
 }

canal/sync.go

@@ -45,6 +45,7 @@ func (c *Canal) startSyncBinlog() error {
 
 timeout = time.Second
 
+curPos := pos.Pos
 //next binlog pos
 pos.Pos = ev.Header.LogPos
 
@@ -66,7 +67,7 @@ func (c *Canal) startSyncBinlog() error {
 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)
+log.Errorf("handle rows event at (%s, %d) error %v", pos.Name, curPos, err)
 return errors.Trace(err)
 }
 continue