move gtid update after all event handlers are executed (go-mysql-org#379)

It is a common use case that we have a separate goroutine to update gtid set. We have to ensure that gtid saver sees updated gtid only after all the tasks are done in event handlers.

Author: taylorchu

canal/sync.go

@@ -6,7 +6,7 @@ import (
 "time"
 
 "github.com/pingcap/errors"
-"github.com/satori/go.uuid"
+uuid "github.com/satori/go.uuid"
 "github.com/siddontang/go-log/log"
 "github.com/siddontang/go-mysql/mysql"
 "github.com/siddontang/go-mysql/replication"
@@ -92,14 +92,14 @@ func (c *Canal) runSyncBinlog() error {
 }
 continue
 case *replication.XIDEvent:
-if e.GSet != nil {
-c.master.UpdateGTIDSet(e.GSet)
-}
 savePos = true
 // try to save the position later
 if err := c.eventHandler.OnXID(pos); err != nil {
 return errors.Trace(err)
 }
+if e.GSet != nil {
+c.master.UpdateGTIDSet(e.GSet)
+}
 case *replication.MariadbGTIDEvent:
 // try to save the GTID later
 gtid, err := mysql.ParseMariadbGTIDSet(e.GTID.String())
@@ -119,9 +119,6 @@ func (c *Canal) runSyncBinlog() error {
 return errors.Trace(err)
 }
 case *replication.QueryEvent:
-if e.GSet != nil {
-c.master.UpdateGTIDSet(e.GSet)
-}
 var (
 mb [][]byte
 db []byte
@@ -159,6 +156,9 @@ func (c *Canal) runSyncBinlog() error {
 if err = c.eventHandler.OnDDL(pos, e); err != nil {
 return errors.Trace(err)
 }
+if e.GSet != nil {
+c.master.UpdateGTIDSet(e.GSet)
+}
 default:
 continue
 }