alexey.rassokhin/tet123
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

DBZ-1033 Mysql binlog reader lost data if restart task when last binlog event is QUERY event.

Browse Source 
https://issues.jboss.org/browse/DBZ-1033
Reset restartEventsToSkip each time updating the restartBinlogPosition.
This commit is contained in:
LiuHanlin 2018-12-11 13:17:49 +08:00 committed by Gunnar Morling
parent fcfc019b6b
commit c58031582d
2 changed files with 5 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/SourceInfo.java
View File

@ -211,6 +211,7 @@ public void setBinlogStartPoint(String binlogFilename, long positionOfFirstEvent
this.restartBinlogPosition = positionOfFirstEvent;
this.currentRowNumber = 0;
this.restartRowsToSkip = 0;
this.restartEventsToSkip = 0;
}
/**
@ -224,6 +225,8 @@ public void setEventPosition(long positionOfCurrentEvent, long eventSizeInBytes)
this.currentEventLengthInBytes = eventSizeInBytes;
if (!inTransaction) {
this.restartBinlogPosition = positionOfCurrentEvent + eventSizeInBytes;
this.restartRowsToSkip = 0;
this.restartEventsToSkip = 0;
}
// Don't set anything else, since the row numbers are set in the offset(int,int) method called at least once
// for each processed event

4
debezium-connector-mysql/src/test/java/io/debezium/connector/mysql/MySqlConnectorIT.java
View File

@ -728,8 +728,8 @@ public void shouldConsumeAllEventsFromDatabaseUsingSnapshot() throws SQLExceptio
} else {
// the replica is not the same server as the master, so it will have a different binlog filename and position ...
}
// Event number is 2 ...
assertThat(persistedOffsetSource.eventsToSkipUponRestart()).isEqualTo(2);
// Last event is 'SHOW MASTER STATUS' which will reset the event number to 0 ...
assertThat(persistedOffsetSource.eventsToSkipUponRestart()).isEqualTo(0);
// GTID set should match the before-inserts GTID set ...
// assertThat(persistedOffsetSource.gtidSet()).isEqualTo(positionBeforeInserts.gtidSet());