jprdonnelly/airbyte
1
0
Fork 0
mirror of synced 2025-12-25 02:09:19 -05:00
Code Issues Packages Projects Releases Wiki Activity

[source-mysql] Upgrade Debezium and mysql connector to 2.7.1.Final (#44013)

Browse Source
This commit is contained in:
Yue Li
2024-08-14 10:15:05 -07:00
committed by GitHub
parent a0eb14eff6
commit ed44c093b3
6 changed files with 28 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
airbyte-integrations/connectors/source-mysql/build.gradle
View File

@@ -24,8 +24,8 @@ application {
dependencies {
implementation 'mysql:mysql-connector-java:8.0.30'
implementation 'io.debezium:debezium-embedded:2.5.4.Final'
implementation 'io.debezium:debezium-connector-mysql:2.5.4.Final'
implementation 'io.debezium:debezium-embedded:2.7.1.Final'
implementation 'io.debezium:debezium-connector-mysql:2.7.1.Final'
testFixturesImplementation 'org.testcontainers:mysql:1.19.0'

2
airbyte-integrations/connectors/source-mysql/metadata.yaml
View File

@@ -9,7 +9,7 @@ data:
connectorSubtype: database
connectorType: source
definitionId: 435bb9a5-7887-4809-aa58-28c27df0d7ad
dockerImageTag: 3.6.9
dockerImageTag: 3.7.0
dockerRepository: airbyte/source-mysql
documentationUrl: https://docs.airbyte.com/integrations/sources/mysql
githubIssueLabel: source-mysql

2
airbyte-integrations/connectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/cdc/CustomMySQLTinyIntOneToBooleanConverter.java
View File

@@ -4,7 +4,7 @@
package io.airbyte.integrations.source.mysql.cdc;
import io.debezium.connector.mysql.converters.TinyIntOneToBooleanConverter;
import io.debezium.connector.binlog.converters.TinyIntOneToBooleanConverter;
import io.debezium.spi.converter.RelationalColumn;
import org.apache.kafka.connect.data.SchemaBuilder;

26
airbyte-integrations/connectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/cdc/MySqlDebeziumStateUtil.java
View File

@@ -23,12 +23,11 @@ import io.airbyte.commons.json.Jsons;
import io.airbyte.protocol.models.v0.ConfiguredAirbyteCatalog;
import io.debezium.config.Configuration;
import io.debezium.connector.common.OffsetReader;
import io.debezium.connector.mysql.GtidSet;
import io.debezium.connector.mysql.MySqlConnectorConfig;
import io.debezium.connector.mysql.MySqlOffsetContext;
import io.debezium.connector.mysql.MySqlOffsetContext.Loader;
import io.debezium.connector.mysql.MySqlPartition;
import io.debezium.connector.mysql.strategy.mysql.MySqlGtidSet;
import io.debezium.connector.mysql.gtid.MySqlGtidSet;
import io.debezium.engine.ChangeEvent;
import io.debezium.pipeline.spi.Offsets;
import io.debezium.pipeline.spi.Partition;
@@ -71,12 +70,12 @@ public class MySqlDebeziumStateUtil implements DebeziumStateUtil {
if (gtidSetFromSavedState.isContainedWithin(availableGtidSet)) {
LOGGER.info("MySQL server current GTID set {} does contain the GTID set required by the connector {}", availableGtidSet,
gtidSetFromSavedState);
final Optional<GtidSet> gtidSetToReplicate = subtractGtidSet(availableGtidSet, gtidSetFromSavedState, database);
final Optional<MySqlGtidSet> gtidSetToReplicate = subtractGtidSet(availableGtidSet, gtidSetFromSavedState, database);
if (gtidSetToReplicate.isPresent()) {
final Optional<GtidSet> purgedGtidSet = purgedGtidSet(database);
final Optional<MySqlGtidSet> purgedGtidSet = purgedGtidSet(database);
if (purgedGtidSet.isPresent()) {
LOGGER.info("MySQL server has already purged {} GTIDs", purgedGtidSet.get());
final Optional<GtidSet> nonPurgedGtidSetToReplicate = subtractGtidSet(gtidSetToReplicate.get(), purgedGtidSet.get(), database);
final Optional<MySqlGtidSet> nonPurgedGtidSetToReplicate = subtractGtidSet(gtidSetToReplicate.get(), purgedGtidSet.get(), database);
if (nonPurgedGtidSetToReplicate.isPresent()) {
LOGGER.info("GTIDs known by the MySQL server but not processed yet {}, for replication are available only {}", gtidSetToReplicate,
nonPurgedGtidSetToReplicate);
@@ -116,8 +115,8 @@ public class MySqlDebeziumStateUtil implements DebeziumStateUtil {
}
}
private Optional<GtidSet> subtractGtidSet(final GtidSet set1, final GtidSet set2, final JdbcDatabase database) {
try (final Stream<GtidSet> stream = database.unsafeResultSetQuery(
private Optional<MySqlGtidSet> subtractGtidSet(final MySqlGtidSet set1, final MySqlGtidSet set2, final JdbcDatabase database) {
try (final Stream<MySqlGtidSet> stream = database.unsafeResultSetQuery(
connection -> {
final PreparedStatement ps = connection.prepareStatement("SELECT GTID_SUBTRACT(?, ?)");
ps.setString(1, set1.toString());
@@ -125,7 +124,7 @@ public class MySqlDebeziumStateUtil implements DebeziumStateUtil {
return ps.executeQuery();
},
resultSet -> new MySqlGtidSet(resultSet.getString(1)))) {
final List<GtidSet> gtidSets = stream.toList();
final List<MySqlGtidSet> gtidSets = stream.toList();
if (gtidSets.isEmpty()) {
return Optional.empty();
} else if (gtidSets.size() == 1) {
@@ -138,8 +137,8 @@ public class MySqlDebeziumStateUtil implements DebeziumStateUtil {
}
}
private Optional<GtidSet> purgedGtidSet(final JdbcDatabase database) {
try (final Stream<Optional<GtidSet>> stream = database.unsafeResultSetQuery(
private Optional<MySqlGtidSet> purgedGtidSet(final JdbcDatabase database) {
try (final Stream<Optional<MySqlGtidSet>> stream = database.unsafeResultSetQuery(
connection -> connection.createStatement().executeQuery("SELECT @@global.gtid_purged"),
resultSet -> {
if (resultSet.getMetaData().getColumnCount() > 0) {
@@ -150,7 +149,7 @@ public class MySqlDebeziumStateUtil implements DebeziumStateUtil {
}
return Optional.empty();
})) {
final List<Optional<GtidSet>> gtidSet = stream.toList();
final List<Optional<MySqlGtidSet>> gtidSet = stream.toList();
if (gtidSet.isEmpty()) {
return Optional.empty();
} else if (gtidSet.size() == 1) {
@@ -237,9 +236,10 @@ public class MySqlDebeziumStateUtil implements DebeziumStateUtil {
final ConfiguredAirbyteCatalog catalog,
final JdbcDatabase database) {
// https://debezium.io/documentation/reference/2.2/connectors/mysql.html#mysql-property-snapshot-mode
// We use the schema_only_recovery property cause using this mode will instruct Debezium to
// We use the recovery property cause using this mode will instruct Debezium to
// construct the db schema history.
properties.setProperty("snapshot.mode", "schema_only_recovery");
// Note that we used to use schema_only_recovery mode, but this mode has been deprecated.
properties.setProperty("snapshot.mode", "recovery");
final String dbName = database.getSourceConfig().get(JdbcUtils.DATABASE_KEY).asText();
// Topic.prefix is sanitized version of database name. At this stage properties does not have this
// value - it's set in RelationalDbDebeziumPropertiesManager.

9
airbyte-integrations/connectors/source-mysql/src/test/java/io/airbyte/integrations/source/mysql/CdcMysqlSourceTest.java
View File

@@ -780,11 +780,10 @@ public class CdcMysqlSourceTest extends CdcSourceTest<MySqlSource, MySQLTestData
}
/**
* This test creates lots of tables increasing the schema history size above the limit of
* {@link AirbyteSchemaHistoryStorage#SIZE_LIMIT_TO_COMPRESS_MB} forcing the
* {@link AirbyteSchemaHistoryStorage#read()} method to compress the schema history blob as part of
* the state message which allows us to test that the next sync is able to work fine when provided
* with a compressed blob in the state.
* This test creates lots of tables increasing the schema history size above the limit of forcing
* the {@link AirbyteSchemaHistoryStorage#read()} method to compress the schema history blob as part
* of the state message which allows us to test that the next sync is able to work fine when
* provided with a compressed blob in the state.
*/
@Test
@Timeout(value = 120)