sanqingleo created FLINK-40314:
----------------------------------
Summary: Introduce a MariaDB CDC connector
Key: FLINK-40314
URL: https://issues.apache.org/jira/browse/FLINK-40314
Project: Flink
Issue Type: New Feature
Components: Flink CDC
Reporter: sanqingleo
h2. Motivation
The existing MySQL CDC connector can connect to some MariaDB deployments
through the MySQL-compatible protocol. However, it does not correctly support
MariaDB GTID parsing or recovery across physical MariaDB servers.
MySQL and MariaDB use incompatible GTID formats:
* MySQL: UUID-based `uuid:interval`
* MariaDB: `domain-server-sequence`
When a MariaDB cluster is scaled, migrated, or failed over, or when a Flink job
restarts, the connector may be routed to another server in the same GTID
domain. The new server can have a different server ID, and its replicated GTID
position may temporarily lag behind the position stored in the Flink checkpoint.
The current MySQL CDC GTID implementation cannot correctly handle this
situation because it interprets GTIDs using the MySQL format and semantics.
PR apache/flink-cdc#4468 has verified MariaDB GTID support against MariaDB 11.4
by adding MariaDB-specific behavior directly to `mysql-cdc`. However, modifying
the MySQL connector introduces regression risks to one of the most critical
Flink CDC source connectors.
Therefore, this issue proposes introducing a standalone
`flink-connector-mariadb-cdc` connector, following an implementation approach
similar to `oceanbase-cdc`.
h2. Why a standalone connector
* Stability isolation: MariaDB-specific changes should not affect the MySQL
CDC execution path.
* Independent evolution: MySQL and MariaDB have diverged significantly,
especially after MySQL 8.4, and are expected to evolve in different directions.
* Upstream precedent: Debezium introduced a standalone
`debezium-connector-mariadb` after extracting shared binlog functionality.
* Simpler configuration: A MariaDB-only connector does not require
`scan.dialect` or dialect-selection branches.
h2. Proposed Design
h3. Module structure
Add the following modules:
* `flink-connector-mariadb-cdc`
* `flink-sql-connector-mariadb-cdc`
The new connector should be registered under:
`flink-cdc-connect/flink-cdc-source-connectors`
The initial module will depend on the unmodified upstream
`flink-connector-mysql-cdc` and reuse its snapshot and source implementation.
{code:java}
flink-connector-mariadb-cdc
├── flink-connector-mysql-cdc
├── mysql:mysql-connector-java
└── flink-cdc-common
{code}
MariaDB behavior will be injected entirely from the new connector without
modifying the `mysql-cdc` source code.
h3. Public API
* SQL factory identifier: `mariadb-cdc`
* Flink package: `org.apache.flink.cdc.connectors.mariadb`
* DataStream API: reuse `MySqlSource.builder()` in the first version
* Connector options: reuse the existing MySQL CDC options, including startup
mode, server ID, and incremental snapshot options
* Do not introduce `scan.dialect`
* For `scan.startup.mode=specific-offset`, the GTID must use the MariaDB
`domain-server-sequence` format
A dedicated `MariaDbSource` DataStream API can be introduced in a follow-up
issue.
h3. MariaDB GTID behavior
The connector should:
* Enable MariaDB replication capability `4` so that `MARIADB_GTID` events are
emitted and the consumed GTID position advances.
* Read the local MariaDB GTID position from `@@gtid_binlog_pos`.
* Obtain the binlog file and position using `SHOW MASTER STATUS`.
* Parse MariaDB GTIDs using the `domain-server-sequence` format.
* Compare GTID positions by domain and sequence.
* Ignore the MariaDB server ID when evaluating GTID containment.
* Validate restored GTIDs and reject invalid MariaDB GTID formats.
* Preserve GTID information during offset serialization and checkpoint
recovery.
h3. Classes to override
The initial implementation should override the following classes from the MySQL
CDC dependency:
* `io.debezium.connector.mysql.MySqlStreamingChangeEventSource`
* `org.apache.flink.cdc.connectors.mysql.debezium.DebeziumUtils`
*
`org.apache.flink.cdc.connectors.mysql.debezium.task.context.StatefulTaskContext`
* `org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset`
The connector should introduce the following MariaDB-specific classes:
* `mariadb.debezium.MariaDBBinaryLogClient`
* `mariadb.debezium.MariaDbConnections`
* `mariadb.source.offset.MariaDbGtidStrategy`
* `mariadb.table.MariaDbTableSourceFactory`
h2. Recovery Semantics
Let:
* `C` be the GTID position stored in the Flink checkpoint.
* `H` be the executed GTID position reported by the target server through
`@@gtid_binlog_pos`.
* `L` be the earliest position still available in the target server's binlog.
The initial implementation will only perform GTID containment validation. It
will not modify, downgrade, or automatically correct the checkpointed position.
If `C > H`, meaning that the selected replica temporarily lags behind the
checkpointed position, task initialization should fail while preserving the
original checkpoint position.
The connector will not implement internal waiting, repeated position queries,
timeout handling, or replica reselection in the first version. Recovery will
rely on Flink's task restart strategy. In production, the replica normally
catches up within a short period, allowing a subsequent restart to recover
successfully.
When reading from a replica, `log_slave_updates=ON`, or the equivalent cloud
database setting, is required so that replicated transactions are written to
the local binlog.
h2. Known Classloader Limitation
The first implementation will override classes using the same fully qualified
class names as classes contained in `mysql-cdc`.
Therefore, `flink-sql-connector-mysql-cdc` and
`flink-sql-connector-mariadb-cdc` must not be placed in the same shared
classloader, such as `FLINK_HOME/lib`.
If both connectors are loaded by the same classloader, JVM classpath ordering
determines which implementation is used:
* If the MySQL implementation is loaded first, `mariadb-cdc` may silently lose
MariaDB replication capability `4`, preventing the consumed GTID from advancing.
* If the MariaDB implementation is loaded first, `mysql-cdc` may execute
MariaDB-specific GTID queries and comparison logic.
The recommended deployment is either:
* Use separate Flink deployments for MySQL CDC and MariaDB CDC; or
* Keep both connector JARs out of `FLINK_HOME/lib` and expose only the
required connector through each job's user classloader.
Supporting `mysql-cdc` and `mariadb-cdc` in the same classloader is not part of
the initial scope.
h2. Testing Plan
MariaDB 11.4 will be used as the initial integration-test baseline.
Tests should cover:
* Initial snapshot reading
* Snapshot-to-binlog transition
* Savepoint and checkpoint recovery
* MariaDB GTID domain and sequence comparison
* GTID containment while ignoring server ID
* Binlog offset serialization and recovery
* Invalid GTID rejection
* `MARIADB_GTID` event handling
* Cross-server-ID recovery
Cross-server-ID recovery can be tested deterministically without a three-node
cluster:
# Create a savepoint.
# Change the MariaDB server ID using `SET GLOBAL server_id`.
# Commit a transaction to an uncaptured table before restoring the job so that
`@@gtid_binlog_pos` contains the new server ID.
# Restore the job and verify that recovery succeeds across the server-ID
boundary.
The documentation will initially declare MariaDB 10.x and 11.x support.
Integration coverage for MariaDB 10.6 and 10.11 LTS can be added in follow-up
work.
h2. Non-Goals
* Do not modify the `mysql-cdc` source code.
* Do not package this connector in `flink-cdc-dist`.
* Do not switch to MariaDB Connector/J in the first version; continue using
MySQL Connector/J because the reused MySQL CDC path currently uses
`jdbc:mysql://` and `com.mysql.cj.jdbc.Driver`.
* Do not support `mysql-cdc` and `mariadb-cdc` in the same classloader.
h2. Implementation Plan
# Create the `flink-connector-mariadb-cdc` module, SQL factory, and SPI
registration.
# Add MariaDB binlog and GTID support.
# Add GTID comparison, containment, serialization, and recovery tests.
# Create the shaded `flink-sql-connector-mariadb-cdc` artifact.
# Add MariaDB CDC documentation.
# Add MariaDB source E2E tests and CI coverage.
h2. Related Work
* FLINK-34807
* apache/flink-cdc#4468
* apache/flink-cdc#2494
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
