diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlConnection.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlConnection.java index aa9699e4eaa..a6c499a68d9 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlConnection.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlConnection.java @@ -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. * diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java index 557a149a2d0..4ddd24fdcbe 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java @@ -6,8 +6,12 @@ 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; @@ -15,6 +19,7 @@ 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; @@ -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; @@ -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 eventDispatcher; private final ErrorHandler errorHandler; @@ -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 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; @@ -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. @@ -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 diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/MySqlValidator.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/MySqlValidator.java index 3f5f18befe2..9bbad03f3e4 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/MySqlValidator.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/MySqlValidator.java @@ -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; @@ -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); diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/DebeziumUtils.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/DebeziumUtils.java index f567dde7347..6392720123a 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/DebeziumUtils.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/DebeziumUtils.java @@ -20,6 +20,10 @@ import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig; import org.apache.flink.cdc.connectors.mysql.source.connection.JdbcConnectionFactory; import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset; +import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffsetBuilder; +import org.apache.flink.cdc.connectors.mysql.source.offset.GtidStrategies; +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.connectors.mysql.source.utils.TableDiscoveryUtils; import org.apache.flink.util.FlinkRuntimeException; @@ -43,6 +47,7 @@ import io.debezium.relational.Tables; import io.debezium.schema.TopicSelector; import io.debezium.util.SchemaNameAdjuster; +import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -102,6 +107,20 @@ public static BinaryLogClient createBinaryClient(Configuration dbzConfiguration) connectorConfig.password()); } + public static BinaryLogClient createBinaryClient( + Configuration dbzConfiguration, String dialect) { + if (MysqlGtidStrategy.DIALECT.equalsIgnoreCase(dialect)) { + return createBinaryClient(dbzConfiguration); + } + + final MySqlConnectorConfig connectorConfig = new MySqlConnectorConfig(dbzConfiguration); + return new MariaDBBinaryLogClient( + connectorConfig.hostname(), + connectorConfig.port(), + connectorConfig.username(), + connectorConfig.password()); + } + /** Creates a new {@link MySqlDatabaseSchema} to monitor the latest MySql database schemas. */ public static MySqlDatabaseSchema createMySqlDatabaseSchema( MySqlConnectorConfig dbzMySqlConfig, boolean isTableIdCaseSensitive) { @@ -116,6 +135,45 @@ public static MySqlDatabaseSchema createMySqlDatabaseSchema( isTableIdCaseSensitive); } + /** + * Fetch the current binlog offset, dialect-aware. For MariaDB the GTID set is read from + * {@code @@gtid_binlog_pos}. For MySQL the behavior is identical to {@link + * #currentBinlogOffset(MySqlConnection)} + */ + public static BinlogOffset currentBinlogOffset(MySqlConnection jdbc, String resolvedDialect) { + if (!MariaDbGtidStrategy.DIALECT.equalsIgnoreCase(resolvedDialect)) { + return currentBinlogOffset(jdbc); + } + + final String showMasterStmt = "Show MASTER STATUS"; + try { + BinlogOffsetBuilder builder = + jdbc.queryAndMap( + showMasterStmt, + rs -> { + if (rs.next()) { + return BinlogOffset.builder() + .setBinlogFilePosition(rs.getString(1), rs.getLong(2)); + } + throw new FlinkRuntimeException( + "Cannot read MariaDB binlog filename/position via '" + + showMasterStmt + + "'. Make sure your server is correctly configured."); + }); + String gtidSet = + jdbc.queryAndMap( + "SELECT @@gtid_binlog_pos", rs -> rs.next() ? rs.getString(1) : null); + if (StringUtils.isNotBlank(gtidSet)) { + builder.setGtidSet(gtidSet); + } + + return builder.build(); + } catch (SQLException e) { + throw new FlinkRuntimeException( + "Cannot read MariaDB binlog offset via '" + showMasterStmt + "'."); + } + } + /** Fetch current binlog offsets in MySql Server. */ public static BinlogOffset currentBinlogOffset(MySqlConnection jdbc) { String showBinaryLogStatement = jdbc.getShowBinaryLogStatement(); @@ -148,6 +206,29 @@ public static BinlogOffset currentBinlogOffset(MySqlConnection jdbc) { } } + /** + * Resolves the effective dialect for a server: returns the pinned value verbatim when + * "configuredDialect" is "mysql"/"mariadb",otherwise run "SELECT VERSION()". + */ + public static String discoverDialect(JdbcConnection jdbc, String configuredDialect) { + if (MysqlGtidStrategy.DIALECT.equalsIgnoreCase(configuredDialect) + || MariaDbGtidStrategy.DIALECT.equalsIgnoreCase(configuredDialect)) { + return GtidStrategies.resolveDialect(configuredDialect, null); + } + + String versionText = null; + try { + versionText = + jdbc.queryAndMap("SELECT VERSION()", rs -> rs.next() ? rs.getString(1) : ""); + } catch (SQLException e) { + LOG.warn( + "Failed to run SELECT VERSION() for dialect auto-detect; " + + "falling back to mysql.", + e); + } + return GtidStrategies.resolveDialect(configuredDialect, versionText); + } + /** Create a TableFilter by database name and table name. */ public static Tables.TableFilter createTableFilter(String database, String table) { final Selectors.TableSelectionPredicateBuilder eligibleTables = diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/MariaDBBinaryLogClient.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/MariaDBBinaryLogClient.java new file mode 100644 index 00000000000..ab42c6cbb98 --- /dev/null +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/MariaDBBinaryLogClient.java @@ -0,0 +1,82 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package org.apache.flink.cdc.connectors.mysql.debezium; + +import com.github.shyiko.mysql.binlog.BinaryLogClient; +import com.github.shyiko.mysql.binlog.network.protocol.command.Command; +import com.github.shyiko.mysql.binlog.network.protocol.command.DumpBinaryLogCommand; +import com.github.shyiko.mysql.binlog.network.protocol.command.QueryCommand; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; + +/** + * A {@link BinaryLogClient} for MariaDB that raises {@code @mariadb_slave_capability} from 1 -> 4 + * before requesting the binlog stream. + * + *

{@code mysql-binlog-connector-java} 0.27.2 hard-codes {@code SET @mariadb_slave_capability=1} + * in {@code com.github.shyiko.mysql.binlog.BinaryLogClient#requestBinaryLogStreamMaria(long)}. With + * capability 1 the MariaDB server keeps the legacy replication protocol and server emits + * per-transaction {@code MARIADB_GTID} event, so the consumed GTID position can never advance (see + * MariaDB MDEV-225 ...). Capability 4 makes + * the server deliver one "MARIADB_GTID" event per transaction, which is what lets the CDC offset's + * GTID set move forward across ckp and failover. + * + *

Only the MariaDB request path is overridden; the MySQL path is inherited unchanged. + */ +public class MariaDBBinaryLogClient extends BinaryLogClient { + + private static final Logger LOG = LoggerFactory.getLogger(MariaDBBinaryLogClient.class); + + public MariaDBBinaryLogClient(String hostname, int port, String username, String password) { + super(hostname, port, username, password); + } + + @Override + protected void requestBinaryLogStreamMaria(long serverId) throws IOException { + Command dumpBinaryLogCommand; + + // https://jira.mariadb.org/browse/MDEV-225 - capability 4 (not the stock 1) is required for + // the server to send per-txn MARIADB_GTID events, to the consumed GTID can advance. + channel.write(new QueryCommand("SET @mariadb_slave_capability=4")); + checkError(channel.read()); + + synchronized (gtidSetAccessLock) { + if (gtidSet != null) { + LOG.debug("MariaDB slave_connect_state = {}", gtidSet); + channel.write( + new QueryCommand( + "SET @slave_connect_state = '" + gtidSet.toString() + "'")); + checkError(channel.read()); + channel.write(new QueryCommand("SET @slave_gtid_strict_mode = 0")); + checkError(channel.read()); + channel.write(new QueryCommand("SET @slave_gtid_ignore_duplicates = 0")); + checkError(channel.read()); + dumpBinaryLogCommand = + new DumpBinaryLogCommand(serverId, "", 0L, isUseSendAnnotateRowsEvent()); + } else { + dumpBinaryLogCommand = + new DumpBinaryLogCommand( + serverId, getBinlogFilename(), getBinlogPosition()); + } + } + + channel.write(dumpBinaryLogCommand); + } +} diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java index b1e6d1dfc86..f7c95ee84dc 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java @@ -22,6 +22,7 @@ import org.apache.flink.cdc.connectors.mysql.debezium.task.context.StatefulTaskContext; import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig; import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset; +import org.apache.flink.cdc.connectors.mysql.source.offset.MariaDbGtidStrategy; import org.apache.flink.cdc.connectors.mysql.source.split.FinishedSnapshotSplitInfo; import org.apache.flink.cdc.connectors.mysql.source.split.MySqlBinlogSplit; import org.apache.flink.cdc.connectors.mysql.source.split.MySqlSplit; @@ -101,7 +102,8 @@ public BinlogSplitReader(MySqlSourceConfig sourceConfig, int subtaskId) { this( new StatefulTaskContext( sourceConfig, - createBinaryClient(sourceConfig.getDbzConfiguration()), + createBinaryClient( + sourceConfig.getDbzConfiguration(), sourceConfig.getDialect()), createMySqlConnection(sourceConfig)), subtaskId); } @@ -123,6 +125,13 @@ public void submitSplit(MySqlSplit mySqlSplit) { this.currentBinlogSplit = mySqlSplit.asBinlogSplit(); configureFilter(); statefulTaskContext.configure(currentBinlogSplit); + if (MariaDbGtidStrategy.DIALECT.equals(statefulTaskContext.getResolvedDialect())) { + LOG.info( + "MariaDB binlog split is submitted: startingOffset={} capturedSchemas={} finishedSnapshotSplits={}", + currentBinlogSplit.getStartingOffset(), + currentBinlogSplit.getTableSchemas().keySet(), + currentBinlogSplit.getFinishedSnapshotSplitInfos().size()); + } this.capturedTableFilter = statefulTaskContext.getSourceConfig().getTableFilter(); this.queue = statefulTaskContext.getQueue(); this.binlogSplitReadTask = @@ -137,7 +146,8 @@ public void submitSplit(MySqlSplit mySqlSplit) { (MySqlStreamingChangeEventSourceMetrics) statefulTaskContext.getStreamingChangeEventSourceMetrics(), currentBinlogSplit, - createEventFilter()); + createEventFilter(), + statefulTaskContext.getResolvedDialect()); executorService.submit( () -> { diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/SnapshotSplitReader.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/SnapshotSplitReader.java index e3549237065..6b2c9aa50f7 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/SnapshotSplitReader.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/SnapshotSplitReader.java @@ -102,7 +102,8 @@ public SnapshotSplitReader( this( new StatefulTaskContext( sourceConfig, - createBinaryClient(sourceConfig.getDbzConfiguration()), + createBinaryClient( + sourceConfig.getDbzConfiguration(), sourceConfig.getDialect()), createMySqlConnection(sourceConfig)), subtaskId, hooks); @@ -149,7 +150,8 @@ public void submitSplit(MySqlSplit mySqlSplit) { StatefulTaskContext.getClock(), currentSnapshotSplit, hooks, - statefulTaskContext.getSourceConfig().isSkipSnapshotBackfill()); + statefulTaskContext.getSourceConfig().isSkipSnapshotBackfill(), + statefulTaskContext.getResolvedDialect()); executorService.execute( () -> { try { @@ -254,7 +256,8 @@ private MySqlBinlogSplitReadTask createBackfillBinlogReadTask( (MySqlStreamingChangeEventSourceMetrics) statefulTaskContext.getStreamingChangeEventSourceMetrics(), backfillBinlogSplit, - event -> true); + event -> true, + statefulTaskContext.getResolvedDialect()); } private void dispatchBinlogEndEvent(MySqlBinlogSplit backFillBinlogSplit) diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/task/MySqlBinlogSplitReadTask.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/task/MySqlBinlogSplitReadTask.java index 7a6e7757f7d..4c55c85de07 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/task/MySqlBinlogSplitReadTask.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/task/MySqlBinlogSplitReadTask.java @@ -68,8 +68,17 @@ public MySqlBinlogSplitReadTask( MySqlTaskContext taskContext, MySqlStreamingChangeEventSourceMetrics metrics, MySqlBinlogSplit binlogSplit, - Predicate eventFilter) { - super(connectorConfig, connection, dispatcher, errorHandler, clock, taskContext, metrics); + Predicate eventFilter, + String dialect) { + super( + connectorConfig, + connection, + dispatcher, + errorHandler, + clock, + taskContext, + metrics, + dialect); this.binlogSplit = binlogSplit; this.eventDispatcher = dispatcher; this.errorHandler = errorHandler; diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/task/MySqlSnapshotSplitReadTask.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/task/MySqlSnapshotSplitReadTask.java index f729f59d05f..a6c4012283d 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/task/MySqlSnapshotSplitReadTask.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/task/MySqlSnapshotSplitReadTask.java @@ -83,6 +83,7 @@ public class MySqlSnapshotSplitReadTask private final SnapshotPhaseHooks hooks; private final boolean isBackfillSkipped; + private final String resolvedDialect; public MySqlSnapshotSplitReadTask( MySqlSourceConfig sourceConfig, @@ -96,7 +97,8 @@ public MySqlSnapshotSplitReadTask( Clock clock, MySqlSnapshotSplit snapshotSplit, SnapshotPhaseHooks hooks, - boolean isBackfillSkipped) { + boolean isBackfillSkipped, + String resolvedDialect) { super(connectorConfig, snapshotChangeEventSourceMetrics); this.sourceConfig = sourceConfig; this.databaseSchema = databaseSchema; @@ -109,6 +111,7 @@ public MySqlSnapshotSplitReadTask( this.snapshotChangeEventSourceMetrics = snapshotChangeEventSourceMetrics; this.hooks = hooks; this.isBackfillSkipped = isBackfillSkipped; + this.resolvedDialect = resolvedDialect; } @Override @@ -153,7 +156,8 @@ protected SnapshotResult doExecute( if (hooks.getPreLowWatermarkAction() != null) { hooks.getPreLowWatermarkAction().accept(jdbcConnection, snapshotSplit); } - final BinlogOffset lowWatermark = DebeziumUtils.currentBinlogOffset(jdbcConnection); + final BinlogOffset lowWatermark = + DebeziumUtils.currentBinlogOffset(jdbcConnection, resolvedDialect); LOG.info( "Snapshot step 1 - Determining low watermark {} for split {}", lowWatermark, @@ -186,7 +190,7 @@ protected SnapshotResult doExecute( highWatermark = lowWatermark; } else { // Get the current binlog offset as HW - highWatermark = DebeziumUtils.currentBinlogOffset(jdbcConnection); + highWatermark = DebeziumUtils.currentBinlogOffset(jdbcConnection, resolvedDialect); } LOG.info( diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/task/context/StatefulTaskContext.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/task/context/StatefulTaskContext.java index 09a5fe8e85c..641e20f2593 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/task/context/StatefulTaskContext.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/task/context/StatefulTaskContext.java @@ -23,6 +23,10 @@ import org.apache.flink.cdc.connectors.mysql.debezium.dispatcher.SignalEventDispatcher; import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig; import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset; +import org.apache.flink.cdc.connectors.mysql.source.offset.GtidStrategies; +import org.apache.flink.cdc.connectors.mysql.source.offset.GtidStrategy; +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.connectors.mysql.source.split.MySqlSplit; import com.github.shyiko.mysql.binlog.BinaryLogClient; @@ -53,6 +57,7 @@ import io.debezium.util.Clock; import io.debezium.util.Collect; import io.debezium.util.SchemaNameAdjuster; +import org.apache.commons.lang3.StringUtils; import org.apache.kafka.connect.data.Struct; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -84,6 +89,8 @@ public class StatefulTaskContext implements AutoCloseable { private final MySqlConnection connection; private final BinaryLogClient binaryLogClient; + private String resolvedDialect = MysqlGtidStrategy.DIALECT; + private MySqlDatabaseSchema databaseSchema; private MySqlTaskContextImpl taskContext; private MySqlOffsetContext offsetContext; @@ -112,6 +119,9 @@ public StatefulTaskContext( public void configure(MySqlSplit mySqlSplit) { // initial stateful objects final boolean tableIdCaseInsensitive = connection.isTableIdCaseSensitive(); + // Resolve the effective dialect once; reused by the offset initialization, + // GTID restore check, and the streaming connect path. + this.resolvedDialect = DebeziumUtils.discoverDialect(connection, sourceConfig.getDialect()); this.topicSelector = MySqlTopicSelector.defaultSelector(connectorConfig); EmbeddedFlinkDatabaseHistory.registerHistory( sourceConfig @@ -195,7 +205,8 @@ protected MySqlOffsetContext loadStartingOffsetState( : initializeEffectiveOffset( mySqlSplit.asBinlogSplit().getStartingOffset(), connection, - sourceConfig); + sourceConfig, + resolvedDialect); LOG.info("Starting offset is initialized to {}", offset); @@ -227,6 +238,10 @@ private boolean checkGtidSet(MySqlOffsetContext offset) { return true; // start at beginning ... } + if (MariaDbGtidStrategy.DIALECT.equalsIgnoreCase(resolvedDialect)) { + return checkMariadbGtidSet(gtidStr); + } + String availableGtidStr = connection.knownGtidSet(); if (availableGtidStr == null || availableGtidStr.trim().isEmpty()) { // Last offsets had GTIDs but the server does not use them ... @@ -275,6 +290,25 @@ private boolean checkGtidSet(MySqlOffsetContext offset) { return false; } + private boolean checkMariadbGtidSet(String restoredGtid) { + String available = connection.mariadbGtidExecuted(); + if (StringUtils.isBlank(available)) { + LOG.warn( + "Connector used MariaDB GTID previously, but the server reports no @@gtid_binlog_pos"); + return false; + } + + GtidStrategy strategy = GtidStrategies.of(MariaDbGtidStrategy.DIALECT); + boolean contained = strategy.isContainedWithin(restoredGtid, available); + if (!contained) { + LOG.info( + "MariaDB restored GTID set {} is not contained within server set {}", + restoredGtid, + available); + } + return contained; + } + private boolean checkBinlogFilename(MySqlOffsetContext offset) { String binlogFilename = offset.getSourceInfo().getString(BINLOG_FILENAME_OFFSET_KEY); if (binlogFilename == null) { @@ -388,6 +422,10 @@ public MySqlConnection getConnection() { return connection; } + public String getResolvedDialect() { + return resolvedDialect; + } + public BinaryLogClient getBinaryLogClient() { return binaryLogClient; } diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceBuilder.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceBuilder.java index 93fa2a0d36a..c4e29a6bf3c 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceBuilder.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceBuilder.java @@ -211,6 +211,15 @@ public MySqlSourceBuilder scanNewlyAddedTableEnabled(boolean scanNewlyAddedTa return this; } + /** + * The database dialect of the source server: "mysql"/"mariadb"/"auto" Defaults to "auto" + * (detect via SELECT VERSION() at startup). + */ + public MySqlSourceBuilder dialect(String dialect) { + this.configFactory.dialect(dialect); + return this; + } + /** Specifies the startup options. */ public MySqlSourceBuilder startupOptions(StartupOptions startupOptions) { this.configFactory.startupOptions(startupOptions); diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java index cf456fcaed0..548d5fb6605 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java @@ -72,6 +72,7 @@ public class MySqlSourceConfig implements Serializable { private final boolean parseOnLineSchemaChanges; public static boolean useLegacyJsonFormat = true; private final boolean assignUnboundedChunkFirst; + private final String dialect; // -------------------------------------------------------------------------------------------- // Debezium Configurations @@ -112,7 +113,8 @@ public class MySqlSourceConfig implements Serializable { boolean parseOnLineSchemaChanges, boolean treatTinyInt1AsBoolean, boolean useLegacyJsonFormat, - boolean assignUnboundedChunkFirst) { + boolean assignUnboundedChunkFirst, + String dialect) { this.hostname = checkNotNull(hostname); this.port = port; this.username = checkNotNull(username); @@ -158,6 +160,7 @@ public class MySqlSourceConfig implements Serializable { this.treatTinyInt1AsBoolean = treatTinyInt1AsBoolean; this.useLegacyJsonFormat = useLegacyJsonFormat; this.assignUnboundedChunkFirst = assignUnboundedChunkFirst; + this.dialect = dialect; } public String getHostname() { @@ -299,4 +302,8 @@ public boolean isSkipSnapshotBackfill() { public boolean isTreatTinyInt1AsBoolean() { return treatTinyInt1AsBoolean; } + + public String getDialect() { + return dialect; + } } diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java index 569b62232db..16baa6b26c1 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java @@ -20,6 +20,7 @@ import org.apache.flink.cdc.common.annotation.Internal; import org.apache.flink.cdc.connectors.mysql.debezium.EmbeddedFlinkDatabaseHistory; import org.apache.flink.cdc.connectors.mysql.source.MySqlSource; +import org.apache.flink.cdc.connectors.mysql.source.offset.GtidStrategies; import org.apache.flink.cdc.connectors.mysql.table.StartupOptions; import org.apache.flink.table.catalog.ObjectPath; @@ -78,6 +79,7 @@ public class MySqlSourceConfigFactory implements Serializable { private boolean treatTinyInt1AsBoolean = true; private boolean useLegacyJsonFormat = true; private boolean assignUnboundedChunkFirst = false; + private String dialect = GtidStrategies.AUTO; public MySqlSourceConfigFactory hostname(String hostname) { this.hostname = hostname; @@ -332,6 +334,11 @@ public MySqlSourceConfigFactory treatTinyInt1AsBoolean(boolean treatTinyInt1AsBo return this; } + public MySqlSourceConfigFactory dialect(String dialect) { + this.dialect = dialect; + return this; + } + /** * Whether to assign the unbounded chunks first during snapshot reading phase. Defaults to * false. @@ -355,6 +362,12 @@ public MySqlSourceConfig createConfig(int subtaskId) { /** Creates a new {@link MySqlSourceConfig} for the given subtask {@code subtaskId}. */ public MySqlSourceConfig createConfig(int subtaskId, String serverName) { checkSupportCheckpointsAfterTasksFinished(closeIdleReaders); + if (!GtidStrategies.isSupportedDialect(dialect)) { + throw new IllegalArgumentException( + String.format( + "Invalid value for 'dialect'. Supported dialects are: [mysql, mariadb, auto], but was: %s", + dialect)); + } Properties props = new Properties(); props.setProperty("database.server.name", serverName); props.setProperty("database.hostname", checkNotNull(hostname)); @@ -444,6 +457,7 @@ public MySqlSourceConfig createConfig(int subtaskId, String serverName) { parseOnLineSchemaChanges, treatTinyInt1AsBoolean, useLegacyJsonFormat, - assignUnboundedChunkFirst); + assignUnboundedChunkFirst, + dialect); } } diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceOptions.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceOptions.java index a8e143f5fc5..6e77a9f93cd 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceOptions.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceOptions.java @@ -292,4 +292,15 @@ public class MySqlSourceOptions { .defaultValue(true) .withDescription( "Whether to assign the unbounded chunks first during snapshot reading phase. This might help reduce the risk of the TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of the largest unbounded chunk."); + public static final ConfigOption SCAN_DIALECT = + ConfigOptions.key("scan.dialect") + .stringType() + .defaultValue("auto") + .withDescription( + "The database dialect of the source server: one of \"mysql\", \"mariadb\" or \"auto\". " + + "When \"auto\"(default), the dialect is detect from SELECT VERSION() at startup." + + "Pin it to \"mysql\"/\"mariadb\" to skip detection. The resolved dialect selects" + + "the GTID strategy used to read the binlog stream and to cap the restored GTID" + + "set on recovery, so MariaDB sources advance and resume from MariaDB GTIDs" + + "(domian-server-sequence) rather than MySQL GTIDs (UUID:txn)."); } diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/BinlogOffset.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/BinlogOffset.java index a7f86b62ea3..70a0757287b 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/BinlogOffset.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/BinlogOffset.java @@ -21,7 +21,6 @@ import org.apache.flink.cdc.common.annotation.PublicEvolving; import org.apache.flink.cdc.common.annotation.VisibleForTesting; -import io.debezium.connector.mysql.GtidSet; import org.apache.commons.lang3.StringUtils; import org.apache.kafka.connect.errors.ConnectException; @@ -188,9 +187,11 @@ public int compareTo(BinlogOffset that) { // The target offset uses GTIDs, so we ideally compare using GTIDs ... if (StringUtils.isNotEmpty(gtidSetStr)) { // Both have GTIDs, so base the comparison entirely on the GTID sets. - GtidSet gtidSet = new GtidSet(gtidSetStr); - GtidSet targetGtidSet = new GtidSet(targetGtidSetStr); - if (gtidSet.equals(targetGtidSet)) { + // The dialect is recovered from the GTID text, + // so MariaDB sets compare with server-id-ignoring semantics + // while MySQL keeps its uuid:interval behavior. + GtidStrategy strategy = GtidStrategies.detect(targetGtidSetStr); + if (strategy.isEqual(gtidSetStr, targetGtidSetStr)) { long restartSkipEvents = this.getRestartSkipEvents(); long targetRestartSkipEvents = that.getRestartSkipEvents(); if (restartSkipEvents != targetRestartSkipEvents) { @@ -202,7 +203,7 @@ public int compareTo(BinlogOffset that) { // The GTIDs are not an exact match, so figure out if this is a subset of the target // offset // ... - return gtidSet.isContainedWithin(targetGtidSet) ? -1 : 1; + return strategy.isContainedWithin(gtidSetStr, targetGtidSetStr) ? -1 : 1; } // The target offset did use GTIDs while this did not use GTIDs. So, we assume // that this offset is older since GTIDs are often enabled but rarely disabled. diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/BinlogOffsetUtils.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/BinlogOffsetUtils.java index 3c192eaa779..acb58c3aba0 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/BinlogOffsetUtils.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/BinlogOffsetUtils.java @@ -42,11 +42,14 @@ public class BinlogOffsetUtils { *

  • EARLIEST: binlog filename = "", position = 0 *
  • TIMESTAMP: set to earliest, as the current implementation is reading from the earliest * offset and drop events earlier than the specified timestamp. - *
  • LATEST: fetch the current binlog by JDBC + *
  • LATEST: fetch the current binlog by JDBC(dialect-aware for MariaDB GTID) * */ public static BinlogOffset initializeEffectiveOffset( - BinlogOffset offset, MySqlConnection connection, MySqlSourceConfig mySqlSourceConfig) { + BinlogOffset offset, + MySqlConnection connection, + MySqlSourceConfig mySqlSourceConfig, + String resolvedDialect) { BinlogOffsetKind offsetKind = offset.getOffsetKind(); switch (offsetKind) { case EARLIEST: @@ -55,7 +58,7 @@ public static BinlogOffset initializeEffectiveOffset( return DebeziumUtils.findBinlogOffset( offset.getTimestampSec() * 1000, connection, mySqlSourceConfig); case LATEST: - return DebeziumUtils.currentBinlogOffset(connection); + return DebeziumUtils.currentBinlogOffset(connection, resolvedDialect); default: return offset; } diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/GtidStrategies.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/GtidStrategies.java new file mode 100644 index 00000000000..960107e3791 --- /dev/null +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/GtidStrategies.java @@ -0,0 +1,77 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package org.apache.flink.cdc.connectors.mysql.source.offset; + +/** Resolves GTID strategies for supported database dialects. */ +public class GtidStrategies { + + /** Config sentinel: detect the dialect from "SELECT VERSION()" at startup. */ + public static final String AUTO = "auto"; + + private static final GtidStrategy MYSQL = new MysqlGtidStrategy(); + private static final GtidStrategy MARIADB = new MariaDbGtidStrategy(); + + private GtidStrategies() {} + + public static boolean isSupportedDialect(String dialect) { + return MariaDbGtidStrategy.DIALECT.equalsIgnoreCase(dialect) + || MysqlGtidStrategy.DIALECT.equalsIgnoreCase(dialect) + || AUTO.equalsIgnoreCase(dialect); + } + + public static GtidStrategy of(String dialect) { + if (MariaDbGtidStrategy.DIALECT.equalsIgnoreCase(dialect)) { + return MARIADB; + } + return MYSQL; + } + + /** + * Resolves the effective dialect from the configured value and the server's "VERSION()" banner. + * A pinned "mysql/mariadb" wins outright; "auto" or any unrecognized value, including "null" + * sniffs the version text, which carries the literal "MariaDB" on MariaDB servers and not on + * MySQL. Default to "mysql". + */ + public static String resolveDialect(String configured, String versionText) { + if (MysqlGtidStrategy.DIALECT.equalsIgnoreCase(configured)) { + return MysqlGtidStrategy.DIALECT; + } + if (MariaDbGtidStrategy.DIALECT.equalsIgnoreCase(configured)) { + return MariaDbGtidStrategy.DIALECT; + } + + if (versionText != null + && versionText.toLowerCase().contains(MariaDbGtidStrategy.DIALECT)) { + return MariaDbGtidStrategy.DIALECT; + } + return MysqlGtidStrategy.DIALECT; + } + + /** + * Recovers the strategy from the GTID text alone, so a stateless {@link BinlogOffset} can still + * pick the correct semantics after a restore. MariaDB's GTID (domain-server-sequence) form has + * no ':', which is exactly what distinguishes it from MySQL's form (uuid:interval). + */ + public static GtidStrategy detect(String gtidText) { + if (MARIADB.canParse(gtidText) && !MYSQL.canParse(gtidText)) { + return MARIADB; + } + + return MYSQL; + } +} diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/GtidStrategy.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/GtidStrategy.java new file mode 100644 index 00000000000..f8ea0d530a4 --- /dev/null +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/GtidStrategy.java @@ -0,0 +1,34 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package org.apache.flink.cdc.connectors.mysql.source.offset; + +import java.io.Serializable; + +/** Dialect-specific operations for GTID parsing, comparison, and recovery. */ +public interface GtidStrategy extends Serializable { + + /** Identifier of this dialect, e.g. "mysql" or "mariadb". */ + String dialect(); + + boolean canParse(String gtidText); + + boolean isEqual(String a, String b); + + /** Whether the {@code sub} GTID set is fully contained within {@code sup}. */ + boolean isContainedWithin(String sub, String sup); +} diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/MariaDbGtidStrategy.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/MariaDbGtidStrategy.java new file mode 100644 index 00000000000..520445b3dd1 --- /dev/null +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/MariaDbGtidStrategy.java @@ -0,0 +1,101 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package org.apache.flink.cdc.connectors.mysql.source.offset; + +import com.github.shyiko.mysql.binlog.MariadbGtidSet; +import org.apache.commons.lang3.StringUtils; + +import java.util.HashMap; +import java.util.Map; + +/** + * MariaDB (domain-server-sequence) GTID strategy. + * + *

    Unlike MySQL's (uuid:interval) model, a MariaDB GTID identifies a position by (domain, server, + * sequence) where the sequence is monotonically increasing per domain across servers. On + * master/replica failover the server id changes but the same domain keeps advancing the sequence, + * so comparison must be keyed on the domain and ignore the server id - otherwise a post-failover + * event looks like a different stream and the offset is mishandled (Debezium dbz#1672, Flink CDC + * issue #2929). + * + *

    The underlying {@code com.github.shyiko} binlog library already parses MariaDB GTID text + * ({@link MariadbGtidSet#isMariaGtidSet(String)}), but its {@code equals()} and {@code + * isContainedWithin()} methods compare the server ID too. The domain-keyed comparison is therefore + * implemented here directly. + */ +public class MariaDbGtidStrategy implements GtidStrategy { + + private static final long seriaVersionUID = 1L; + + public static final String DIALECT = "mariadb"; + + @Override + public String dialect() { + return DIALECT; + } + + @Override + public boolean canParse(String gtidText) { + return gtidText != null && MariadbGtidSet.isMariaGtidSet(gtidText.trim()); + } + + @Override + public boolean isEqual(String a, String b) { + return toDomainSequence(a).equals(toDomainSequence(b)); + } + + /** + * MariaDB GTIDs have the form {@code domain-server-sequence}. A primary failover may change the + * server ID while the same domain continues advancing. Containment is therefore evaluated by + * {@code domain + sequence}; the server ID is deliberately ignored. For example, {@code + * 1-100-500} is contained within {@code 1-101-520}, but {@code 1-100-521} is not. If a domain + * present in {@code sub} is absent from {@code sup}, containment is also false. + */ + @Override + public boolean isContainedWithin(String sub, String sup) { + Map supSequence = toDomainSequence(sup); + return toDomainSequence(sub).entrySet().stream() + .allMatch( + entry -> supSequence.getOrDefault(entry.getKey(), -1L) >= entry.getValue()); + } + + /** Converts GTID text into the highest observed sequence for each replication domain. */ + private static Map toDomainSequence(String gtidText) { + if (StringUtils.isBlank(gtidText)) { + return new HashMap<>(); + } + + Map domainToSeq = new HashMap<>(); + for (String tuple : gtidText.split(",")) { + String trimmedTuple = tuple.trim(); + if (StringUtils.isBlank(trimmedTuple)) { + continue; + } + + String[] parts = trimmedTuple.split("-"); + if (parts.length != 3) { + throw new IllegalArgumentException("Invalid MariaDB gtid format: " + trimmedTuple); + } + + Long domain = Long.parseLong(parts[0].trim()); + Long sequence = Long.parseLong(parts[2].trim()); + domainToSeq.merge(domain, sequence, Math::max); + } + return domainToSeq; + } +} diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/MysqlGtidStrategy.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/MysqlGtidStrategy.java new file mode 100644 index 00000000000..40ef5a5cf88 --- /dev/null +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/MysqlGtidStrategy.java @@ -0,0 +1,60 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package org.apache.flink.cdc.connectors.mysql.source.offset; + +import io.debezium.connector.mysql.GtidSet; + +/** MySQL UUID-and-interval GTID strategy. */ +public class MysqlGtidStrategy implements GtidStrategy { + + private static final long seriaVersionUID = 1L; + + public static final String DIALECT = "mysql"; + + @Override + public String dialect() { + return DIALECT; + } + + @Override + public boolean canParse(String gtidText) { + if (gtidText == null) { + return false; + } + + try { + new GtidSet(gtidText); + + // MySQL GTID sets are uuid:interval; the ':' separator distinguishes them from + // MariaDB's domain-server-sequence form, which GtidSet would silently mis-parse. + return gtidText.isBlank() || gtidText.contains(":"); + } catch (RuntimeException e) { + return false; + } + } + + @Override + public boolean isEqual(String a, String b) { + return new GtidSet(a).equals(new GtidSet(b)); + } + + @Override + public boolean isContainedWithin(String sub, String sup) { + return new GtidSet(sub).isContainedWithin(new GtidSet(sup)); + } +} diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSource.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSource.java index 2a1f0519435..c243cc38d56 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSource.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSource.java @@ -103,6 +103,7 @@ public class MySqlTableSource implements ScanTableSource, SupportsReadingMetadat private final boolean assignUnboundedChunkFirst; private final boolean appendOnly; + private final String dialect; // -------------------------------------------------------------------------------------------- // Mutable attributes @@ -144,7 +145,8 @@ public MySqlTableSource( boolean parseOnlineSchemaChanges, boolean useLegacyJsonFormat, boolean assignUnboundedChunkFirst, - boolean appendOnly) { + boolean appendOnly, + String dialect) { this.physicalSchema = physicalSchema; this.port = port; this.hostname = checkNotNull(hostname); @@ -178,6 +180,7 @@ public MySqlTableSource( this.useLegacyJsonFormat = useLegacyJsonFormat; this.assignUnboundedChunkFirst = assignUnboundedChunkFirst; this.appendOnly = appendOnly; + this.dialect = dialect; } @Override @@ -233,6 +236,7 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) { .startupOptions(startupOptions) .deserializer(deserializer) .scanNewlyAddedTableEnabled(scanNewlyAddedTableEnabled) + .dialect(dialect) .closeIdleReaders(closeIdleReaders) .jdbcProperties(jdbcProperties) .heartbeatInterval(heartbeatInterval) @@ -330,7 +334,8 @@ public DynamicTableSource copy() { parseOnlineSchemaChanges, useLegacyJsonFormat, assignUnboundedChunkFirst, - appendOnly); + appendOnly, + dialect); source.metadataKeys = metadataKeys; source.producedDataType = producedDataType; return source; @@ -376,7 +381,8 @@ public boolean equals(Object o) { && parseOnlineSchemaChanges == that.parseOnlineSchemaChanges && useLegacyJsonFormat == that.useLegacyJsonFormat && assignUnboundedChunkFirst == that.assignUnboundedChunkFirst - && Objects.equals(appendOnly, that.appendOnly); + && Objects.equals(appendOnly, that.appendOnly) + && Objects.equals(dialect, that.dialect); } @Override @@ -413,7 +419,8 @@ public int hashCode() { parseOnlineSchemaChanges, useLegacyJsonFormat, assignUnboundedChunkFirst, - appendOnly); + appendOnly, + dialect); } @Override diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactory.java index 5ea430d94e7..d8614357f26 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactory.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactory.java @@ -21,6 +21,8 @@ import org.apache.flink.cdc.connectors.mysql.source.config.ServerIdRange; import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset; import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffsetBuilder; +import org.apache.flink.cdc.connectors.mysql.source.offset.GtidStrategies; +import org.apache.flink.cdc.connectors.mysql.source.offset.MariaDbGtidStrategy; import org.apache.flink.cdc.connectors.mysql.source.utils.ObjectUtils; import org.apache.flink.cdc.connectors.mysql.utils.OptionUtils; import org.apache.flink.cdc.debezium.table.DebeziumOptions; @@ -52,6 +54,7 @@ /** Factory for creating configured instance of {@link MySqlTableSource}. */ public class MySqlTableSourceFactory implements DynamicTableSourceFactory { + private static final Logger LOGGER = LoggerFactory.getLogger(MySqlTableSourceFactory.class); private static final String IDENTIFIER = "mysql-cdc"; @@ -90,6 +93,8 @@ public DynamicTableSource createDynamicTableSource(Context context) { config.get(MySqlSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND); boolean scanNewlyAddedTableEnabled = config.get(MySqlSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED); + String dialect = config.get(MySqlSourceOptions.SCAN_DIALECT); + validateDialect(dialect); Duration heartbeatInterval = config.get(MySqlSourceOptions.HEARTBEAT_INTERVAL); String chunkKeyColumn = config.getOptional(MySqlSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN) @@ -110,6 +115,8 @@ public DynamicTableSource createDynamicTableSource(Context context) { boolean appendOnly = config.get(MySqlSourceOptions.SCAN_READ_CHANGELOG_AS_APPEND_ONLY_ENABLED); + validateDialectRuntimeSupport(dialect, enableParallelRead); + if (enableParallelRead) { validatePrimaryKeyIfEnableParallel(physicalSchema, chunkKeyColumn); validateIntegerOption( @@ -156,7 +163,8 @@ public DynamicTableSource createDynamicTableSource(Context context) { parseOnLineSchemaChanges, useLegacyJsonFormat, assignUnboundedChunkFirst, - appendOnly); + appendOnly, + dialect); } @Override @@ -198,6 +206,7 @@ public Set> optionalOptions() { options.add(MySqlSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND); options.add(MySqlSourceOptions.CONNECT_MAX_RETRIES); options.add(MySqlSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED); + options.add(MySqlSourceOptions.SCAN_DIALECT); options.add(MySqlSourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED); options.add(MySqlSourceOptions.HEARTBEAT_INTERVAL); options.add(MySqlSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN); @@ -323,6 +332,31 @@ private String validateAndGetServerId(ReadableConfig configuration) { return serverIdValue; } + private static void validateDialect(String dialect) { + if (!GtidStrategies.isSupportedDialect(dialect)) { + throw new ValidationException( + String.format( + "Invalid value for option '%s'. Supported values are [mysql, mariadb, auto], but was: %s", + MySqlSourceOptions.SCAN_DIALECT.key(), dialect)); + } + } + + /** + * MariaDB dialect wires its GTID read/recovery path through the incremental-snapshot (parallel) + * source only. Fail fast when "dialect=mariadb" is combined with the legacy no-incremental + * reader instead of silently falling back to the MySQL GTID path, which would neither advance + * nor resume from MariaDB GTIDs. + */ + private static void validateDialectRuntimeSupport(String dialect, boolean enableParallelRead) { + if (MariaDbGtidStrategy.DIALECT.equalsIgnoreCase(dialect) && !enableParallelRead) { + throw new ValidationException( + String.format( + "Option '%s' = 'mariadb' requires '%s' = 'true'", + MySqlSourceOptions.SCAN_DIALECT.key(), + MySqlSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ENABLED.key())); + } + } + /** Checks the value of given integer option is valid. */ private void validateIntegerOption( ConfigOption option, int optionValue, int exclusiveMin) { diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSourceMariaDbGtidTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSourceMariaDbGtidTest.java new file mode 100644 index 00000000000..878efddd54a --- /dev/null +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSourceMariaDbGtidTest.java @@ -0,0 +1,67 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package io.debezium.connector.mysql; + +import com.github.shyiko.mysql.binlog.event.Event; +import com.github.shyiko.mysql.binlog.event.EventHeaderV4; +import com.github.shyiko.mysql.binlog.event.EventType; +import com.github.shyiko.mysql.binlog.event.MariadbGtidEventData; +import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.EventDataWrapper; +import org.junit.jupiter.api.Test; + +import static io.debezium.connector.mysql.MySqlStreamingChangeEventSource.mariadbGtidOf; +import static org.assertj.core.api.Assertions.assertThat; + +/** Unit Test for the MariaDB GTID resume/ingest {@link MySqlStreamingChangeEventSource} */ +class MySqlStreamingChangeEventSourceMariaDbGtidTest { + + /** + * The GTID must use the server id from the event header, not the payload. The shyiko 0.27.2 + * {@code MariadbGtidEventDataDeserializer} never sets the payload server id (always 0), so + * reading it would ckp "0-0-N" -dirty state that survives restarts. + */ + @Test + void mariadbGtidOfUsesHeaderServerIdNotBrokenPayloadServerId() { + EventHeaderV4 eventHeaderV4 = new EventHeaderV4(); + eventHeaderV4.setEventType(EventType.MARIADB_GTID); + eventHeaderV4.setServerId(7L); + + MariadbGtidEventData data = new MariadbGtidEventData(); + data.setDomainId(0); + data.setSequence(42L); + + // Reproduce the deserializer bug: the payload server id is never populated + assertThat(data.getServerId()).isEqualTo(0L); + + assertThat(mariadbGtidOf(new Event(eventHeaderV4, data))).isEqualTo("0-7-42"); + } + + @Test + void mariadbGtidOfUnwrapsEventDataWrapper() { + EventHeaderV4 eventHeaderV4 = new EventHeaderV4(); + eventHeaderV4.setEventType(EventType.MARIADB_GTID); + eventHeaderV4.setServerId(5L); + + MariadbGtidEventData data = new MariadbGtidEventData(); + data.setDomainId(1L); + data.setSequence(99L); + + Event event = new Event(eventHeaderV4, new EventDataWrapper(data, data)); + assertThat(mariadbGtidOf(event)).isEqualTo("1-5-99"); + } +} diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/DebeziumUtilsTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/DebeziumUtilsTest.java index eeae681ea7b..c07a9077c85 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/DebeziumUtilsTest.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/DebeziumUtilsTest.java @@ -21,6 +21,7 @@ import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory; import org.apache.flink.cdc.connectors.mysql.table.StartupOptions; +import com.github.shyiko.mysql.binlog.BinaryLogClient; import io.debezium.connector.mysql.MySqlConnection; import org.assertj.core.api.Assertions; import org.junit.jupiter.api.Test; @@ -65,6 +66,42 @@ void testCreateMySqlConnection() { connection2.connectionString()); } + @Test + void testCreateBinaryClientReturnsMariaDbClientForMariadbDialect() { + BinaryLogClient client = + DebeziumUtils.createBinaryClient( + getConfig(new Properties()).getDbzConfiguration(), "mariadb"); + Assertions.assertThat(client).isInstanceOf(MariaDBBinaryLogClient.class); + } + + @Test + void testCreateBinaryClientReturnsMariaDbClientForAutoDialect() { + BinaryLogClient client = + DebeziumUtils.createBinaryClient( + getConfig(new Properties()).getDbzConfiguration(), "auto"); + Assertions.assertThat(client).isInstanceOf(MariaDBBinaryLogClient.class); + } + + @Test + void testCreateBinaryClientIsCaseInsensitiveForMysql() { + BinaryLogClient client = + DebeziumUtils.createBinaryClient( + getConfig(new Properties()).getDbzConfiguration(), "MySQL"); + Assertions.assertThat(client) + .isExactlyInstanceOf(BinaryLogClient.class) + .isNotInstanceOf(MariaDBBinaryLogClient.class); + } + + @Test + void testCreateBinaryClientReturnStockClientForMysqlDialect() { + BinaryLogClient client = + DebeziumUtils.createBinaryClient( + getConfig(new Properties()).getDbzConfiguration(), "mysql"); + Assertions.assertThat(client) + .isExactlyInstanceOf(BinaryLogClient.class) + .isNotInstanceOf(MariaDBBinaryLogClient.class); + } + private MySqlSourceConfig getConfig(Properties jdbcProperties) { return new MySqlSourceConfigFactory() .startupOptions(StartupOptions.initial()) diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReaderTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReaderTest.java index e7fd2c4cbb9..6ecd57dd120 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReaderTest.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReaderTest.java @@ -27,6 +27,7 @@ import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig; import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory; import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset; +import org.apache.flink.cdc.connectors.mysql.source.offset.MysqlGtidStrategy; import org.apache.flink.cdc.connectors.mysql.source.split.FinishedSnapshotSplitInfo; import org.apache.flink.cdc.connectors.mysql.source.split.MySqlBinlogSplit; import org.apache.flink.cdc.connectors.mysql.source.split.MySqlSnapshotSplit; @@ -1668,7 +1669,8 @@ protected MySqlOffsetContext loadStartingOffsetState( : initializeEffectiveOffset( mySqlSplit.asBinlogSplit().getStartingOffset(), getConnection(), - getSourceConfig()); + getSourceConfig(), + MysqlGtidStrategy.DIALECT); LOG.info("Starting offset is initialized to {}", offset); diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MariaDbDialectITCase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MariaDbDialectITCase.java new file mode 100644 index 00000000000..3b78316d953 --- /dev/null +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MariaDbDialectITCase.java @@ -0,0 +1,242 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package org.apache.flink.cdc.connectors.mysql.source; + +import org.apache.flink.cdc.connectors.mysql.testutils.MariaDbContainer; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.TableResult; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.types.Row; +import org.apache.flink.util.CloseableIterator; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.output.Slf4jLogConsumer; +import org.testcontainers.lifecycle.Startables; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Iterator; +import java.util.List; +import java.util.Random; + +import static org.apache.flink.cdc.connectors.mysql.source.MySqlSourceTestBase.assertEqualsInAnyOrder; + +/** + * IT case for proving the {@code mysql-cdc} connector reads a real MariaDB server when {@code + * scan.dialect=mariadb}: the snapshot rows are captured, then binlog INSERT/UPDATE/DELETE changes + * are captured. + */ +public class MariaDbDialectITCase { + + private static final Logger LOG = LoggerFactory.getLogger(MariaDbDialectITCase.class); + + private final StreamExecutionEnvironment env = + StreamExecutionEnvironment.getExecutionEnvironment(); + private final StreamTableEnvironment tEnv = + StreamTableEnvironment.create( + env, EnvironmentSettings.newInstance().inStreamingMode().build()); + + private MariaDbContainer mariaDbContainer; + + @BeforeEach + void setup() { + env.setParallelism(1); + env.enableCheckpointing(1000); + mariaDbContainer = new MariaDbContainer(); + mariaDbContainer.withLogConsumer(new Slf4jLogConsumer(LOG)); + LOG.info("Starting MariaDb Container"); + Startables.deepStart(mariaDbContainer).join(); + LOG.info("Started MariaDb Container"); + } + + @AfterEach + void tearDown() { + if (mariaDbContainer != null) { + mariaDbContainer.stop(); + } + } + + @Test + void testMariaDbSnapshotAndBinlog() throws Exception { + LOG.info("Testing MariaDb Snapshot and Binlog"); + initializeMariaDbTable(); + + String sourceDDL = + String.format( + "CREATE TABLE products (" + + "`id` INT NOT NULL," + + "name STRING," + + "wight DECIMAL(10,3)," + + "primary key (`id`) not enforced" + + ") with (" + + " 'connector' = 'mysql-cdc'," + + " 'scan.dialect' = 'mariadb'," + + " 'hostname' = '%s'," + + " 'port' = '%s'," + + " 'username' = '%s'," + + " 'password' = '%s'," + + " 'database-name' = '%s'," + + " 'table-name' = '%s'," + + " 'server-time-zone' = 'UTC'," + + " 'server-id' = '%s'" + + ")", + mariaDbContainer.getHost(), + mariaDbContainer.getDatabasePort(), + mariaDbContainer.getUsername(), + mariaDbContainer.getPassword(), + mariaDbContainer.getDatabaseName(), + "products", + getServerId()); + tEnv.executeSql(sourceDDL); + + TableResult result = tEnv.executeSql("SELECT `id`, `name`, `wight` FROM products"); + CloseableIterator collect = result.collect(); + + String[] expectSnapshot = { + "+I[101, scooter, 3.440]", "+I[102, car, 8.111]", "+I[103, hammer, 811.550]" + }; + + assertEqualsInAnyOrder( + Arrays.asList(expectSnapshot), fetchRows(collect, expectSnapshot.length)); + + try (Connection connection = getJdbcConnection(); + Statement statement = connection.createStatement()) { + statement.execute("UPDATE products SET `wight` = 5.100 WHERE `id` = 103;"); + statement.execute("INSERT INTO products values (104, 'jacket', 0.200);"); + statement.execute("DELETE FROM products WHERE `id` = 101;"); + } + + String[] expectedBinlog = { + "-U[103, hammer, 811.550]", + "+U[103, hammer, 5.100]", + "+I[104, jacket, 0.200]", + "-D[101, scooter, 3.440]" + }; + assertEqualsInAnyOrder( + Arrays.asList(expectedBinlog), fetchRows(collect, expectedBinlog.length)); + + result.getJobClient().get().cancel().get(); + } + + @Test + void testMariaDbTimestampStartup() throws Exception { + LOG.info("Testing MariaDb Timestamp start up"); + initializeMariaDbTable(); + + Thread.sleep(2000); + long startTimestampMillis = System.currentTimeMillis(); + Thread.sleep(2000); + + try (Connection connection = getJdbcConnection(); + Statement statement = connection.createStatement()) { + statement.execute("INSERT INTO products values (104, 'jacket', 0.200);"); + statement.execute("UPDATE products SET `wight` = 5.100 WHERE `id` = 103;"); + } + + String sourceDDL = + String.format( + "CREATE TABLE timestamp_products (" + + "`id` INT NOT NULL," + + "name STRING," + + "wight DECIMAL(10,3)," + + "primary key (`id`) not enforced" + + ") with (" + + " 'connector' = 'mysql-cdc'," + + " 'scan.dialect' = 'mariadb'," + + " 'scan.startup.mode' = 'timestamp'," + + " 'scan.startup.timestamp-millis' = '%s'," + + " 'hostname' = '%s'," + + " 'port' = '%s'," + + " 'username' = '%s'," + + " 'password' = '%s'," + + " 'database-name' = '%s'," + + " 'table-name' = '%s'," + + " 'server-time-zone' = 'UTC'," + + " 'server-id' = '%s'" + + ")", + startTimestampMillis, + mariaDbContainer.getHost(), + mariaDbContainer.getDatabasePort(), + mariaDbContainer.getUsername(), + mariaDbContainer.getPassword(), + mariaDbContainer.getDatabaseName(), + "products", + getServerId()); + tEnv.executeSql(sourceDDL); + + TableResult result = + tEnv.executeSql("SELECT `id`, `name`, `wight` FROM timestamp_products"); + CloseableIterator collect = result.collect(); + + String[] expectedTimestampRows = { + "+I[104, jacket, 0.200]", "-U[103, hammer, 811.550]", "+U[103, hammer, 5.100]" + }; + + assertEqualsInAnyOrder( + Arrays.asList(expectedTimestampRows), + fetchRows(collect, expectedTimestampRows.length)); + result.getJobClient().get().cancel().get(); + } + + private void initializeMariaDbTable() throws Exception { + try (Connection connection = getJdbcConnection(); + Statement statement = connection.createStatement()) { + statement.execute( + "CREATE TABLE products (" + + "id INT NOT NULL PRIMARY KEY, " + + "name VARCHAR(255), " + + "wight DECIMAL(10,3));"); + + statement.execute( + "INSERT INTO products VALUES (101, 'scooter', 3.44)," + + "(102, 'car', 8.111)," + + "(103, 'hammer', 811.55)"); + } + } + + private Connection getJdbcConnection() throws Exception { + return DriverManager.getConnection( + mariaDbContainer.getJdbcUrl(), + mariaDbContainer.getUsername(), + mariaDbContainer.getPassword()); + } + + private String getServerId() { + Random random = new Random(); + int serverId = random.nextInt(100) + 5400; + return serverId + "-" + (serverId + env.getParallelism()); + } + + private static List fetchRows(Iterator iter, int size) { + List rows = new ArrayList<>(); + while (size > 0 && iter.hasNext()) { + Row row = iter.next(); + rows.add(row.toString()); + size--; + } + return rows; + } +} diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MariaDbDialectSavepointITCase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MariaDbDialectSavepointITCase.java new file mode 100644 index 00000000000..ae58e58d805 --- /dev/null +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MariaDbDialectSavepointITCase.java @@ -0,0 +1,262 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package org.apache.flink.cdc.connectors.mysql.source; + +import org.apache.flink.api.common.eventtime.WatermarkStrategy; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.cdc.connectors.mysql.table.StartupOptions; +import org.apache.flink.cdc.connectors.mysql.testutils.MariaDbContainer; +import org.apache.flink.cdc.connectors.mysql.testutils.TestTable; +import org.apache.flink.cdc.connectors.mysql.testutils.TestTableSchemas; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.StateRecoveryOptions; +import org.apache.flink.core.execution.CheckpointingMode; +import org.apache.flink.core.execution.JobClient; +import org.apache.flink.core.execution.SavepointFormatType; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.datastream.DataStreamSource; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.api.operators.collect.CollectResultIterator; +import org.apache.flink.streaming.api.operators.collect.CollectResultIteratorAdapter; +import org.apache.flink.streaming.api.operators.collect.CollectSinkOperator; +import org.apache.flink.streaming.api.operators.collect.CollectSinkOperatorFactory; +import org.apache.flink.streaming.api.operators.collect.CollectStreamSink; +import org.apache.flink.table.data.RowData; +import org.apache.flink.test.junit5.MiniClusterExtension; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.output.Slf4jLogConsumer; +import org.testcontainers.lifecycle.Startables; + +import java.lang.reflect.Field; +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.Random; +import java.util.UUID; +import java.util.function.Function; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Integration Test for MariaDB restores from savepoint. The snapshot is taken while the source is + * in the binlog phase, a change is made after the savepoint, and the job is restored from the + * savepoint. Asserting that the post-savepoint UPDATE is captured exactly once proves the MariaDB + * GTID offset survives a savepoint round trip: the stored {@code @@gtid_binlog_pos} text is + * recovered as a MariaDB offset (stateless {@link + * org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset} + {@code + * GtidStrategies.detect}). The equivalent MySQL sp/retore path is covered by {@link + * SpecificStartingOffsetITCase} + */ +class MariaDbDialectSavepointITCase { + + private static final Logger LOG = LoggerFactory.getLogger(MariaDbDialectSavepointITCase.class); + + @RegisterExtension static MiniClusterExtension miniCluster = new MiniClusterExtension(); + + private MariaDbContainer mariaDbContainer; + private final TestTable customers = + new TestTable("test", "customers", TestTableSchemas.CUSTOMERS); + + @BeforeEach + void setup() throws Exception { + mariaDbContainer = new MariaDbContainer(); + mariaDbContainer.withLogConsumer(new Slf4jLogConsumer(LOG)); + LOG.info("Starting MariaDbDialectSavepointITCase..."); + Startables.deepStart(Stream.of(mariaDbContainer)).join(); + LOG.info("Started MariaDbDialectSavepointITCase."); + initializeCustomersTable(); + } + + @AfterEach + void tearDown() throws Exception { + if (mariaDbContainer != null) { + mariaDbContainer.stop(); + } + } + + @Test + void testSavepointAndRestoreAcrossBinlog() throws Exception { + StreamExecutionEnvironment env = getExecutionEnvironment(); + MySqlSource source = + getSourceBuilder().startupOptions(StartupOptions.initial()).build(); + DataStreamSource stream = + env.fromSource(source, WatermarkStrategy.noWatermarks(), "mariadb-savepoint-test"); + CollectResultIterator iterator = addCollector(env, stream); + + StreamExecutionEnvironment restoreEnv = getExecutionEnvironment(); + // duplicateTransformations + env.getTransformations().forEach(restoreEnv::addOperator); + + JobClient jobClient = env.executeAsync(); + iterator.setJobClient(jobClient); + List snapshotRows = fetchRowData(iterator, 3, customers::stringify); + assertThat(snapshotRows) + .containsExactlyInAnyOrder( + "+I[15213, Alice, Rome, 123456987]", + "+I[15513, Bob, Milan, 123456987]", + "+I[18213, Adrian, Shenzhen, 123456987]"); + + // Take a sp while the source is in the binlog phase. The sp stores a + // MariaDB "domain-server-sequence" GTID offset (read from @@gtid_binlog_pos) + Path spDir = Files.createTempDirectory("mariadb-savepoint-test"); + String spPath = + jobClient + .stopWithSavepoint( + false, + spDir.toAbsolutePath().toString(), + SavepointFormatType.DEFAULT) + .get(); + jobClient.cancel(); + + executeStatement("UPDATE customers SET name = 'Alicia' WHERE id = 15213"); + + // Restore from the savepoint and check the post-savepoint change is captured exactly-once + setupSavepoint(restoreEnv, spPath); + JobClient restoredJobClient = restoreEnv.executeAsync(); + iterator.setJobClient(restoredJobClient); + List rowsAfterRestore = fetchRowData(iterator, 2, customers::stringify); + assertThat(rowsAfterRestore) + .containsExactly( + "-U[15213, Alice, Rome, 123456987]", "+U[15213, Alicia, Rome, 123456987]"); + + restoredJobClient.cancel(); + } + + private MySqlSourceBuilder getSourceBuilder() { + return MySqlSource.builder() + .hostname(mariaDbContainer.getHost()) + .port(mariaDbContainer.getDatabasePort()) + .username(mariaDbContainer.getUsername()) + .password(mariaDbContainer.getPassword()) + .databaseList(mariaDbContainer.getDatabaseName()) + .tableList(customers.getTableId()) + .dialect("mariadb") + .serverTimeZone("UTC") + .serverId(getServerId()) + .deserializer(customers.getDeserializer()); + } + + private void initializeCustomersTable() throws Exception { + try (Connection connection = getJdbcConnection(); + Statement statement = connection.createStatement()) { + statement.execute( + "CREATE TABLE customers (" + + "id BIGINT NOT NULL PRIMARY KEY, " + + "name VARCHAR(255)," + + "address VARCHAR(255)," + + "phone_number VARCHAR(255));"); + statement.execute( + "INSERT INTO customers VALUES " + + "(15213, 'Alice', 'Rome', '123456987')," + + "(15513, 'Bob', 'Milan', '123456987')," + + "(18213, 'Adrian', 'Shenzhen', '123456987');"); + } + } + + private void executeStatement(String... statements) throws Exception { + try (Connection connection = getJdbcConnection(); + Statement statement = connection.createStatement()) { + for (String sql : statements) { + statement.execute(sql); + } + } + } + + private Connection getJdbcConnection() throws Exception { + return DriverManager.getConnection( + mariaDbContainer.getJdbcUrl(), + mariaDbContainer.getUsername(), + mariaDbContainer.getPassword()); + } + + private String getServerId() { + Random random = new Random(); + int serverId = random.nextInt(100) + 5400; + return serverId + "-" + (serverId + 2); + } + + private CollectResultIterator addCollector( + StreamExecutionEnvironment env, DataStream stream) { + TypeSerializer serializer = + stream.getTransformation() + .getOutputType() + .createSerializer(env.getConfig().getSerializerConfig()); + String accumulatorName = "dataStreamCollector_" + UUID.randomUUID(); + CollectSinkOperatorFactory factory = + new CollectSinkOperatorFactory<>(serializer, accumulatorName); + CollectStreamSink sink = new CollectStreamSink<>(stream, factory); + // Set both name and uid to the same value. The uid is used by Flink to generate + // OperatorId via StreamGraphHasherV2.generateUserSpecifiedHash(uid), and the same + // uid string must be passed to CollectResultIteratorAdapter for coordinator lookup + String operatorUid = "Data stream collect sink"; + sink.name(operatorUid).uid(operatorUid); + CollectSinkOperator operator = (CollectSinkOperator) factory.getOperator(); + env.addOperator(sink.getTransformation()); + return new CollectResultIteratorAdapter<>( + operatorUid, + operator, + serializer, + accumulatorName, + env.getCheckpointConfig(), + 10000L); + } + + private List fetchRowData( + Iterator iter, int size, Function stringFunction) { + List rows = new ArrayList<>(size); + while (size > 0 && iter.hasNext()) { + rows.add(iter.next()); + size--; + } + return rows.stream().map(stringFunction).collect(Collectors.toList()); + } + + private void setupSavepoint(StreamExecutionEnvironment env, String savepointPath) + throws Exception { + ClassLoader classLoader = Thread.currentThread().getContextClassLoader(); + Class clazz = + classLoader.loadClass( + "org.apache.flink.streaming.api.environment.StreamExecutionEnvironment"); + Field field = clazz.getDeclaredField("configuration"); + field.setAccessible(true); + Configuration configuration = (Configuration) field.get(env); + configuration.set(StateRecoveryOptions.SAVEPOINT_PATH, savepointPath); + } + + private StreamExecutionEnvironment getExecutionEnvironment() { + StreamExecutionEnvironment env = + StreamExecutionEnvironment.getExecutionEnvironment() + .setParallelism(1) + .enableCheckpointing(100); + env.getCheckpointConfig().setCheckpointingConsistencyMode(CheckpointingMode.EXACTLY_ONCE); + return env; + } +} diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/offset/BinlogOffsetSerializerTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/offset/BinlogOffsetSerializerTest.java new file mode 100644 index 00000000000..d59c850c8de --- /dev/null +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/offset/BinlogOffsetSerializerTest.java @@ -0,0 +1,104 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package org.apache.flink.cdc.connectors.mysql.source.offset; + +import org.junit.jupiter.api.Test; + +import java.nio.charset.StandardCharsets; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Unit Test for {@link BinlogOffsetSerializer} state-recovery compatibility: a checkpoint written + * by the Pre-MariaDB-GTID-support version, must still deserialize losslessly with the current code, + * and the recovered MySQL GTID offset must route and order identically. The MariaDB-GTID-support PR + * add no field to {@link BinlogOffset} and left the serializer untouched, so existing MySQL Jobs + * restore from old savepoint with zero migration. + */ +class BinlogOffsetSerializerTest { + + // + private static final String LEGACY_MYSQL_GTID_JSON = + "{\n" + + " \"file\" : \"mysql-bin.000003\",\n" + + " \"pos\" : \"4567\",\n" + + " \"event\" : \"2\",\n" + + " \"row\" : \"1\",\n" + + " \"gtids\" : \"24bc7850-2c16-11e6-a073-0242ac110002:1-1116\",\n" + + " \"server_id\" : \"223344\",\n" + + " \"ts_sec\": \"170000000\"\n" + + "}"; + private static final String MYSQL_UUID = "24bc7850-2c16-11e6-a073-0242ac110002"; + + @Test + void deserializesLegacyGtidOffset() throws Exception { + BinlogOffset offset = + BinlogOffsetSerializer.INSTANCE.deserialize( + LEGACY_MYSQL_GTID_JSON.getBytes(StandardCharsets.UTF_8)); + + assertThat(offset.getFilename()).isEqualTo("mysql-bin.000003"); + assertThat(offset.getPosition()).isEqualTo(4567L); + assertThat(offset.getRestartSkipEvents()).isEqualTo(2L); + assertThat(offset.getRestartSkipRows()).isEqualTo(1); + assertThat(offset.getGtidSet()).isEqualTo(MYSQL_UUID + ":1-1116"); + assertThat(offset.getServerId()).isEqualTo(223344L); + assertThat(offset.getTimestampSec()).isEqualTo(170000000L); + + // The offset stays stateless after restore: no dialect marker was ever stored, + // and the refactor did not add one + assertThat(offset.getOffset()).doesNotContainKey("dialect"); + } + + @Test + void roundTripsThroughCurrentSerializer() throws Exception { + BinlogOffsetSerializer serializer = BinlogOffsetSerializer.INSTANCE; + + BinlogOffset original = + serializer.deserialize(LEGACY_MYSQL_GTID_JSON.getBytes(StandardCharsets.UTF_8)); + + byte[] reSerialized = serializer.serialize(original); + BinlogOffset restoredOffset = serializer.deserialize(reSerialized); + + assertThat(restoredOffset).isEqualTo(original); + assertThat(restoredOffset.getOffset()).isEqualTo(original.getOffset()); + } + + @Test + void recoveredMysqlGtidOffsetRoutesToMysqlStrategy() throws Exception { + BinlogOffsetSerializer serializer = BinlogOffsetSerializer.INSTANCE; + BinlogOffset offset = + serializer.deserialize(LEGACY_MYSQL_GTID_JSON.getBytes(StandardCharsets.UTF_8)); + + assertThat(GtidStrategies.detect(offset.getGtidSet())) + .isInstanceOf(MysqlGtidStrategy.class); + } + + @Test + void compareToOrderingUnchangedForRecoveredMysqlGtids() { + // Drive the actual recovery comparison path (BinlogOffset.compareTo -> + // GtidStrategies.detect -> MySqlGtidStrategy) end to end, not just the strategy in + // isolation, to prove offset seeking keeps the pre-refactor ordering for MySQL GTIDs. + BinlogOffset earlier = BinlogOffset.ofGtidSet(MYSQL_UUID + ":1-1116"); + BinlogOffset later = BinlogOffset.ofGtidSet(MYSQL_UUID + ":1-2000"); + + assertThat(earlier.isBefore(later)).isTrue(); + assertThat(later.isAfter(earlier)).isTrue(); + assertThat(earlier.isAtOrBefore(earlier)).isTrue(); + assertThat(earlier.compareTo(earlier)).isZero(); + } +} diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/offset/GtidStrategiesTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/offset/GtidStrategiesTest.java new file mode 100644 index 00000000000..462c144e99a --- /dev/null +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/offset/GtidStrategiesTest.java @@ -0,0 +1,63 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package org.apache.flink.cdc.connectors.mysql.source.offset; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Unit tests for {@link GtidStrategies}. */ +class GtidStrategiesTest { + + private static final String MARIADB_VERSION = "5.5.5-10.6.12-MariaDB-1:10.6.12+maria~ubu2005"; + private static final String MYSQL_VERSION = "8.0.32"; + + @Test + void autoConstantIsAuto() { + assertThat(GtidStrategies.AUTO).isEqualTo("auto"); + } + + @Test + void pinnedDialectAlwaysWinsRegardlessOfVersion() { + assertThat(GtidStrategies.resolveDialect("mysql", MARIADB_VERSION)).isEqualTo("mysql"); + assertThat(GtidStrategies.resolveDialect("MySQL", MYSQL_VERSION)).isEqualTo("mysql"); + + assertThat(GtidStrategies.resolveDialect("MariaDB", MARIADB_VERSION)).isEqualTo("mariadb"); + assertThat(GtidStrategies.resolveDialect("mariadb", MYSQL_VERSION)).isEqualTo("mariadb"); + } + + @Test + void autoDetectVersionBanner() { + assertThat(GtidStrategies.resolveDialect("auto", MARIADB_VERSION)).isEqualTo("mariadb"); + assertThat(GtidStrategies.resolveDialect("auto", "10.11.2-MARIADB")).isEqualTo("mariadb"); + assertThat(GtidStrategies.resolveDialect("auto", MYSQL_VERSION)).isEqualTo("mysql"); + assertThat(GtidStrategies.resolveDialect("AUTO", "5.7.41-log")).isEqualTo("mysql"); + } + + @Test + void absentOrUnknownDefaultToMysql() { + assertThat(GtidStrategies.resolveDialect("auto", null)).isEqualTo("mysql"); + assertThat(GtidStrategies.resolveDialect("auto", "")).isEqualTo("mysql"); + + // null/unknown configured behaves like auto + assertThat(GtidStrategies.resolveDialect(null, MYSQL_VERSION)).isEqualTo("mysql"); + assertThat(GtidStrategies.resolveDialect(null, MARIADB_VERSION)).isEqualTo("mariadb"); + assertThat(GtidStrategies.resolveDialect("postgres", MARIADB_VERSION)).isEqualTo("mariadb"); + assertThat(GtidStrategies.resolveDialect(null, null)).isEqualTo("mysql"); + } +} diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/offset/MariaDbGtidStrategyTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/offset/MariaDbGtidStrategyTest.java new file mode 100644 index 00000000000..1de62ff5864 --- /dev/null +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/offset/MariaDbGtidStrategyTest.java @@ -0,0 +1,95 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package org.apache.flink.cdc.connectors.mysql.source.offset; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Unit tests for {@link MariaDbGtidStrategy}. MariaDB GTIDs are "domain-server-sequence" and the + * sequence advances per domain across server, so comparison must key on the domain and ignore the + * server id - otherwise a post-failover event (same domain, new server id) looks like a different + * stream and the offset is mishandled (Debezium dbz#1672, Flink CDC issue #2929) + */ +class MariaDbGtidStrategyTest { + + private final MariaDbGtidStrategy strategy = new MariaDbGtidStrategy(); + + @Test + void dialectIsMariadb() { + assertThat(strategy.dialect()).isEqualTo("mariadb"); + assertThat(GtidStrategies.of("mariadb")).isInstanceOf(MariaDbGtidStrategy.class); + assertThat(GtidStrategies.of("Mariadb")).isInstanceOf(MariaDbGtidStrategy.class); + } + + @Test + void canParseMariaDbGtid() { + assertThat(strategy.canParse("0-1-100")).isTrue(); + assertThat(strategy.canParse("0-1-100,1-2-50")).isTrue(); + // MySQL uuid:interval form MUST NOT be claimed by the MariaDB parser. + assertThat(strategy.canParse("A:1-100")).isFalse(); + assertThat(strategy.canParse(null)).isFalse(); + } + + @Test + void detectRoutesByGtidText() { + assertThat(GtidStrategies.detect("0-1-100")).isInstanceOf(MariaDbGtidStrategy.class); + assertThat(GtidStrategies.detect("0-1-100,1-2-50")).isInstanceOf(MariaDbGtidStrategy.class); + // uuid:interval text stays on the MySQL strategy + assertThat(GtidStrategies.detect("A:1-100")).isInstanceOf(MysqlGtidStrategy.class); + } + + @Test + void isEqualIgnoresServerId() { + // Same domain + same sequence, different server id (failover) -> equal + assertThat(strategy.isEqual("0-1-100", "0-2-100")).isTrue(); + + // different sequence + assertThat(strategy.isEqual("0-1-100", "0-1-101")).isFalse(); + // different domain + assertThat(strategy.isEqual("0-1-100", "1-1-100")).isFalse(); + } + + @Test + void isEqualHandlesMultipleDomainsOrderIndependently() { + assertThat(strategy.isEqual("0-1-100,1-2-50", "1-9-50,0-7-100")).isTrue(); + assertThat(strategy.isEqual("0-1-100,1-2-50", "0-1-100")).isFalse(); + } + + @Test + void isContainedWithinKeysOnDomainAndIgnoresServerId() { + // Failover: sub at seq 100 on server 1, sup advanced to seq 102 on server 2 -> contained + assertThat(strategy.isContainedWithin("0-1-100", "0-2-102")).isTrue(); + assertThat(strategy.isContainedWithin("0-1-100", "0-2-100")).isTrue(); + + // sub ahead of sup -> not contained. + assertThat(strategy.isContainedWithin("0-1-101", "0-2-100")).isFalse(); + // domain missing + assertThat(strategy.isContainedWithin("0-1-100,1-1-1", "0-2-100")).isFalse(); + // multi-domain + assertThat(strategy.isContainedWithin("0-1-100,1-1-50", "0-2-100,1-9-60")).isTrue(); + } + + @Test + void emptyGtidSets() { + assertThat(strategy.isEqual("", "")).isTrue(); + assertThat(strategy.isContainedWithin("", "0-1-100")).isTrue(); + assertThat(strategy.isContainedWithin("0-1-100", "")).isFalse(); + } +} diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/offset/MysqlGtidStrategyTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/offset/MysqlGtidStrategyTest.java new file mode 100644 index 00000000000..5fff9068472 --- /dev/null +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/offset/MysqlGtidStrategyTest.java @@ -0,0 +1,130 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package org.apache.flink.cdc.connectors.mysql.source.offset; + +import io.debezium.connector.mysql.GtidSet; +import org.junit.jupiter.api.Test; + +import java.util.Random; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Unit Test for {@link MysqlGtidStrategy}. This strategy MUST BE equivalent to the inline {@code + * new GtidSet.equals()/isContanedWithin()} logic that {@link BinlogOffset} used before the {@link + * GtidStrategy} abstraction was extracted. + */ +class MysqlGtidStrategyTest { + + private final MysqlGtidStrategy strategy = new MysqlGtidStrategy(); + + // old (reference) impl: exactly what BinlogOffset.compareTo() did + private static boolean oldIsEqual(String a, String b) { + return new GtidSet(a).equals(new GtidSet(b)); + } + + private static boolean oldIsContainedWithin(String sub, String sup) { + return new GtidSet(sub).isContainedWithin(new GtidSet(sup)); + } + + @Test + void dialectIsMysql() { + assertThat(strategy.dialect()).isEqualTo("mysql"); + assertThat(GtidStrategies.of("mysql")).isInstanceOf(MysqlGtidStrategy.class); + assertThat(GtidStrategies.of(null)).isInstanceOf(MysqlGtidStrategy.class); + } + + @Test + void detectReturnsMysqlForUuidInterval() { + assertThat(GtidStrategies.detect("A:1-100")).isInstanceOf(MysqlGtidStrategy.class); + } + + @Test + void canParseMysqlGtidText() { + assertThat(strategy.canParse("A:1-100")).isTrue(); + assertThat(strategy.canParse("a:1-100:102-200,B:20-200")).isTrue(); + assertThat(strategy.canParse("")).isTrue(); + assertThat(strategy.canParse(null)).isFalse(); + } + + @Test + void isEqualMatchesOldImpl() { + String[][] pairs = { + {"A:1-100", "A:1-100"}, + {"A:1-100", "A:1-50"}, + {"A:1-100", "B:1-100"}, + {"A:1-100:102-200,B:20-200", "A:1-100:102-200,B:20-200"}, + {"A:1-100,B:1-10", "B:1-10,A:1-100"}, + {"A:1-100", "A:1-100,C:1-5"} + }; + for (String[] pair : pairs) { + assertThat(strategy.isEqual(pair[0], pair[1])) + .as("isEqual(%s,%s)", pair[0], pair[1]) + .isEqualTo(oldIsEqual(pair[0], pair[1])); + } + } + + @Test + void isContainedWithinMatchesOldImpl() { + String[][] pairs = { + {"A:1-50", "A:1-100"}, + {"A:1-100", "A:1-100"}, + {"A:1-100", "A:1-50"}, + {"A:1-100", "A:1-100:102-200,B:20-200"}, + {"A:1-100,C:1-5", "A:1-100"}, + {"B:1-10", "A:1-100,B:1-10"}, + }; + for (String[] pair : pairs) { + assertThat(strategy.isContainedWithin(pair[0], pair[1])) + .as("isContainedWithin(%s,%s)", pair[0], pair[1]) + .isEqualTo(oldIsContainedWithin(pair[0], pair[1])); + } + } + + @Test + void fuzzCrossValidationAgainstOldImpl() { + Random random = new Random(42); + for (int i = 0; i < 2000; i++) { + String a = randomGtid(random); + String b = randomGtid(random); + + assertThat(strategy.isEqual(a, b)) + .as("isEqual(%s,%s)", a, b) + .isEqualTo(oldIsEqual(a, b)); + assertThat(strategy.isContainedWithin(a, b)) + .as("isContainedWithin(%s,%s)", a, b) + .isEqualTo(oldIsContainedWithin(a, b)); + } + } + + private static String randomGtid(Random random) { + // Build an uuid:interval set over a small server + char[] servers = {'A', 'B', 'C'}; + int count = 1 + random.nextInt(servers.length); + StringBuilder sb = new StringBuilder(); + for (int i = 0; i < count; i++) { + if (i > 0) { + sb.append(","); + } + int start = 1 + random.nextInt(50); + int end = start + random.nextInt(50); + sb.append(servers[i]).append(":").append(start).append("-").append(end); + } + return sb.toString(); + } +} diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactoryTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactoryTest.java index 01c2dff84da..50741cd0343 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactoryTest.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactoryTest.java @@ -32,7 +32,6 @@ import org.apache.flink.table.factories.Factory; import org.apache.flink.table.factories.FactoryUtil; -import org.assertj.core.api.Assertions; import org.junit.jupiter.api.Test; import java.time.Duration; @@ -58,6 +57,8 @@ import static org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST; import static org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceOptions.SCAN_SNAPSHOT_FETCH_SIZE; import static org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceOptions.USE_LEGACY_JSON_FORMAT; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Test for {@link MySqlTableSource} created by {@link MySqlTableSourceFactory}. */ class MySqlTableSourceFactoryTest { @@ -129,8 +130,9 @@ void testCommonProperties() { PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(), USE_LEGACY_JSON_FORMAT.defaultValue(), SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST.defaultValue(), - false); - Assertions.assertThat(actualSource).isEqualTo(expectedSource); + false, + "auto"); + assertThat(actualSource).isEqualTo(expectedSource); } @Test @@ -179,8 +181,9 @@ void testEnableParallelReadSource() { PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(), USE_LEGACY_JSON_FORMAT.defaultValue(), SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST.defaultValue(), - false); - Assertions.assertThat(actualSource).isEqualTo(expectedSource); + false, + "auto"); + assertThat(actualSource).isEqualTo(expectedSource); } @Test @@ -225,8 +228,9 @@ void testEnableParallelReadSourceWithSingleServerId() { PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(), USE_LEGACY_JSON_FORMAT.defaultValue(), SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST.defaultValue(), - false); - Assertions.assertThat(actualSource).isEqualTo(expectedSource); + false, + "auto"); + assertThat(actualSource).isEqualTo(expectedSource); } @Test @@ -269,8 +273,9 @@ void testEnableParallelReadSourceLatestOffset() { PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(), USE_LEGACY_JSON_FORMAT.defaultValue(), SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST.defaultValue(), - false); - Assertions.assertThat(actualSource).isEqualTo(expectedSource); + false, + "auto"); + assertThat(actualSource).isEqualTo(expectedSource); } @Test @@ -330,14 +335,13 @@ void testOptionalProperties() { PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(), true, SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST.defaultValue(), - false); - Assertions.assertThat(actualSource) - .isEqualTo(expectedSource) - .isInstanceOf(MySqlTableSource.class); + false, + "auto"); + assertThat(actualSource).isEqualTo(expectedSource).isInstanceOf(MySqlTableSource.class); MySqlTableSource actualMySqlTableSource = (MySqlTableSource) actualSource; Properties parellelProperties = new Properties(); parellelProperties.put("test", "test"); - Assertions.assertThat( + assertThat( actualMySqlTableSource.getParallelDbzProperties( actualMySqlTableSource.getDbzProperties())) .isEqualTo(parellelProperties); @@ -389,8 +393,9 @@ void testStartupFromSpecificOffset() { PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(), USE_LEGACY_JSON_FORMAT.defaultValue(), SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST.defaultValue(), - false); - Assertions.assertThat(actualSource).isEqualTo(expectedSource); + false, + "auto"); + assertThat(actualSource).isEqualTo(expectedSource); } @Test @@ -431,8 +436,9 @@ void testStartupFromInitial() { PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(), USE_LEGACY_JSON_FORMAT.defaultValue(), SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST.defaultValue(), - false); - Assertions.assertThat(actualSource).isEqualTo(expectedSource); + false, + "auto"); + assertThat(actualSource).isEqualTo(expectedSource); } @Test @@ -474,8 +480,9 @@ void testStartupFromEarliestOffset() { PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(), USE_LEGACY_JSON_FORMAT.defaultValue(), SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST.defaultValue(), - false); - Assertions.assertThat(actualSource).isEqualTo(expectedSource); + false, + "auto"); + assertThat(actualSource).isEqualTo(expectedSource); } @Test @@ -518,8 +525,9 @@ void testStartupFromSpecificTimestamp() { PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(), USE_LEGACY_JSON_FORMAT.defaultValue(), SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST.defaultValue(), - false); - Assertions.assertThat(actualSource).isEqualTo(expectedSource); + false, + "auto"); + assertThat(actualSource).isEqualTo(expectedSource); } @Test @@ -560,8 +568,9 @@ void testStartupFromLatestOffset() { PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(), USE_LEGACY_JSON_FORMAT.defaultValue(), SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST.defaultValue(), - false); - Assertions.assertThat(actualSource).isEqualTo(expectedSource); + false, + "auto"); + assertThat(actualSource).isEqualTo(expectedSource); } @Test @@ -607,17 +616,81 @@ void testMetadataColumns() { PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(), USE_LEGACY_JSON_FORMAT.defaultValue(), SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST.defaultValue(), - false); + false, + "auto"); expectedSource.producedDataType = SCHEMA_WITH_METADATA.toSourceRowDataType(); expectedSource.metadataKeys = Arrays.asList("op_ts", "database_name"); - Assertions.assertThat(actualSource).isEqualTo(expectedSource); + assertThat(actualSource).isEqualTo(expectedSource); + } + + @Test + void testDialectOptions() { + Map properties = getAllOptions(); + // The MariaDB dialect is only supported by the incremental-snapshot source, so the + // dialect option is exercised on that (supported) path. + properties.put("scan.incremental.snapshot.enabled", "true"); + properties.put("scan.dialect", "mariadb"); + + DynamicTableSource actualSource = createTableSource(properties); + MySqlTableSource expectedSource = + new MySqlTableSource( + SCHEMA, + 3306, + MY_LOCALHOST, + MY_DATABASE, + MY_TABLE, + MY_USERNAME, + MY_PASSWORD, + ZoneId.systemDefault(), + PROPERTIES, + null, + true, + SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE.defaultValue(), + CHUNK_META_GROUP_SIZE.defaultValue(), + SCAN_SNAPSHOT_FETCH_SIZE.defaultValue(), + CONNECT_TIMEOUT.defaultValue(), + CONNECT_MAX_RETRIES.defaultValue(), + CONNECTION_POOL_SIZE.defaultValue(), + CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND.defaultValue(), + CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND.defaultValue(), + StartupOptions.initial(), + false, + false, + new Properties(), + HEARTBEAT_INTERVAL.defaultValue(), + null, + SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(), + PARSE_ONLINE_SCHEMA_CHANGES.defaultValue(), + USE_LEGACY_JSON_FORMAT.defaultValue(), + SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST.defaultValue(), + false, + "mariadb"); + assertThat(actualSource).isEqualTo(expectedSource); + } + + @Test + void testDialectMariadbRejectedWithoutIncrementalSnapshot() { + Map properties = getAllOptions(); + properties.put("scan.dialect", "mariadb"); + assertThatThrownBy(() -> createTableSource(properties)) + .hasStackTraceContaining( + "Option 'scan.dialect' = 'mariadb' requires 'scan.incremental.snapshot.enabled' = 'true'"); + } + + @Test + void testInvalidDialect() { + Map properties = getAllOptions(); + properties.put("scan.dialect", "postgres"); + assertThatThrownBy(() -> createTableSource(properties)) + .hasRootCauseMessage( + "Invalid value for option 'scan.dialect'. Supported values are [mysql, mariadb, auto], but was: postgres"); } @Test void testValidation() { // validate illegal port - Assertions.assertThatThrownBy( + assertThatThrownBy( () -> { Map properties = getAllOptions(); properties.put("port", "123b"); @@ -626,7 +699,7 @@ void testValidation() { .hasStackTraceContaining("Could not parse value '123b' for key 'port'."); // validate illegal server id - Assertions.assertThatThrownBy( + assertThatThrownBy( () -> { Map properties = getAllOptions(); properties.put("server-id", "123b"); @@ -636,7 +709,7 @@ void testValidation() { .hasStackTraceContaining("The server id 123b is not a valid numeric."); // validate illegal connect.timeout - Assertions.assertThatThrownBy( + assertThatThrownBy( () -> { Map properties = getAllOptions(); properties.put("scan.incremental.snapshot.enabled", "true"); @@ -647,7 +720,7 @@ void testValidation() { "The value of option 'connect.timeout' cannot be less than PT0.25S, but actual is PT0.24S"); // validate illegal split size - Assertions.assertThatThrownBy( + assertThatThrownBy( () -> { Map properties = getAllOptions(); properties.put("scan.incremental.snapshot.enabled", "true"); @@ -660,7 +733,7 @@ void testValidation() { // validate illegal fetch size - Assertions.assertThatThrownBy( + assertThatThrownBy( () -> { Map properties = getAllOptions(); properties.put("scan.incremental.snapshot.enabled", "true"); @@ -671,7 +744,7 @@ void testValidation() { "The value of option 'scan.snapshot.fetch.size' must larger than 1, but is 1"); // validate illegal split meta group size - Assertions.assertThatThrownBy( + assertThatThrownBy( () -> { Map properties = getAllOptions(); properties.put("scan.incremental.snapshot.enabled", "true"); @@ -682,7 +755,7 @@ void testValidation() { "The value of option 'chunk-meta.group.size' must larger than 1, but is 1"); // validate illegal split meta group size - Assertions.assertThatThrownBy( + assertThatThrownBy( () -> { Map properties = getAllOptions(); properties.put("scan.incremental.snapshot.enabled", "true"); @@ -693,7 +766,7 @@ void testValidation() { "The value of option 'chunk-key.even-distribution.factor.upper-bound' must larger than or equals 1.0, but is 0.8"); // validate illegal connection pool size - Assertions.assertThatThrownBy( + assertThatThrownBy( () -> { Map properties = getAllOptions(); properties.put("scan.incremental.snapshot.enabled", "true"); @@ -704,7 +777,7 @@ void testValidation() { "The value of option 'connection.pool.size' must larger than 1, but is 1"); // validate illegal connect max retry times - Assertions.assertThatThrownBy( + assertThatThrownBy( () -> { Map properties = getAllOptions(); properties.put("scan.incremental.snapshot.enabled", "true"); @@ -719,13 +792,13 @@ void testValidation() { for (ConfigOption requiredOption : factory.requiredOptions()) { Map properties = getAllOptions(); properties.remove(requiredOption.key()); - Assertions.assertThatThrownBy(() -> createTableSource(properties)) + assertThatThrownBy(() -> createTableSource(properties)) .hasStackTraceContaining( "Missing required options are:\n\n" + requiredOption.key()); } // validate unsupported option - Assertions.assertThatThrownBy( + assertThatThrownBy( () -> { Map properties = getAllOptions(); properties.put("unknown", "abc"); @@ -734,7 +807,7 @@ void testValidation() { .hasStackTraceContaining("Unsupported options:\n\nunknown"); // validate unsupported option - Assertions.assertThatThrownBy( + assertThatThrownBy( () -> { Map properties = getAllOptions(); properties.put("scan.startup.mode", "abc"); @@ -746,7 +819,7 @@ void testValidation() { + "but was: abc"); // validate invalid database-name - Assertions.assertThatThrownBy( + assertThatThrownBy( () -> { Map properties = getAllOptions(); properties.put("database-name", "*_invalid_db"); @@ -758,7 +831,7 @@ void testValidation() { "*_invalid_db")); // validate invalid table-name - Assertions.assertThatThrownBy( + assertThatThrownBy( () -> { Map properties = getAllOptions(); properties.put("table-name", "*_invalid_table"); @@ -810,8 +883,9 @@ void testEnablingExperimentalOptions() { true, true, true, - false); - Assertions.assertThat(actualSource).isEqualTo(expectedSource); + false, + "auto"); + assertThat(actualSource).isEqualTo(expectedSource); } private Map getAllOptions() { diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/testutils/MariaDbContainer.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/testutils/MariaDbContainer.java new file mode 100644 index 00000000000..6b84c9c0d63 --- /dev/null +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/testutils/MariaDbContainer.java @@ -0,0 +1,113 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package org.apache.flink.cdc.connectors.mysql.testutils; + +import org.testcontainers.containers.JdbcDatabaseContainer; +import org.testcontainers.utility.DockerImageName; + +import java.util.HashSet; +import java.util.Set; + +/** + * Docker container for MariaDB, used by the MariaDB dialect integration tests. Binlog and ROW + * format are enabled through server command-line flags (rather than a mounted my.cnf) so the + * container is fully self-contained; MariaDB tracks {@code @@gtid_binlog_pos} automatically once + * the binary log is on. + */ +@SuppressWarnings("rawtypes") +public class MariaDbContainer extends JdbcDatabaseContainer { + + public static final String IMAGE = "mariadb"; + public static final String VERSION = "11.4"; + public static final int MARIADB_PORT = 3306; + + private final String databaseName = "test"; + private final String username = "root"; + private final String password = "test"; + + public MariaDbContainer() { + this(VERSION); + } + + public MariaDbContainer(String version) { + super(DockerImageName.parse(IMAGE + ":" + version)); + addExposedPort(MARIADB_PORT); + // Enable the binary log in ROW format; + setCommand("--log-bin=mysql-bin", "--binlog-format=ROW", "--server-id=223344"); + } + + @Override + protected Set getLivenessCheckPorts() { + return new HashSet<>(getMappedPort(MARIADB_PORT)); + } + + protected void configure() { + addEnv("MARIADB_DATABASE", databaseName); + addEnv("MARIADB_ROOT_PASSWORD", password); + setStartupAttempts(3); + } + + @Override + public String getDriverClassName() { + try { + Class.forName("com.mysql.cj.jdbc.Driver"); + return "com.mysql.cj.jdbc.Driver"; + } catch (ClassNotFoundException e) { + return "com.mysql.jdbc.Driver"; + } + } + + public String getJdbcUrl(String databaseName) { + return "jdbc:mysql://" + + getHost() + + ":" + + getDatabasePort() + + "/" + + databaseName + + "?useSSL=false&&allowPublicKeyRetrieval=true"; + } + + @Override + public String getJdbcUrl() { + return getJdbcUrl(databaseName); + } + + public int getDatabasePort() { + return getMappedPort(MARIADB_PORT); + } + + @Override + public String getDatabaseName() { + return databaseName; + } + + @Override + public String getUsername() { + return username; + } + + @Override + public String getPassword() { + return password; + } + + @Override + protected String getTestQueryString() { + return "SELECT 1"; + } +}