debezium/dbz#1200 Support for GTID tags

.github/workflows/mysql-84.yml

@@ -0,0 +1,44 @@
+#
+# Copyright Debezium Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# This workflow will build a Java project with Maven
+# For more information see: https://help.github.com/actions/language-and-framework-guides/building-and-testing-java-with-maven
+
+name: Build and run tests against MySQL 8.4
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+    branches:
+      - main
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v6
+    - uses: ./.github/actions/setup-java
+    - name: Cache Maven packages
+      uses: actions/cache@v4
+      with:
+        path: ~/.m2
+        key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }}
+        restore-keys: ${{ runner.os }}-m2
+    - name: Build with Maven
+      run: ./mvnw -B install -Pmysql-8.4 -Dgpg.skip=true --file pom.xml

pom.xml

@@ -241,6 +241,9 @@
                     <includes>
                         <include>**/*IntegrationTest.java</include>
                     </includes>
+                    <systemPropertyVariables>
+                        <mysql.image>${mysql.image}</mysql.image>
+                    </systemPropertyVariables>
                 </configuration>
                 <executions>
                     <execution>
@@ -312,7 +315,7 @@
         <profile>
             <id>mysql-8.0</id>
             <properties>
-                <mysql.image>container-registry.oracle.com/mysql/community-server:8.0</mysql.image>
+                <mysql.image>mirror.gcr.io/mysql:8.0</mysql.image>
             </properties>
         </profile>
         <profile>
@@ -321,6 +324,12 @@
                 <mysql.image>mirror.gcr.io/library/mariadb:10.6</mysql.image>
             </properties>
         </profile>
+        <profile>
+            <id>mysql-8.4</id>
+            <properties>
+                <mysql.image>mirror.gcr.io/mysql:8.4</mysql.image>
+            </properties>
+        </profile>
     </profiles>
 
 </project>

src/main/java/com/github/shyiko/mysql/binlog/GtidSet.java

@@ -45,20 +45,36 @@ public static GtidSet parse(String gtidStr) {
     }
     /**
      * @param gtidSet gtid set comprised of closed intervals (like MySQL's executed_gtid_set).
+     * Supports both legacy format (uuid:intervals) and MySQL 8.3+ tagged format (tag:uuid:intervals).
      */
     public GtidSet(String gtidSet) {
         String[] uuidSets = (gtidSet == null || gtidSet.isEmpty()) ? new String[0] :
             gtidSet.replace("\n", "").split(",");
         for (String uuidSet : uuidSets) {
-            int uuidSeparatorIndex = uuidSet.indexOf(":");
-            UUID sourceId = UUID.fromString(uuidSet.substring(0, uuidSeparatorIndex));
-            List<Interval> intervals = new ArrayList<Interval>();
-            String[] rawIntervals = uuidSet.substring(uuidSeparatorIndex + 1).split(":");
-            for (String interval : rawIntervals) {
-                String[] is = interval.split("-");
+            final String[] parts = uuidSet.split(":");
+
+            // Determine if this is a tagged GTID (tag:uuid:intervals) or legacy (uuid:intervals)
+            int uuidIndex = 0;
+            int intervalsStartIndex = 1;
+
+            // Check if first part is a tag (not a valid UUID format)
+            // UUID format: 8-4-4-4-12 hex digits with dashes
+            if (parts.length >= 3 && !isValidUuidFormat(parts[0])) {
+                // Tagged format: tag:uuid:intervals...
+                uuidIndex = 1;
+                intervalsStartIndex = 2;
+            }
+
+            final UUID sourceId = UUID.fromString(parts[uuidIndex]);
+            final List<Interval> intervals = new ArrayList<Interval>();
+
+            // Parse intervals starting from the correct index
+            for (int i = intervalsStartIndex; i < parts.length; i++) {
+                final String interval = parts[i];
+                final String[] is = interval.split("-");
                 long[] split = new long[is.length];
-                for (int i = 0, e = is.length; i < e; i++) {
-                    split[i] = Long.parseLong(is[i]);
+                for (int j = 0, e = is.length; j < e; j++) {
+                    split[j] = Long.parseLong(is[j]);
                 }
                 if (split.length == 1) {
                     split = new long[] {split[0], split[0]};
@@ -69,6 +85,17 @@ public GtidSet(String gtidSet) {
         }
     }
 
+    /**
+     * Checks if a string matches the UUID format (8-4-4-4-12 hex digits).
+     *
+     * @param str the string to check
+     * @return true if the string is a valid UUID format, false otherwise
+     */
+    private static boolean isValidUuidFormat(final String str) {
+        // UUID format: 8-4-4-4-12 hex digits with dashes
+        return str.matches("[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}");
+    }
+
     /**
      * Get an immutable collection of the {@link UUIDSet range of GTIDs for a single server}.
      * @return the {@link UUIDSet GTID ranges for each server}; never null

src/main/java/com/github/shyiko/mysql/binlog/event/MySqlGtid.java

@@ -2,25 +2,90 @@
 
 import java.util.UUID;
 
+/**
+ * Represents a MySQL GTID (Global Transaction Identifier).
+ * Supports both legacy format (uuid:transaction_id) and MySQL 8.3+ tagged format (tag:uuid:transaction_id).
+ */
 public class MySqlGtid {
+    private final String tag;
     private final UUID serverId;
     private final long transactionId;
 
-    public MySqlGtid(UUID serverId, long transactionId) {
+    /**
+     * Creates a MySqlGtid without a tag (legacy format).
+     *
+     * @param serverId the server UUID
+     * @param transactionId the transaction ID
+     */
+    public MySqlGtid(final UUID serverId, final long transactionId) {
+        this(null, serverId, transactionId);
+    }
+
+    /**
+     * Creates a MySqlGtid with an optional tag (MySQL 8.3+ format).
+     *
+     * @param tag the optional tag (null or empty for legacy format)
+     * @param serverId the server UUID
+     * @param transactionId the transaction ID
+     */
+    public MySqlGtid(final String tag, final UUID serverId, final long transactionId) {
+        this.tag = tag;
         this.serverId = serverId;
         this.transactionId = transactionId;
     }
 
-    public static MySqlGtid fromString(String gtid) {
-        String[] split = gtid.split(":");
-        String sourceId = split[0];
-        long transactionId = Long.parseLong(split[1]);
-        return new MySqlGtid(UUID.fromString(sourceId), transactionId);
+    /**
+     * Parses a GTID string in either legacy or tagged format.
+     * <p>
+     * Supported formats:
+     * <ul>
+     *   <li>Legacy: uuid:transaction_id</li>
+     *   <li>Tagged (MySQL 8.3+): tag:uuid:transaction_id</li>
+     * </ul>
+     *
+     * @param gtid the GTID string to parse
+     * @return the parsed MySqlGtid
+     * @throws IllegalArgumentException if the format is invalid
+     */
+    public static MySqlGtid fromString(final String gtid) {
+        final String[] split = gtid.split(":");
+
+        if (split.length == 2) {
+            // Legacy format: uuid:transaction_id
+            final String sourceId = split[0];
+            final long transactionId = Long.parseLong(split[1]);
+            return new MySqlGtid(UUID.fromString(sourceId), transactionId);
+        }
+        else if (split.length == 3) {
+            // Tagged format: tag:uuid:transaction_id
+            final String tag = split[0];
+            final String sourceId = split[1];
+            final long transactionId = Long.parseLong(split[2]);
+            return new MySqlGtid(tag, UUID.fromString(sourceId), transactionId);
+        }
+        else {
+            throw new IllegalArgumentException(
+                "Invalid GTID format: " + gtid +
+                ". Expected format: 'uuid:transaction_id' or 'tag:uuid:transaction_id'"
+            );
+        }
     }
 
     @Override
     public String toString() {
-        return serverId.toString()+":"+transactionId;
+        if (tag != null && !tag.isEmpty()) {
+            return tag + ":" + serverId.toString() + ":" + transactionId;
+        }
+        return serverId.toString() + ":" + transactionId;
+    }
+
+    /**
+     * Gets the optional tag (MySQL 8.3+).
+     *
+     * @return the tag, or null if not present
+     */
+    public String getTag() {
+        return tag;
     }
 
     public UUID getServerId() {

src/test/java/com/github/shyiko/mysql/binlog/BinaryLogClientGTIDIntegrationTest.java

@@ -197,7 +197,12 @@ public void execute(Statement statement) throws SQLException {
 
     private String getExecutedGtidSet(MySQLConnection master) throws SQLException {
         final String[] initialGTIDSet = new String[1];
-        master.query("show master status", new Callback<ResultSet>() {
+        // MySQL 8.4+ renamed SHOW MASTER STATUS to SHOW BINARY LOG STATUS
+        String version = System.getProperty("mysql.image", "mysql:8.0");
+        boolean isMySQL84Plus = version.contains("8.4") || version.contains("8.5") || version.contains("9.");
+        String statusQuery = isMySQL84Plus ? "SHOW BINARY LOG STATUS" : "SHOW MASTER STATUS";
+
+        master.query(statusQuery, new Callback<ResultSet>() {
             @Override
             public void execute(ResultSet rs) throws SQLException {
                 rs.next();