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

DBZ-7500 Fix MySQL 8 event timestamp resolution logic error where fallback to seconds occurs erroneously for non-GTID events

Browse Source
This commit is contained in:
Lourens Naude 2024-02-16 23:35:05 +00:00 committed by Chris Cranford
parent 2ac78005eb
commit 70aa83b371
1 changed files with 11 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java
View File

@ -219,14 +219,14 @@ protected void onEvent(MySqlOffsetContext offsetContext, Event event) {
return; return;
} }
eventTimestamp = getEventTimestamp(event, eventTs); setEventTimestamp(event, eventTs);
ts = clock.currentTimeInMillis() - eventTimestamp.toEpochMilli(); ts = clock.currentTimeInMillis() - eventTimestamp.toEpochMilli();
LOGGER.trace("Current milliseconds behind source: {} ms", ts); LOGGER.trace("Current milliseconds behind source: {} ms", ts);
metrics.setMilliSecondsBehindSource(ts); metrics.setMilliSecondsBehindSource(ts);
} }
private Instant getEventTimestamp(Event event, long eventTs) { private void setEventTimestamp(Event event, long eventTs) {
// Prefer higher resolution replication timestamps from MySQL 8 GTID events, if possible // Prefer higher resolution replication timestamps from MySQL 8 GTID events, if possible
if (isGtidModeEnabled) { if (isGtidModeEnabled) {
if (event.getHeader().getEventType() == EventType.GTID) { if (event.getHeader().getEventType() == EventType.GTID) {
@ -234,13 +234,18 @@ private Instant getEventTimestamp(Event event, long eventTs) {
final long gtidEventTs = gtidEvent.getOriginalCommitTimestamp(); final long gtidEventTs = gtidEvent.getOriginalCommitTimestamp();
if (gtidEventTs != 0) { if (gtidEventTs != 0) {
// >= MySQL 8.0.1, prefer the higher resolution replication timestamp // >= MySQL 8.0.1, prefer the higher resolution replication timestamp
return Instant.EPOCH.plus(gtidEventTs, ChronoUnit.MICROS); eventTimestamp = Instant.EPOCH.plus(gtidEventTs, ChronoUnit.MICROS);
}
else {
// Fallback to second resolution event timestamps
eventTimestamp = Instant.ofEpochMilli(eventTs);
} }
} }
} }
else {
// Fallback to second resolution event timestamps // Fallback to second resolution event timestamps
return Instant.ofEpochMilli(eventTs); eventTimestamp = Instant.ofEpochMilli(eventTs);
}
} }
protected void ignoreEvent(MySqlOffsetContext offsetContext, Event event) { protected void ignoreEvent(MySqlOffsetContext offsetContext, Event event) {