Fix binlog end log pos over lost data

go/binlog/gomysql_reader.go

@@ -78,6 +78,24 @@ func (this *GoMySQLReader) GetCurrentBinlogCoordinates() *mysql.BinlogCoordinate
 
 // StreamEvents
 func (this *GoMySQLReader) handleRowsEvent(ev *replication.BinlogEvent, rowsEvent *replication.RowsEvent, entriesChannel chan<- *BinlogEntry) error {
+	if err := func() error {
+		if this.LastAppliedRowsEventHint.IsEmpty() {
+			return nil
+		}
+
+		if this.currentCoordinates.LogFile != this.LastAppliedRowsEventHint.LogFile {
+			return nil
+		}
+
+		if this.LastAppliedRowsEventHint.LogPos+this.currentCoordinates.EventSize >= 1<<32 {
+			// Unexpected rows event, the previous binlog log_pos + current binlog event_size is overflow 4 bytes
+			return fmt.Errorf("Unexpected rows event at %+v, the binlog end_log_pos is overflow 4 bytes", this.currentCoordinates)
+		}
+		return nil
+	}(); err != nil {
+		return err
+	}
+
 	if this.currentCoordinates.SmallerThanOrEquals(&this.LastAppliedRowsEventHint) {
 		this.migrationContext.Log.Debugf("Skipping handled query at %+v", this.currentCoordinates)
 		return nil
@@ -141,6 +159,7 @@ func (this *GoMySQLReader) StreamEvents(canStopStreaming func() bool, entriesCha
 			this.currentCoordinatesMutex.Lock()
 			defer this.currentCoordinatesMutex.Unlock()
 			this.currentCoordinates.LogPos = int64(ev.Header.LogPos)
+			this.currentCoordinates.EventSize = int64(ev.Header.EventSize)
 		}()
 
 		switch binlogEvent := ev.Event.(type) {

go/mysql/binlog.go

@@ -14,8 +14,9 @@ import (
 
 // BinlogCoordinates described binary log coordinates in the form of log file & log position.
 type BinlogCoordinates struct {
-	LogFile string
-	LogPos  int64
+	LogFile   string
+	LogPos    int64
+	EventSize int64
 }
 
 // ParseInstanceKey will parse an InstanceKey from a string representation such as 127.0.0.1:3306