leosanqing opened a new pull request, #4468: URL: https://github.com/apache/flink-cdc/pull/4468
• Fixes [FLINK-34807](https://issues.apache.org/jira/browse/FLINK-34807) ## Motivation We use TDSQL(based on MariaDB) in our production environment. When the MariaDB cluster changes, for example during scaling, migration, or failover, or when a Flink job restarts, the connector may be routed to a different MariaDB server. Although the new server belongs to the same GTID domain, its server ID may differ and its replicated GTID position may temporarily lag behind the position stored in the Flink checkpoint. The current MySQL CDC connector cannot recover automatically in this situation because its GTID handling is based on MySQL's UUID:interval format and does not understand MariaDB's `domain-server-sequence` format. An earlier PR, [#2494](https://github.com/apache/flink-cdc/pull/2494), attempted to add MariaDB support. This PR adopts one important finding from that work: the MariaDB replication capability must be set to `4` so that `MARIADB_GTID` events are emitted and the consumed GTID position can advance. However, the earlier implementation does not fully address the production recovery scenarios described above. This PR introduces a more complete and extensible solution. ## Description This PR adds dialect-aware MariaDB GTID support to the MySQL CDC connector. MariaDB and MySQL use incompatible GTID models: - MySQL: `uuid:interval` - MariaDB: `domain-server-sequence` A new `GtidStrategy` abstraction separates the parsing, comparison, containment, and recovery behavior of the two dialects. The implementation: - Adds config option `scan.dialect` with `mysql`, `mariadb`, and `auto` modes. - Add MariaDbGtidStrategy**(MOST IMPORTTANT)** - Reads the local MariaDB binlog position from `@@gtid_binlog_pos`. - Compares MariaDB GTIDs by domain and sequence instead of server ID. - Corrects a restored GTID against the GTID position available on the target server. - Enables MariaDB replication capability `4` so that GTID events advance the checkpointed offset. ## Comparison with the Earlier MariaDB PR We use the earlier implementation in [#2494](https://github.com/apache/flink-cdc/pull/2494) at first, but In pre-prod env we find several limitations could cause recovery problems. ### 1. Server ID is treated as part of the recovery identity(Main) The earlier implementation uses `MariadbGtidSet.isContainedWithin()` directly. That comparison includes the MariaDB server ID. In a MariaDB cluster, the server ID may change after failover, migration, scaling, or routing to another replica, while the GTID domain continues to represent the same logical stream. In production environments, especially with cloud databases that have a proxy layer, this scenario is very common. This PR compares MariaDB GTIDs by domain and sequence. The server ID is preserved in the stored GTID text, but it is not used to determine whether one position contains another. ### 2. The restored GTID may be ahead of the selected replica(sometimes) The earlier implementation passes the checkpointed GTID directly to the binlog client: After a routing change, the selected replica may temporarily lag behind the checkpointed position. ### 3. MariaDB handling is embedded in the MySQL code path ### 4. Test coverage does not cover production recovery scenarios ## Tests Added or updated tests for: - MariaDB domain-based comparison and containment. - Restored GTID correction against the target server. - Binlog offset serialization and recovery. - Verified with JDK 11, Docker, and MariaDB 11.4: ``` mvn test \ -pl flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc \ -Dtest=MariaDbDialectITCase,MariaDbDialectSavepointITCase \ -Dspotless.check.skip=true \ -Drat.skip=true ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
