address comments

Author: cameron-p-m

canal/handler.go

@@ -14,7 +14,7 @@ type EventHandler interface {
 	OnDDL(header *replication.EventHeader, nextPos mysql.Position, queryEvent *replication.QueryEvent) error
 	OnRow(e *RowsEvent) error
 	OnXID(header *replication.EventHeader, nextPos mysql.Position) error
-	OnGTID(header *replication.EventHeader, gtidEvent mysql.GTIDEvent) error
+	OnGTID(header *replication.EventHeader, gtidEvent mysql.BinlogGTIDEvent) error
 	// OnPosSynced Use your own way to sync position. When force is true, sync position immediately.
 	OnPosSynced(header *replication.EventHeader, pos mysql.Position, set mysql.GTIDSet, force bool) error
 	String() string
@@ -32,9 +32,11 @@ func (h *DummyEventHandler) OnTableChanged(*replication.EventHeader, string, str
 func (h *DummyEventHandler) OnDDL(*replication.EventHeader, mysql.Position, *replication.QueryEvent) error {
 	return nil
 }
-func (h *DummyEventHandler) OnRow(*RowsEvent) error                                 { return nil }
-func (h *DummyEventHandler) OnXID(*replication.EventHeader, mysql.Position) error   { return nil }
-func (h *DummyEventHandler) OnGTID(*replication.EventHeader, mysql.GTIDEvent) error { return nil }
+func (h *DummyEventHandler) OnRow(*RowsEvent) error                               { return nil }
+func (h *DummyEventHandler) OnXID(*replication.EventHeader, mysql.Position) error { return nil }
+func (h *DummyEventHandler) OnGTID(*replication.EventHeader, mysql.BinlogGTIDEvent) error {
+	return nil
+}
 func (h *DummyEventHandler) OnPosSynced(*replication.EventHeader, mysql.Position, mysql.GTIDSet, bool) error {
 	return nil
 }

canal/sync.go

@@ -119,7 +119,7 @@ func (c *Canal) runSyncBinlog() error {
 			if err := c.eventHandler.OnGTID(ev.Header, e); err != nil {
 				return errors.Trace(err)
 			}
-		case *replication.MySQLGTIDEvent:
+		case *replication.GTIDEvent:
 			if err := c.eventHandler.OnGTID(ev.Header, e); err != nil {
 				return errors.Trace(err)
 			}

mysql/gtid.go

@@ -30,7 +30,8 @@ func ParseGTIDSet(flavor string, s string) (GTIDSet, error) {
 	}
 }
 
-type GTIDEvent interface {
+type BinlogGTIDEvent interface {
 	GTIDNext() (GTIDSet, error)
-	Decode(data []byte) error
+	SequenceNumber() int64
+	LastCommitted() int64
 }

replication/binlogsyncer.go

@@ -141,7 +141,7 @@ type BinlogSyncer struct {
 	prevGset, currGset GTIDSet
 
 	// instead of GTIDSet.Clone, use this to speed up calculate prevGset
-	prevMySQLGTIDEvent *MySQLGTIDEvent
+	prevMySQLGTIDEvent *GTIDEvent
 
 	running bool
 
@@ -811,7 +811,7 @@ func (b *BinlogSyncer) parseEvent(s *BinlogStreamer, data []byte) error {
 		b.nextPos.Name = string(event.NextLogName)
 		b.nextPos.Pos = uint32(event.Position)
 		b.cfg.Logger.Infof("rotate to %s", b.nextPos)
-	case *MySQLGTIDEvent:
+	case *GTIDEvent:
 		if b.prevGset == nil {
 			break
 		}

replication/event.go

@@ -356,12 +356,12 @@ func (e *QueryEvent) Dump(w io.Writer) {
 	fmt.Fprintln(w)
 }
 
-type MySQLGTIDEvent struct {
+type GTIDEvent struct {
 	CommitFlag     uint8
 	SID            []byte
 	GNO            int64
-	LastCommitted  int64
-	SequenceNumber int64
+	lastCommitted  int64
+	sequenceNumber int64
 
 	// ImmediateCommitTimestamp/OriginalCommitTimestamp are introduced in MySQL-8.0.1, see:
 	// https://mysqlhighavailability.com/replication-features-in-mysql-8-0-1/
@@ -378,7 +378,7 @@ type MySQLGTIDEvent struct {
 	OriginalServerVersion  uint32
 }
 
-func (e *MySQLGTIDEvent) Decode(data []byte) error {
+func (e *GTIDEvent) Decode(data []byte) error {
 	pos := 0
 	e.CommitFlag = data[pos]
 	pos++
@@ -390,9 +390,9 @@ func (e *MySQLGTIDEvent) Decode(data []byte) error {
 	if len(data) >= 42 {
 		if data[pos] == LogicalTimestampTypeCode {
 			pos++
-			e.LastCommitted = int64(binary.LittleEndian.Uint64(data[pos:]))
+			e.lastCommitted = int64(binary.LittleEndian.Uint64(data[pos:]))
 			pos += PartLogicalTimestampLength
-			e.SequenceNumber = int64(binary.LittleEndian.Uint64(data[pos:]))
+			e.sequenceNumber = int64(binary.LittleEndian.Uint64(data[pos:]))
 			pos += 8
 
 			// IMMEDIATE_COMMIT_TIMESTAMP_LENGTH = 7
@@ -441,7 +441,7 @@ func (e *MySQLGTIDEvent) Decode(data []byte) error {
 	return nil
 }
 
-func (e *MySQLGTIDEvent) Dump(w io.Writer) {
+func (e *GTIDEvent) Dump(w io.Writer) {
 	fmtTime := func(t time.Time) string {
 		if t.IsZero() {
 			return "<n/a>"
@@ -452,8 +452,8 @@ func (e *MySQLGTIDEvent) Dump(w io.Writer) {
 	fmt.Fprintf(w, "Commit flag: %d\n", e.CommitFlag)
 	u, _ := uuid.FromBytes(e.SID)
 	fmt.Fprintf(w, "GTID_NEXT: %s:%d\n", u.String(), e.GNO)
-	fmt.Fprintf(w, "LAST_COMMITTED: %d\n", e.LastCommitted)
-	fmt.Fprintf(w, "SEQUENCE_NUMBER: %d\n", e.SequenceNumber)
+	fmt.Fprintf(w, "LAST_COMMITTED: %d\n", e.lastCommitted)
+	fmt.Fprintf(w, "SEQUENCE_NUMBER: %d\n", e.sequenceNumber)
 	fmt.Fprintf(w, "Immediate commmit timestamp: %d (%s)\n", e.ImmediateCommitTimestamp, fmtTime(e.ImmediateCommitTime()))
 	fmt.Fprintf(w, "Orignal commmit timestamp: %d (%s)\n", e.OriginalCommitTimestamp, fmtTime(e.OriginalCommitTime()))
 	fmt.Fprintf(w, "Transaction length: %d\n", e.TransactionLength)
@@ -462,23 +462,31 @@ func (e *MySQLGTIDEvent) Dump(w io.Writer) {
 	fmt.Fprintln(w)
 }
 
-func (e *MySQLGTIDEvent) GTIDNext() (GTIDSet, error) {
+func (e *GTIDEvent) GTIDNext() (GTIDSet, error) {
 	u, err := uuid.FromBytes(e.SID)
 	if err != nil {
 		return nil, err
 	}
 	return ParseMysqlGTIDSet(strings.Join([]string{u.String(), strconv.FormatInt(e.GNO, 10)}, ":"))
 }
 
+func (e *GTIDEvent) SequenceNumber() int64 {
+	return e.sequenceNumber
+}
+
+func (e *GTIDEvent) LastCommitted() int64 {
+	return e.lastCommitted
+}
+
 // ImmediateCommitTime returns the commit time of this trx on the immediate server
 // or zero time if not available.
-func (e *MySQLGTIDEvent) ImmediateCommitTime() time.Time {
+func (e *GTIDEvent) ImmediateCommitTime() time.Time {
 	return microSecTimestampToTime(e.ImmediateCommitTimestamp)
 }
 
 // OriginalCommitTime returns the commit time of this trx on the original server
 // or zero time if not available.
-func (e *MySQLGTIDEvent) OriginalCommitTime() time.Time {
+func (e *GTIDEvent) OriginalCommitTime() time.Time {
 	return microSecTimestampToTime(e.OriginalCommitTimestamp)
 }
 
@@ -637,6 +645,14 @@ func (e *MariadbGTIDEvent) GTIDNext() (GTIDSet, error) {
 	return ParseMariadbGTIDSet(e.GTID.String())
 }
 
+func (e *MariadbGTIDEvent) SequenceNumber() int64 {
+	return int64(e.GTID.SequenceNumber)
+}
+
+func (e *MariadbGTIDEvent) LastCommitted() int64 {
+	return int64(e.GTID.SequenceNumber - 1)
+}
+
 type MariadbGTIDListEvent struct {
 	GTIDs []MariadbGTID
 }

replication/event_test.go

@@ -90,7 +90,7 @@ func TestGTIDEventMysql8NewFields(t *testing.T) {
 	}
 
 	for _, tc := range testcases {
-		ev := new(MySQLGTIDEvent)
+		ev := new(GTIDEvent)
 		err := ev.Decode(tc.data)
 		require.NoError(t, err)
 		require.Equal(t, tc.expectImmediateCommitTimestamp, ev.ImmediateCommitTimestamp)

replication/parser.go

@@ -283,9 +283,9 @@ func (p *BinlogParser) parseEvent(h *EventHeader, data []byte, rawData []byte) (
 			case ROWS_QUERY_EVENT:
 				e = &RowsQueryEvent{}
 			case GTID_EVENT:
-				e = &MySQLGTIDEvent{}
+				e = &GTIDEvent{}
 			case ANONYMOUS_GTID_EVENT:
-				e = &MySQLGTIDEvent{}
+				e = &GTIDEvent{}
 			case BEGIN_LOAD_QUERY_EVENT:
 				e = &BeginLoadQueryEvent{}
 			case EXECUTE_LOAD_QUERY_EVENT: