Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix binlog end log pos over lost data #1367

Merged
merged 14 commits into from
Mar 12, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions go/binlog/gomysql_reader.go
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,10 @@ func (this *GoMySQLReader) GetCurrentBinlogCoordinates() *mysql.BinlogCoordinate

// StreamEvents
func (this *GoMySQLReader) handleRowsEvent(ev *replication.BinlogEvent, rowsEvent *replication.RowsEvent, entriesChannel chan<- *BinlogEntry) error {
if this.currentCoordinates.IsLogPosOverflowBeyond4Bytes(&this.LastAppliedRowsEventHint) {
return fmt.Errorf("Unexpected rows event at %+v, the binlog end_log_pos is overflow 4 bytes", this.currentCoordinates)
}

if this.currentCoordinates.SmallerThanOrEquals(&this.LastAppliedRowsEventHint) {
this.migrationContext.Log.Debugf("Skipping handled query at %+v", this.currentCoordinates)
return nil
Expand Down Expand Up @@ -141,6 +145,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) {
Expand Down
29 changes: 27 additions & 2 deletions go/mysql/binlog.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -74,3 +75,27 @@ func (this *BinlogCoordinates) SmallerThanOrEquals(other *BinlogCoordinates) boo
}
return this.LogFile == other.LogFile && this.LogPos == other.LogPos
}

// IsLogPosOverflowBeyond4Bytes returns true if the coordinate endpos is overflow beyond 4 bytes.
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@shaohk last request, can we add a reference to why a value over 4 bytes is an overflow?

Copy link
Contributor Author

@shaohk shaohk Feb 26, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@timvaillancourt OK, I have added. Please take a look again, Thks.

// The binlog event end_log_pos field type is defined as uint32, 4 bytes.
// https://github.com/go-mysql-org/go-mysql/blob/master/replication/event.go
// https://dev.mysql.com/doc/dev/mysql-server/latest/page_protocol_replication_binlog_event.html#sect_protocol_replication_binlog_event_header
// Issue: https://github.com/github/gh-ost/issues/1366
func (this *BinlogCoordinates) IsLogPosOverflowBeyond4Bytes(preCoordinate *BinlogCoordinates) bool {
if preCoordinate == nil {
return false
}
if preCoordinate.IsEmpty() {
return false
}

if this.LogFile != preCoordinate.LogFile {
return false
}

if preCoordinate.LogPos+this.EventSize >= 1<<32 {
// Unexpected rows event, the previous binlog log_pos + current binlog event_size is overflow 4 bytes
return true
}
return false
}
44 changes: 44 additions & 0 deletions go/mysql/binlog_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
package mysql

import (
"math"
"testing"

"github.com/openark/golib/log"
Expand Down Expand Up @@ -52,3 +53,46 @@ func TestBinlogCoordinatesAsKey(t *testing.T) {

test.S(t).ExpectEquals(len(m), 3)
}

func TestIsLogPosOverflowBeyond4Bytes(t *testing.T) {
{
var preCoordinates *BinlogCoordinates
curCoordinates := &BinlogCoordinates{LogFile: "mysql-bin.00017", LogPos: 10321, EventSize: 1100}
test.S(t).ExpectFalse(curCoordinates.IsLogPosOverflowBeyond4Bytes(preCoordinates))
}
{
preCoordinates := &BinlogCoordinates{LogFile: "mysql-bin.00017", LogPos: 1100, EventSize: 1100}
curCoordinates := &BinlogCoordinates{LogFile: "mysql-bin.00017", LogPos: int64(uint32(preCoordinates.LogPos + 1100)), EventSize: 1100}
test.S(t).ExpectFalse(curCoordinates.IsLogPosOverflowBeyond4Bytes(preCoordinates))
}
{
preCoordinates := &BinlogCoordinates{LogFile: "mysql-bin.00016", LogPos: 1100, EventSize: 1100}
curCoordinates := &BinlogCoordinates{LogFile: "mysql-bin.00017", LogPos: int64(uint32(preCoordinates.LogPos + 1100)), EventSize: 1100}
test.S(t).ExpectFalse(curCoordinates.IsLogPosOverflowBeyond4Bytes(preCoordinates))
}
{
preCoordinates := &BinlogCoordinates{LogFile: "mysql-bin.00017", LogPos: math.MaxUint32 - 1001, EventSize: 1000}
curCoordinates := &BinlogCoordinates{LogFile: "mysql-bin.00017", LogPos: int64(uint32(preCoordinates.LogPos + 1000)), EventSize: 1000}
test.S(t).ExpectFalse(curCoordinates.IsLogPosOverflowBeyond4Bytes(preCoordinates))
}
{
preCoordinates := &BinlogCoordinates{LogFile: "mysql-bin.00017", LogPos: math.MaxUint32 - 1000, EventSize: 1000}
curCoordinates := &BinlogCoordinates{LogFile: "mysql-bin.00017", LogPos: int64(uint32(preCoordinates.LogPos + 1000)), EventSize: 1000}
test.S(t).ExpectFalse(curCoordinates.IsLogPosOverflowBeyond4Bytes(preCoordinates))
}
{
preCoordinates := &BinlogCoordinates{LogFile: "mysql-bin.00017", LogPos: math.MaxUint32 - 999, EventSize: 1000}
curCoordinates := &BinlogCoordinates{LogFile: "mysql-bin.00017", LogPos: int64(uint32(preCoordinates.LogPos + 1000)), EventSize: 1000}
test.S(t).ExpectTrue(curCoordinates.IsLogPosOverflowBeyond4Bytes(preCoordinates))
}
{
preCoordinates := &BinlogCoordinates{LogFile: "mysql-bin.00017", LogPos: int64(uint32(math.MaxUint32 - 500)), EventSize: 1000}
curCoordinates := &BinlogCoordinates{LogFile: "mysql-bin.00017", LogPos: int64(uint32(preCoordinates.LogPos + 1000)), EventSize: 1000}
test.S(t).ExpectTrue(curCoordinates.IsLogPosOverflowBeyond4Bytes(preCoordinates))
}
{
preCoordinates := &BinlogCoordinates{LogFile: "mysql-bin.00017", LogPos: math.MaxUint32, EventSize: 1000}
curCoordinates := &BinlogCoordinates{LogFile: "mysql-bin.00017", LogPos: int64(uint32(preCoordinates.LogPos + 1000)), EventSize: 1000}
test.S(t).ExpectTrue(curCoordinates.IsLogPosOverflowBeyond4Bytes(preCoordinates))
}
}