Support MySQL 8 high resolution replication timestamps from GTID events

COPYRIGHT.txt

@@ -299,6 +299,7 @@ Liu Hanlin
 Liu Lang Wa
 Liz Chatman
 Lokesh Sanapalli
+Lourens Naudé
 Luca Scannapieco
 Luis Garcés-Erice
 Lukas Krejci

debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java

@@ -9,6 +9,7 @@
 import java.sql.SQLException;
 import java.time.Duration;
 import java.time.Instant;
+import java.time.temporal.ChronoUnit;
 import java.util.EnumMap;
 import java.util.List;
 import java.util.Map;
@@ -87,6 +88,7 @@ public class MySqlStreamingChangeEventSource implements StreamingChangeEventSour
     private long initialEventsToSkip = 0L;
     private boolean skipEvent = false;
     private boolean ignoreDmlEventByGtidSource = false;
+    private final boolean isGtidModeEnabled;
     private final Predicate<String> gtidDmlSourceFilter;
     private final AtomicLong totalRecordCounter = new AtomicLong();
     private volatile Map<String, ?> lastOffset = null;
@@ -190,6 +192,7 @@ public MySqlStreamingChangeEventSource(MySqlConnectorConfig connectorConfig, Abs
         Configuration configuration = connectorConfig.getConfig();
         boolean filterDmlEventsByGtidSource = configuration.getBoolean(MySqlConnectorConfig.GTID_SOURCE_FILTER_DML_EVENTS);
         gtidDmlSourceFilter = filterDmlEventsByGtidSource ? connectorConfig.gtidSourceFilter() : null;
+        isGtidModeEnabled = connection.isGtidModeEnabled();
     }
 
     protected void onEvent(MySqlOffsetContext offsetContext, Event event) {
@@ -212,11 +215,30 @@ protected void onEvent(MySqlOffsetContext offsetContext, Event event) {
             return;
         }
 
-        ts = clock.currentTimeInMillis() - eventTs;
+        eventTimestamp = getEventTimestamp(event, eventTs);
+
+        ts = clock.currentTimeInMillis() - eventTimestamp.toEpochMilli();
         LOGGER.trace("Current milliseconds behind source: {} ms", ts);
         metrics.setMilliSecondsBehindSource(ts);
     }
 
+    private Instant getEventTimestamp(Event event, long eventTs) {
+        // Prefer higher resolution replication timestamps from MySQL 8 GTID events, if possible
+        if (isGtidModeEnabled) {
+            if (event.getHeader().getEventType() == EventType.GTID) {
+                GtidEventData gtidEvent = unwrapData(event);
+                final long gtidEventTs = gtidEvent.getOriginalCommitTimestamp();
+                if (gtidEventTs != 0) {
+                    // >= MySQL 8.0.1, prefer the higher resolution replication timestamp
+                    return Instant.EPOCH.plus(gtidEventTs, ChronoUnit.MICROS);
+                }
+            }
+        }
+
+        // Fallback to second resolution event timestamps
+        return Instant.ofEpochMilli(eventTs);
+    }
+
     protected void ignoreEvent(MySqlOffsetContext offsetContext, Event event) {
         LOGGER.trace("Ignoring event due to missing handler: {}", event);
     }
@@ -228,10 +250,6 @@ protected void handleEvent(MySqlPartition partition, MySqlOffsetContext offsetCo
         }
 
         final EventHeader eventHeader = event.getHeader();
-        // Update the source offset info. Note that the client returns the value in *milliseconds*, even though the binlog
-        // contains only *seconds* precision ...
-        // HEARTBEAT events have no timestamp; only set the timestamp if the event is not a HEARTBEAT
-        eventTimestamp = !eventHeader.getEventType().equals(EventType.HEARTBEAT) ? Instant.ofEpochMilli(eventHeader.getTimestamp()) : null;
         offsetContext.setBinlogServerId(eventHeader.getServerId());
 
         final EventType eventType = eventHeader.getEventType();
@@ -869,7 +887,6 @@ public void execute(ChangeEventSourceContext context, MySqlPartition partition,
             client.registerEventListener((event) -> logEvent(effectiveOffsetContext, event));
         }
 
-        final boolean isGtidModeEnabled = connection.isGtidModeEnabled();
         metrics.setIsGtidModeEnabled(isGtidModeEnabled);
 
         // Get the current GtidSet from MySQL so we can get a filtered/merged GtidSet based off of the last Debezium checkpoint.

debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/SourceInfo.java

@@ -195,18 +195,6 @@ public void setBinlogServerId(long serverId) {
         this.serverId = serverId;
     }
 
-    /**
-     * Set the number of <em>seconds</em> since Unix epoch (January 1, 1970) as found within the MySQL binary log file.
-     * Note that the value in the binlog events is in seconds, but the library we use returns the value in milliseconds
-     * (with only second precision and therefore all fractions of a second are zero). We capture this as seconds
-     * since that is the precision that MySQL uses.
-     *
-     * @param timestampInSeconds the timestamp in <em>seconds</em> found within the binary log file
-     */
-    public void setBinlogTimestampSeconds(long timestampInSeconds) {
-        this.sourceTime = Instant.ofEpochSecond(timestampInSeconds);
-    }
-
     public void setSourceTime(Instant timestamp) {
         sourceTime = timestamp;
     }
@@ -272,10 +260,6 @@ long getCurrentBinlogPosition() {
         return currentBinlogPosition;
     }
 
-    long getBinlogTimestampSeconds() {
-        return (sourceTime == null) ? 0 : sourceTime.getEpochSecond();
-    }
-
     int getCurrentRowNumber() {
         return currentRowNumber;
     }

debezium-connector-mysql/src/test/java/io/debezium/connector/mysql/SourceInfoTest.java

@@ -8,6 +8,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertTrue;
 
+import java.time.Instant;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
@@ -625,7 +626,7 @@ public void shouldNotSetNullGtidSet() {
     @Test
     public void shouldHaveTimestamp() {
         sourceWith(offset(100, 5, true));
-        source.setBinlogTimestampSeconds(1_024);
+        source.setSourceTime(Instant.ofEpochSecond(1_024, 0));
         source.databaseEvent("mysql");
         assertThat(source.struct().get("ts_ms")).isEqualTo(1_024_000L);
     }

jenkins-jobs/scripts/config/Aliases.txt

@@ -241,4 +241,5 @@ Lars M Johansson,Lars M. Johansson
 ahmedrachid,Ahmed Rachid Hazourli
 sherpa003,Jiri Kulhanek
 slknijnenburg,Sebastiaan Knijnenburg
-baabgai,baabgai
+baabgai,baabgai
+methodmissing,Lourens Naudé

pom.xml

@@ -124,7 +124,7 @@
         <!-- Database drivers, should align with databases -->
         <version.postgresql.driver>42.6.0</version.postgresql.driver>
         <version.mysql.driver>8.0.33</version.mysql.driver>
-        <version.mysql.binlog>0.28.3</version.mysql.binlog>
+        <version.mysql.binlog>0.29.0</version.mysql.binlog>
         <version.mongo.driver>4.11.0</version.mongo.driver>
         <version.sqlserver.driver>12.4.2.jre8</version.sqlserver.driver>
         <version.oracle.driver>21.6.0.0</version.oracle.driver>