Skip to content

handle fake rotate event as real if binlog filename changed #753

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.

Sign up for GitHub

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

Merged
merged 3 commits into from
Dec 14, 2022
Merged

handle fake rotate event as real if binlog filename changed #753

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ef0bcca
handle fake roate event as real if binlog filename changed
Dec 12, 2022
1902f25
no need to preserve fakeRoteEventLogName if not changed
Dec 12, 2022
bf4f639
fix typo
Dec 12, 2022
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
21 changes: 8 additions & 13 deletions canal/sync.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,10 +43,6 @@ func (c *Canal) runSyncBinlog() error {
savePos := false
force := false

// The name of the binlog file received in the fake rotate event.
// It must be preserved until the new position is saved.
fakeRotateLogName := ""

for {
ev, err := s.GetEvent(c.ctx)
if err != nil {
Expand All @@ -63,11 +59,16 @@ func (c *Canal) runSyncBinlog() error {
if ev.Header.LogPos == 0 {
switch e := ev.Event.(type) {
case *replication.RotateEvent:
fakeRotateLogName = string(e.NextLogName)
fakeRotateLogName := string(e.NextLogName)
c.cfg.Logger.Infof("received fake rotate event, next log name is %s", e.NextLogName)
if fakeRotateLogName != c.master.Position().Name {
c.cfg.Logger.Info("log name changed, the fake rotate event will be handled as a real rotate event")
} else {
continue
}
default:
continue
}

continue
}

savePos = false
Expand All @@ -79,11 +80,6 @@ func (c *Canal) runSyncBinlog() error {
// next binlog pos
pos.Pos = ev.Header.LogPos

// new file name received in the fake rotate event
if fakeRotateLogName != "" {
pos.Name = fakeRotateLogName
}

// We only save position with RotateEvent and XIDEvent.
// For RowsEvent, we can't save the position until meeting XIDEvent
// which tells the whole transaction is over.
Expand Down Expand Up @@ -174,7 +170,6 @@ func (c *Canal) runSyncBinlog() error {
if savePos {
c.master.Update(pos)
c.master.UpdateTimestamp(ev.Header.Timestamp)
fakeRotateLogName = ""

if err := c.eventHandler.OnPosSynced(ev.Header, pos, c.master.GTIDSet(), force); err != nil {
return errors.Trace(err)
Expand Down