Skip to content

[FLINK-34807][mysql] Add MariaDB GTID support #4468

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
leosanqing wants to merge 3 commits into
base: master
Choose a base branch
from
Open

[FLINK-34807][mysql] Add MariaDB GTID support #4468

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a57fb0f
[FLINK-34807][mysql] Add MariaDB GTID support
leosanqing Jul 10, 2026
7b484b5
[FLINK-34807][mysql] Add MariaDB GTID support
leosanqing Jul 11, 2026
3499f65
[FLINK-34807][mysql] Add MariaDB GTID support
leosanqing Jul 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions .../flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlConnection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -304,6 +304,26 @@ public String knownGtidSet() {
}
}

/**
* return the GTID set present in MariaDB's binary log "@@gtid_binlog_pos" Empty string if none.
*/
public String mariadbGtidExecuted() {
try {
return queryAndMap(
"SELECT @@gtid_binlog_pos",
rs -> {
if (rs.next() && rs.getMetaData().getColumnCount() > 0) {
String v = rs.getString(1);
return v == null ? "" : v;
}
return "";
});
} catch (SQLException e) {
throw new DebeziumException(
"Unexpect error while reading MariaDB @@gtid_binlog_pos: ", e);
}
}

/**
* Determine the difference between two sets.
*
Expand Down
164 changes: 128 additions & 36 deletions ...-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,15 +6,20 @@

package io.debezium.connector.mysql;

import org.apache.flink.cdc.connectors.mysql.source.offset.MariaDbGtidStrategy;
import org.apache.flink.cdc.connectors.mysql.source.offset.MysqlGtidStrategy;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.BinaryLogClient.LifecycleListener;
import com.github.shyiko.mysql.binlog.MariadbGtidSet;
import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData;
import com.github.shyiko.mysql.binlog.event.Event;
import com.github.shyiko.mysql.binlog.event.EventData;
import com.github.shyiko.mysql.binlog.event.EventHeader;
import com.github.shyiko.mysql.binlog.event.EventHeaderV4;
import com.github.shyiko.mysql.binlog.event.EventType;
import com.github.shyiko.mysql.binlog.event.GtidEventData;
import com.github.shyiko.mysql.binlog.event.MariadbGtidEventData;
import com.github.shyiko.mysql.binlog.event.QueryEventData;
import com.github.shyiko.mysql.binlog.event.RotateEventData;
import com.github.shyiko.mysql.binlog.event.RowsQueryEventData;
Expand Down Expand Up @@ -48,6 +53,7 @@
import io.debezium.util.Metronome;
import io.debezium.util.Strings;
import io.debezium.util.Threads;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.event.Level;
Expand Down Expand Up @@ -129,6 +135,7 @@ public class MySqlStreamingChangeEventSource
private final MySqlTaskContext taskContext;
private final MySqlConnectorConfig connectorConfig;
private final MySqlConnection connection;
private final String dialect;
private final EventDispatcher<MySqlPartition, TableId> eventDispatcher;
private final ErrorHandler errorHandler;

Expand Down Expand Up @@ -204,10 +211,31 @@ public MySqlStreamingChangeEventSource(
Clock clock,
MySqlTaskContext taskContext,
MySqlStreamingChangeEventSourceMetrics metrics) {
this(
connectorConfig,
connection,
dispatcher,
errorHandler,
clock,
taskContext,
metrics,
MysqlGtidStrategy.DIALECT);
}

public MySqlStreamingChangeEventSource(
MySqlConnectorConfig connectorConfig,
MySqlConnection connection,
EventDispatcher<MySqlPartition, TableId> dispatcher,
ErrorHandler errorHandler,
Clock clock,
MySqlTaskContext taskContext,
MySqlStreamingChangeEventSourceMetrics metrics,
String dialect) {

this.taskContext = taskContext;
this.connectorConfig = connectorConfig;
this.connection = connection;
this.dialect = dialect;
this.clock = clock;
this.eventDispatcher = dispatcher;
this.errorHandler = errorHandler;
Expand Down Expand Up @@ -592,6 +620,61 @@ protected void handleRowsQuery(MySqlOffsetContext offsetContext, Event event) {
offsetContext.setQuery(lastRowsQueryEventData.getQuery());
}

private void connectMariaDb(MySqlOffsetContext effectiveOffsetContext) {
metrics.setIsGtidModeEnabled(true);
eventHandlers.put(
EventType.MARIADB_GTID,
event -> handleMariadbGtidEvent(effectiveOffsetContext, event));

String restoreGtidStr = effectiveOffsetContext.gtidSet();
if (StringUtils.isNotBlank(restoreGtidStr)) {
LOGGER.info(
"MariaDB: resuming binlog from restored GTID set {} (binlog file={} pos={})",
restoreGtidStr,
effectiveOffsetContext.getSource().binlogFilename(),
effectiveOffsetContext.getSource().binlogPosition());
client.setGtidSet(restoreGtidStr);
effectiveOffsetContext.setCompletedGtidSet(restoreGtidStr);
gtidSet = new MariadbGtidSet(restoreGtidStr);
} else {
LOGGER.info(
"MariaDB: no restored GTID, starting from binlog file={} position={}",
effectiveOffsetContext.getSource().binlogFilename(),
effectiveOffsetContext.getSource().binlogPosition());
client.setBinlogFilename(effectiveOffsetContext.getSource().binlogFilename());
client.setBinlogPosition(effectiveOffsetContext.getSource().binlogPosition());
gtidSet = new MariadbGtidSet("");
}
}

protected void handleMariadbGtidEvent(MySqlOffsetContext offsetContext, Event event) {
String gtid = mariadbGtidOf(event);
gtidSet.add(gtid);
offsetContext.startGtid(gtid, gtidSet.toString());
metrics.onGtidChange(gtid);
}

/**
* Build the "domain-server-sequence" MariaDB GTID string from a "MARIADB_GTID" event. The
* server id taken from the event "header" rather than from the payload, matching how MariaDB
* records GTIDs and the server expose "gtid_binlog_pos"; using the payload server id would
* produce a GTID that does not line up with the server's own set and break GTID-based resume.
*/
static String mariadbGtidOf(Event event) {
EventData eventData = event.getData();
if (eventData instanceof EventDeserializer.EventDataWrapper) {
eventData = ((EventDeserializer.EventDataWrapper) eventData).getInternal();
}

MariadbGtidEventData gtidEventData = (MariadbGtidEventData) eventData;
EventHeader header = event.getHeader();
return gtidEventData.getDomainId()
+ "-"
+ header.getServerId()
+ "-"
+ gtidEventData.getSequence();
}

/**
* Handle the supplied event with an {@link QueryEventData} by possibly recording the DDL
* statements as changes in the MySQL schemas.
Expand Down Expand Up @@ -1125,46 +1208,55 @@ public void execute(
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.
String availableServerGtidStr = connection.knownGtidSet();
if (isGtidModeEnabled) {
// The server is using GTIDs, so enable the handler ...
eventHandlers.put(
EventType.GTID, (event) -> handleGtidEvent(effectiveOffsetContext, event));

// Now look at the GTID set from the server and what we've previously seen ...
GtidSet availableServerGtidSet = new GtidSet(availableServerGtidStr);

// also take into account purged GTID logs
GtidSet purgedServerGtidSet = connection.purgedGtidSet();
LOGGER.info("GTID set purged on server: {}", purgedServerGtidSet);

GtidSet filteredGtidSet =
filterGtidSet(
effectiveOffsetContext, availableServerGtidSet, purgedServerGtidSet);
if (filteredGtidSet != null) {
// We've seen at least some GTIDs, so start reading from the filtered GTID set ...
LOGGER.info("Registering binlog reader with GTID set: {}", filteredGtidSet);
String filteredGtidSetStr = filteredGtidSet.toString();
client.setGtidSet(filteredGtidSetStr);
effectiveOffsetContext.setCompletedGtidSet(filteredGtidSetStr);
gtidSet = new com.github.shyiko.mysql.binlog.GtidSet(filteredGtidSetStr);
if (MariaDbGtidStrategy.DIALECT.equalsIgnoreCase(dialect)) {
// MariaDB has no GTID_MODE / SHOW MASTER STATUS GTID column; resume is driven by the
// shyiko client's native MariaDB GTID support.
connectMariaDb(effectiveOffsetContext);
} else {
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.
String availableServerGtidStr = connection.knownGtidSet();
if (isGtidModeEnabled) {
// The server is using GTIDs, so enable the handler ...
eventHandlers.put(
EventType.GTID, (event) -> handleGtidEvent(effectiveOffsetContext, event));

// Now look at the GTID set from the server and what we've previously seen ...
GtidSet availableServerGtidSet = new GtidSet(availableServerGtidStr);

// also take into account purged GTID logs
GtidSet purgedServerGtidSet = connection.purgedGtidSet();
LOGGER.info("GTID set purged on server: {}", purgedServerGtidSet);

GtidSet filteredGtidSet =
filterGtidSet(
effectiveOffsetContext,
availableServerGtidSet,
purgedServerGtidSet);
if (filteredGtidSet != null) {
// We've seen at least some GTIDs, so start reading from the filtered GTID set
// ...
LOGGER.info("Registering binlog reader with GTID set: {}", filteredGtidSet);
String filteredGtidSetStr = filteredGtidSet.toString();
client.setGtidSet(filteredGtidSetStr);
effectiveOffsetContext.setCompletedGtidSet(filteredGtidSetStr);
gtidSet = new com.github.shyiko.mysql.binlog.GtidSet(filteredGtidSetStr);
} else {
// We've not yet seen any GTIDs, so that means we have to start reading the
// binlog from the beginning ...
client.setBinlogFilename(effectiveOffsetContext.getSource().binlogFilename());
client.setBinlogPosition(effectiveOffsetContext.getSource().binlogPosition());
gtidSet = new com.github.shyiko.mysql.binlog.GtidSet("");
}
} else {
// We've not yet seen any GTIDs, so that means we have to start reading the binlog
// from the beginning ...
// The server is not using GTIDs, so start reading the binlog based upon where we
// last left off ...
client.setBinlogFilename(effectiveOffsetContext.getSource().binlogFilename());
client.setBinlogPosition(effectiveOffsetContext.getSource().binlogPosition());
gtidSet = new com.github.shyiko.mysql.binlog.GtidSet("");
}
} else {
// The server is not using GTIDs, so start reading the binlog based upon where we last
// left off ...
client.setBinlogFilename(effectiveOffsetContext.getSource().binlogFilename());
client.setBinlogPosition(effectiveOffsetContext.getSource().binlogPosition());
}

// We may be restarting in the middle of a transaction, so see how far into the transaction
Expand Down
21 changes: 16 additions & 5 deletions ...nnector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/MySqlValidator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,8 @@
import org.apache.flink.cdc.connectors.mysql.debezium.DebeziumUtils;
import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig;
import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceOptions;
import org.apache.flink.cdc.connectors.mysql.source.offset.MariaDbGtidStrategy;
import org.apache.flink.cdc.connectors.mysql.source.offset.MysqlGtidStrategy;
import org.apache.flink.cdc.debezium.Validator;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.ValidationException;
Expand Down Expand Up @@ -69,11 +71,20 @@ public MySqlValidator(MySqlSourceConfig sourceConfig) {
@Override
public void validate() {
try (JdbcConnection connection = createJdbcConnection(sourceConfig, dbzProperties)) {
checkVersion(connection);
checkBinlogFormat(connection);
checkBinlogRowImage(connection);
checkBinlogRowValueOptions(connection);
checkTimeZone(connection);
String resolvedDialect =
sourceConfig != null
? DebeziumUtils.discoverDialect(connection, sourceConfig.getDialect())
: MysqlGtidStrategy.DIALECT;
if (MariaDbGtidStrategy.DIALECT.equalsIgnoreCase(resolvedDialect)) {
checkBinlogFormat(connection);
LOG.info("MariaDB dialect detected; skipping MySQL-only validation checks.");
} else {
checkVersion(connection);
checkBinlogFormat(connection);
checkBinlogRowImage(connection);
checkBinlogRowValueOptions(connection);
checkTimeZone(connection);
}
} catch (SQLException ex) {
throw new TableException(
"Unexpected error while connecting to MySQL and validating", ex);
Expand Down
Loading