MariaDB 11.4 | CDC not supported

[marceltn]

Hello, we faced issues with MariaDB v10+ with CDC / incremental sync fails with binlog privileges warning + runtime error.
We solved that 2 issues in our case but just sharing here for future MariaDB support.

Context

MariaDB 11.4 as a source
Tapdata v3.27.0
PDK connector v1.4.4

Steps:

1. create 2 new MariaDB connector for source and target connection (different DBs)
2. setup the permissions to read binlog
   GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'root'@'%';
3. test the source connector to face warning message
4. test an incremental sync to face exception message

---

First issue: test source connector

There is a warning for read binlog because we use the same MySQL validation that changed on MariaDB latest versions.

tapdata-connectors/connectors-common/mysql-core/src/main/java/io/tapdata/connector/mysql/MysqlConnectionTest.java

Line 355 in 3346cef

REPLICATION_CLIENT("REPLICATION CLIENT|SUPER", false),

-        REPLICATION_CLIENT("REPLICATION CLIENT|SUPER", false),
+        REPLICATION_CLIENT("BINLOG MONITOR", false),

and warning message just to help on search:
User does not have privileges [REPLICATION CLIENT|SUPER], will not be able to use the incremental sync feature.

---

Second issue: incremental sync failure

After try the incremental sync from MariaDB, it's not working because the command to access the binlogs changed on MariaDB latest versions.
MariaDB still support SHOW MASTER STATUS but not for SHOW BINARY LOG STATUS.
This are more tricky fix as there are no major version grater than 8 from MySQL and only for MariaDB. It's fragile but it works for our current needs.

tapdata-connectors/connectors-common/mysql-core/src/main/java/io/tapdata/connector/mysql/MysqlJdbcContextV2.java

Line 145 in 3346cef

if (majorVersion == 8 && minorVersion >= 4 || majorVersion > 8) {

    public MysqlBinlogPosition readBinlogPosition() throws Throwable {
        AtomicReference<MysqlBinlogPosition> mysqlBinlogPositionAtomicReference = new AtomicReference<>();
        String binLogStatusSql = "SHOW MASTER STATUS";
        try (
                Connection connection = getConnection()
        ) {
            DatabaseMetaData databaseMetaData = connection.getMetaData();
            String version = databaseMetaData.getDatabaseMajorVersion() + "." + databaseMetaData.getDatabaseMinorVersion();
            String[] versionNums = version.split("\\.");
            if (versionNums.length >= 2) {
                int majorVersion = Integer.parseInt(versionNums[0]);
                int minorVersion = Integer.parseInt(versionNums[1]);
-                if (majorVersion == 8 && minorVersion >= 4 || majorVersion > 8) {
+                if (majorVersion == 8 && minorVersion >= 4 || (majorVersion > 8 && majorVersion < 10)) {
                    binLogStatusSql = "SHOW BINARY LOG STATUS";
                }
            }
        }

And also here, with the same solution above:

tapdata-connectors/connectors-common/debezium-bucket/debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/MySqlConnection.java

Line 240 in 3346cef

if (majorVersion == 8 && minorVersion >= 4 || majorVersion > 8) {

And exception message just to help on search:

io.tapdata.exception.NodeException: Call timestamp to stream offset function failed, will stop task, type: mariadb-io.tapdata-1.0-SNAPSHOT-public, errors: SQLSyntaxErrorException  You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near 'LOG STATUS' at line 1