diff --git a/airbyte-cdk/java/airbyte-cdk/README.md b/airbyte-cdk/java/airbyte-cdk/README.md index 8a4b12af0de84..f7310f4c1b07c 100644 --- a/airbyte-cdk/java/airbyte-cdk/README.md +++ b/airbyte-cdk/java/airbyte-cdk/README.md @@ -166,6 +166,7 @@ MavenLocal debugging steps: | Version | Date | Pull Request | Subject | |:--------|:-----------|:-----------------------------------------------------------|:---------------------------------------------------------------------------------------------------------------------------------------------------------------| +| 0.20.3 | 2024-02-09 | [\#34580](https://github.com/airbytehq/airbyte/pull/34580) | Support special chars in mysql/mssql database name. | | 0.20.2 | 2024-02-12 | [\#35111](https://github.com/airbytehq/airbyte/pull/35144) | Make state emission from async framework synchronized. | | 0.20.1 | 2024-02-11 | [\#35111](https://github.com/airbytehq/airbyte/pull/35111) | Fix GlobalAsyncStateManager stats counting logic. | | 0.20.0 | 2024-02-09 | [\#34562](https://github.com/airbytehq/airbyte/pull/34562) | Add new test cases to BaseTypingDedupingTest to exercise special characters. | diff --git a/airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties b/airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties index c9e7db78c85e3..21b134115f083 100644 --- a/airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties +++ b/airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties @@ -1 +1 @@ -version=0.20.2 +version=0.20.3 \ No newline at end of file diff --git a/airbyte-cdk/java/airbyte-cdk/db-sources/src/main/java/io/airbyte/cdk/integrations/debezium/internals/DebeziumPropertiesManager.java b/airbyte-cdk/java/airbyte-cdk/db-sources/src/main/java/io/airbyte/cdk/integrations/debezium/internals/DebeziumPropertiesManager.java index 509a3e8982bb3..4bae69e9999bc 100644 --- a/airbyte-cdk/java/airbyte-cdk/db-sources/src/main/java/io/airbyte/cdk/integrations/debezium/internals/DebeziumPropertiesManager.java +++ b/airbyte-cdk/java/airbyte-cdk/db-sources/src/main/java/io/airbyte/cdk/integrations/debezium/internals/DebeziumPropertiesManager.java @@ -6,6 +6,7 @@ import com.fasterxml.jackson.databind.JsonNode; import io.airbyte.protocol.models.v0.ConfiguredAirbyteCatalog; +import io.debezium.spi.common.ReplacementFunction; import java.util.Optional; import java.util.Properties; @@ -76,8 +77,8 @@ public Properties getDebeziumProperties( props.setProperty("max.queue.size.in.bytes", BYTE_VALUE_256_MB); // WARNING : Never change the value of this otherwise all the connectors would start syncing from - // scratch - props.setProperty(TOPIC_PREFIX_KEY, getName(config)); + // scratch. + props.setProperty(TOPIC_PREFIX_KEY, sanitizeTopicPrefix(getName(config))); // includes props.putAll(getIncludeConfiguration(catalog, config)); @@ -85,6 +86,33 @@ public Properties getDebeziumProperties( return props; } + public static String sanitizeTopicPrefix(final String topicName) { + StringBuilder sanitizedNameBuilder = new StringBuilder(topicName.length()); + boolean changed = false; + + for (int i = 0; i < topicName.length(); ++i) { + char c = topicName.charAt(i); + if (isValidCharacter(c)) { + sanitizedNameBuilder.append(c); + } else { + sanitizedNameBuilder.append(ReplacementFunction.UNDERSCORE_REPLACEMENT.replace(c)); + changed = true; + } + } + + if (changed) { + return sanitizedNameBuilder.toString(); + } else { + return topicName; + } + } + + // We need to keep the validation rule the same as debezium engine, which is defined here: + // https://github.com/debezium/debezium/blob/c51ef3099a688efb41204702d3aa6d4722bb4825/debezium-core/src/main/java/io/debezium/schema/AbstractTopicNamingStrategy.java#L178 + private static boolean isValidCharacter(char c) { + return c == '.' || c == '_' || c == '-' || c >= 'A' && c <= 'Z' || c >= 'a' && c <= 'z' || c >= '0' && c <= '9'; + } + protected abstract Properties getConnectionConfiguration(final JsonNode config); protected abstract String getName(final JsonNode config); diff --git a/airbyte-cdk/java/airbyte-cdk/db-sources/src/testFixtures/java/io/airbyte/cdk/integrations/debezium/CdcSourceTest.java b/airbyte-cdk/java/airbyte-cdk/db-sources/src/testFixtures/java/io/airbyte/cdk/integrations/debezium/CdcSourceTest.java index a0ee71a226d0d..26544fe47fbd3 100644 --- a/airbyte-cdk/java/airbyte-cdk/db-sources/src/testFixtures/java/io/airbyte/cdk/integrations/debezium/CdcSourceTest.java +++ b/airbyte-cdk/java/airbyte-cdk/db-sources/src/testFixtures/java/io/airbyte/cdk/integrations/debezium/CdcSourceTest.java @@ -219,6 +219,19 @@ protected void writeRecords( recordJson.get(modelCol).asText()); } + protected void deleteMessageOnIdCol(final String streamName, final String idCol, final int idValue) { + testdb.with("DELETE FROM %s.%s WHERE %s = %s", modelsSchema(), streamName, idCol, idValue); + } + + protected void deleteCommand(final String streamName) { + testdb.with("DELETE FROM %s.%s", modelsSchema(), streamName); + } + + protected void updateCommand(final String streamName, final String modelCol, final String modelVal, final String idCol, final int idValue) { + testdb.with("UPDATE %s.%s SET %s = '%s' WHERE %s = %s", modelsSchema(), streamName, + modelCol, modelVal, COL_ID, 11); + } + static protected Set removeDuplicates(final Set messages) { final Set existingDataRecordsWithoutUpdated = new HashSet<>(); final Set output = new HashSet<>(); @@ -346,7 +359,7 @@ void testDelete() throws Exception { final List stateMessages1 = extractStateMessages(actualRecords1); assertExpectedStateMessages(stateMessages1); - testdb.with("DELETE FROM %s.%s WHERE %s = %s", modelsSchema(), MODELS_STREAM_NAME, COL_ID, 11); + deleteMessageOnIdCol(MODELS_STREAM_NAME, COL_ID, 11); final JsonNode state = Jsons.jsonNode(Collections.singletonList(stateMessages1.get(stateMessages1.size() - 1))); final AutoCloseableIterator read2 = source() @@ -375,8 +388,7 @@ void testUpdate() throws Exception { final List stateMessages1 = extractStateMessages(actualRecords1); assertExpectedStateMessages(stateMessages1); - testdb.with("UPDATE %s.%s SET %s = '%s' WHERE %s = %s", modelsSchema(), MODELS_STREAM_NAME, - COL_MODEL, updatedModel, COL_ID, 11); + updateCommand(MODELS_STREAM_NAME, COL_MODEL, updatedModel, COL_ID, 11); final JsonNode state = Jsons.jsonNode(Collections.singletonList(stateMessages1.get(stateMessages1.size() - 1))); final AutoCloseableIterator read2 = source() @@ -536,8 +548,7 @@ void testCdcAndFullRefreshInSameSync() throws Exception { @DisplayName("When no records exist, no records are returned.") void testNoData() throws Exception { - testdb.with("DELETE FROM %s.%s", modelsSchema(), MODELS_STREAM_NAME); - + deleteCommand(MODELS_STREAM_NAME); final AutoCloseableIterator read = source().read(config(), getConfiguredCatalog(), null); final List actualRecords = AutoCloseableIterators.toListAndClose(read); diff --git a/airbyte-integrations/connectors/source-mysql/build.gradle b/airbyte-integrations/connectors/source-mysql/build.gradle index a2f866a741181..24c52f470102f 100644 --- a/airbyte-integrations/connectors/source-mysql/build.gradle +++ b/airbyte-integrations/connectors/source-mysql/build.gradle @@ -6,7 +6,7 @@ plugins { } airbyteJavaConnector { - cdkVersionRequired = '0.19.0' + cdkVersionRequired = '0.20.3' features = ['db-sources'] useLocalCdk = false } diff --git a/airbyte-integrations/connectors/source-mysql/metadata.yaml b/airbyte-integrations/connectors/source-mysql/metadata.yaml index 69357901d32ac..16b607af14ba5 100644 --- a/airbyte-integrations/connectors/source-mysql/metadata.yaml +++ b/airbyte-integrations/connectors/source-mysql/metadata.yaml @@ -9,7 +9,7 @@ data: connectorSubtype: database connectorType: source definitionId: 435bb9a5-7887-4809-aa58-28c27df0d7ad - dockerImageTag: 3.3.4 + dockerImageTag: 3.3.5 dockerRepository: airbyte/source-mysql documentationUrl: https://docs.airbyte.com/integrations/sources/mysql githubIssueLabel: source-mysql diff --git a/airbyte-integrations/connectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/cdc/MySqlDebeziumStateUtil.java b/airbyte-integrations/connectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/cdc/MySqlDebeziumStateUtil.java index ae2f99f34b4a7..dda584d1de03e 100644 --- a/airbyte-integrations/connectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/cdc/MySqlDebeziumStateUtil.java +++ b/airbyte-integrations/connectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/cdc/MySqlDebeziumStateUtil.java @@ -237,8 +237,11 @@ public JsonNode constructInitialDebeziumState(final Properties properties, // We use the schema_only_recovery property cause using this mode will instruct Debezium to // construct the db schema history. properties.setProperty("snapshot.mode", "schema_only_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. final AirbyteFileOffsetBackingStore offsetManager = AirbyteFileOffsetBackingStore.initializeState( - constructBinlogOffset(database, database.getSourceConfig().get(JdbcUtils.DATABASE_KEY).asText()), + constructBinlogOffset(database, dbName, DebeziumPropertiesManager.sanitizeTopicPrefix(dbName)), Optional.empty()); final AirbyteSchemaHistoryStorage schemaHistoryStorage = AirbyteSchemaHistoryStorage.initializeDBHistory(new SchemaHistory<>(Optional.empty(), false), COMPRESSION_ENABLED); @@ -303,13 +306,13 @@ public static JsonNode serialize(final Map offset, final SchemaH * Method to construct initial Debezium state which can be passed onto Debezium engine to make it * process binlogs from a specific file and position and skip snapshot phase */ - private JsonNode constructBinlogOffset(final JdbcDatabase database, final String dbName) { - return format(getStateAttributesFromDB(database), dbName, Instant.now()); + private JsonNode constructBinlogOffset(final JdbcDatabase database, final String debeziumName, final String topicPrefixName) { + return format(getStateAttributesFromDB(database), debeziumName, topicPrefixName, Instant.now()); } @VisibleForTesting - public JsonNode format(final MysqlDebeziumStateAttributes attributes, final String dbName, final Instant time) { - final String key = "[\"" + dbName + "\",{\"server\":\"" + dbName + "\"}]"; + public JsonNode format(final MysqlDebeziumStateAttributes attributes, final String debeziumName, final String topicPrefixName, final Instant time) { + final String key = "[\"" + debeziumName + "\",{\"server\":\"" + topicPrefixName + "\"}]"; final String gtidSet = attributes.gtidSet().isPresent() ? ",\"gtids\":\"" + attributes.gtidSet().get() + "\"" : ""; final String value = "{\"transaction_id\":null,\"ts_sec\":" + time.getEpochSecond() + ",\"file\":\"" + attributes.binlogFilename() + "\",\"pos\":" diff --git a/airbyte-integrations/connectors/source-mysql/src/test/java/io/airbyte/integrations/source/mysql/CdcMysqlSourceTest.java b/airbyte-integrations/connectors/source-mysql/src/test/java/io/airbyte/integrations/source/mysql/CdcMysqlSourceTest.java index 27392901935ec..5825ae0848df3 100644 --- a/airbyte-integrations/connectors/source-mysql/src/test/java/io/airbyte/integrations/source/mysql/CdcMysqlSourceTest.java +++ b/airbyte-integrations/connectors/source-mysql/src/test/java/io/airbyte/integrations/source/mysql/CdcMysqlSourceTest.java @@ -100,7 +100,12 @@ protected void purgeAllBinaryLogs() { @Override protected String createSchemaSqlFmt() { - return "CREATE DATABASE IF NOT EXISTS %s;"; + return "CREATE DATABASE IF NOT EXISTS `%s`;"; + } + + @Override + protected String createTableSqlFmt() { + return "CREATE TABLE `%s`.`%s`(%s);"; } @Override @@ -176,6 +181,36 @@ protected void addCdcDefaultCursorField(final AirbyteStream stream) { } } + @Override + protected void writeRecords( + final JsonNode recordJson, + final String dbName, + final String streamName, + final String idCol, + final String makeIdCol, + final String modelCol) { + testdb.with("INSERT INTO `%s` .`%s` (%s, %s, %s) VALUES (%s, %s, '%s');", dbName, streamName, + idCol, makeIdCol, modelCol, + recordJson.get(idCol).asInt(), recordJson.get(makeIdCol).asInt(), + recordJson.get(modelCol).asText()); + } + + @Override + protected void deleteMessageOnIdCol(final String streamName, final String idCol, final int idValue) { + testdb.with("DELETE FROM `%s`.`%s` WHERE %s = %s", modelsSchema(), streamName, idCol, idValue); + } + + @Override + protected void deleteCommand(final String streamName) { + testdb.with("DELETE FROM `%s`.`%s`", modelsSchema(), streamName); + } + + @Override + protected void updateCommand(final String streamName, final String modelCol, final String modelVal, final String idCol, final int idValue) { + testdb.with("UPDATE `%s`.`%s` SET %s = '%s' WHERE %s = %s", modelsSchema(), streamName, + modelCol, modelVal, COL_ID, 11); + } + @Test protected void syncWithReplicationClientPrivilegeRevokedFailsCheck() throws Exception { testdb.with("REVOKE REPLICATION CLIENT ON *.* FROM %s@'%%';", testdb.getUserName()); diff --git a/airbyte-integrations/connectors/source-mysql/src/test/java/io/airbyte/integrations/source/mysql/CdcMysqlSourceWithSpecialDbNameTest.java b/airbyte-integrations/connectors/source-mysql/src/test/java/io/airbyte/integrations/source/mysql/CdcMysqlSourceWithSpecialDbNameTest.java new file mode 100644 index 0000000000000..06c30176b4705 --- /dev/null +++ b/airbyte-integrations/connectors/source-mysql/src/test/java/io/airbyte/integrations/source/mysql/CdcMysqlSourceWithSpecialDbNameTest.java @@ -0,0 +1,20 @@ +/* + * Copyright (c) 2023 Airbyte, Inc., all rights reserved. + */ + +package io.airbyte.integrations.source.mysql; + +import io.airbyte.integrations.source.mysql.MySQLTestDatabase.BaseImage; +import io.airbyte.integrations.source.mysql.MySQLTestDatabase.ContainerModifier; + +public class CdcMysqlSourceWithSpecialDbNameTest extends CdcMysqlSourceTest { + + public static final String INVALID_DB_NAME = "invalid@name"; + + @Override + protected MySQLTestDatabase createTestDatabase() { + return MySQLTestDatabase.inWithDbName(BaseImage.MYSQL_8, INVALID_DB_NAME, ContainerModifier.INVALID_TIMEZONE_CEST, ContainerModifier.CUSTOM_NAME) + .withCdcPermissions(); + } + +} diff --git a/airbyte-integrations/connectors/source-mysql/src/test/java/io/airbyte/integrations/source/mysql/MysqlDebeziumStateUtilTest.java b/airbyte-integrations/connectors/source-mysql/src/test/java/io/airbyte/integrations/source/mysql/MysqlDebeziumStateUtilTest.java index 1194f8ae96da3..284fdaa300f06 100644 --- a/airbyte-integrations/connectors/source-mysql/src/test/java/io/airbyte/integrations/source/mysql/MysqlDebeziumStateUtilTest.java +++ b/airbyte-integrations/connectors/source-mysql/src/test/java/io/airbyte/integrations/source/mysql/MysqlDebeziumStateUtilTest.java @@ -92,7 +92,7 @@ public void debeziumInitialStateConstructTest() throws SQLException { public void formatTestWithGtid() { final MySqlDebeziumStateUtil mySqlDebeziumStateUtil = new MySqlDebeziumStateUtil(); final JsonNode debeziumState = mySqlDebeziumStateUtil.format(new MysqlDebeziumStateAttributes("binlog.000002", 633, - Optional.of("3E11FA47-71CA-11E1-9E33-C80AA9429562:1-5")), "db_fgnfxvllud", Instant.parse("2023-06-06T08:36:10.341842Z")); + Optional.of("3E11FA47-71CA-11E1-9E33-C80AA9429562:1-5")), "db_fgnfxvllud", "db_fgnfxvllud", Instant.parse("2023-06-06T08:36:10.341842Z")); final Map stateAsMap = Jsons.object(debeziumState, Map.class); Assertions.assertEquals(1, stateAsMap.size()); Assertions.assertTrue(stateAsMap.containsKey("[\"db_fgnfxvllud\",{\"server\":\"db_fgnfxvllud\"}]")); @@ -113,7 +113,7 @@ public void formatTestWithGtid() { debeziumState, config); Assertions.assertTrue(parsedOffset.isPresent()); final JsonNode stateGeneratedUsingParsedOffset = - mySqlDebeziumStateUtil.format(parsedOffset.get(), "db_fgnfxvllud", Instant.parse("2023-06-06T08:36:10.341842Z")); + mySqlDebeziumStateUtil.format(parsedOffset.get(), "db_fgnfxvllud", "db_fgnfxvllud", Instant.parse("2023-06-06T08:36:10.341842Z")); Assertions.assertEquals(debeziumState, stateGeneratedUsingParsedOffset); } @@ -121,7 +121,7 @@ public void formatTestWithGtid() { public void formatTestWithoutGtid() { final MySqlDebeziumStateUtil mySqlDebeziumStateUtil = new MySqlDebeziumStateUtil(); final JsonNode debeziumState = mySqlDebeziumStateUtil.format(new MysqlDebeziumStateAttributes("binlog.000002", 633, - Optional.empty()), "db_fgnfxvllud", Instant.parse("2023-06-06T08:36:10.341842Z")); + Optional.empty()), "db_fgnfxvllud", "db_fgnfxvllud", Instant.parse("2023-06-06T08:36:10.341842Z")); final Map stateAsMap = Jsons.object(debeziumState, Map.class); Assertions.assertEquals(1, stateAsMap.size()); Assertions.assertTrue(stateAsMap.containsKey("[\"db_fgnfxvllud\",{\"server\":\"db_fgnfxvllud\"}]")); @@ -141,7 +141,7 @@ public void formatTestWithoutGtid() { debeziumState, config); Assertions.assertTrue(parsedOffset.isPresent()); final JsonNode stateGeneratedUsingParsedOffset = - mySqlDebeziumStateUtil.format(parsedOffset.get(), "db_fgnfxvllud", Instant.parse("2023-06-06T08:36:10.341842Z")); + mySqlDebeziumStateUtil.format(parsedOffset.get(), "db_fgnfxvllud", "db_fgnfxvllud", Instant.parse("2023-06-06T08:36:10.341842Z")); Assertions.assertEquals(debeziumState, stateGeneratedUsingParsedOffset); } diff --git a/airbyte-integrations/connectors/source-mysql/src/testFixtures/java/io/airbyte/integrations/source/mysql/MySQLContainerFactory.java b/airbyte-integrations/connectors/source-mysql/src/testFixtures/java/io/airbyte/integrations/source/mysql/MySQLContainerFactory.java index 2222cced0dcfe..2e9fba65fbb70 100644 --- a/airbyte-integrations/connectors/source-mysql/src/testFixtures/java/io/airbyte/integrations/source/mysql/MySQLContainerFactory.java +++ b/airbyte-integrations/connectors/source-mysql/src/testFixtures/java/io/airbyte/integrations/source/mysql/MySQLContainerFactory.java @@ -35,6 +35,8 @@ public void withMoscowTimezone(MySQLContainer container) { container.withEnv("TZ", "Europe/Moscow"); } + public void withCustomName(MySQLContainer container) {} // do nothing + public void withRootAndServerCertificates(MySQLContainer container) { execInContainer(container, "sed -i '31 a ssl' /etc/my.cnf", diff --git a/airbyte-integrations/connectors/source-mysql/src/testFixtures/java/io/airbyte/integrations/source/mysql/MySQLTestDatabase.java b/airbyte-integrations/connectors/source-mysql/src/testFixtures/java/io/airbyte/integrations/source/mysql/MySQLTestDatabase.java index 99b98a5465dfa..3e0b1511d5d0b 100644 --- a/airbyte-integrations/connectors/source-mysql/src/testFixtures/java/io/airbyte/integrations/source/mysql/MySQLTestDatabase.java +++ b/airbyte-integrations/connectors/source-mysql/src/testFixtures/java/io/airbyte/integrations/source/mysql/MySQLTestDatabase.java @@ -37,7 +37,8 @@ public enum ContainerModifier { ROOT_AND_SERVER_CERTIFICATES("withRootAndServerCertificates"), CLIENT_CERTITICATE("withClientCertificate"), NETWORK("withNetwork"), - ; + + CUSTOM_NAME("withCustomName"); public final String methodName; @@ -53,6 +54,15 @@ static public MySQLTestDatabase in(BaseImage baseImage, ContainerModifier... met return new MySQLTestDatabase(container).initialized(); } + static public MySQLTestDatabase inWithDbName(BaseImage baseImage, String dbName, ContainerModifier... methods) { + String[] methodNames = Stream.of(methods).map(im -> im.methodName).toList().toArray(new String[0]); + final var container = new MySQLContainerFactory().shared(baseImage.reference, methodNames); + MySQLTestDatabase db = new MySQLTestDatabase(container); + db.setDatabaseName(dbName); + db.initialized(); + return db; + } + public MySQLTestDatabase(MySQLContainer container) { super(container); } @@ -70,6 +80,26 @@ public MySQLTestDatabase withoutStrictMode() { } static private final int MAX_CONNECTIONS = 1000; + private String databaseName = ""; + + @Override + public String getDatabaseName() { + if (databaseName.isBlank()) { + return super.getDatabaseName(); + } else { + return databaseName; + } + } + + @Override + public void close() { + super.close(); + databaseName = ""; + } + + public void setDatabaseName(final String databaseName) { + this.databaseName = databaseName; + } @Override protected Stream> inContainerBootstrapCmd() { @@ -80,18 +110,19 @@ protected Stream> inContainerBootstrapCmd() { "sh", "-c", "ln -s -f /var/lib/mysql/mysql.sock /var/run/mysqld/mysqld.sock"), mysqlCmd(Stream.of( String.format("SET GLOBAL max_connections=%d", MAX_CONNECTIONS), - String.format("CREATE DATABASE %s", getDatabaseName()), + String.format("CREATE DATABASE \\`%s\\`", getDatabaseName()), String.format("CREATE USER '%s' IDENTIFIED BY '%s'", getUserName(), getPassword()), // Grant privileges also to the container's user, which is not root. String.format("GRANT ALL PRIVILEGES ON *.* TO '%s', '%s' WITH GRANT OPTION", getUserName(), getContainer().getUsername())))); + } @Override protected Stream inContainerUndoBootstrapCmd() { return mysqlCmd(Stream.of( String.format("DROP USER '%s'", getUserName()), - String.format("DROP DATABASE %s", getDatabaseName()))); + String.format("DROP DATABASE \\`%s\\`", getDatabaseName()))); } @Override diff --git a/docs/integrations/sources/mysql.md b/docs/integrations/sources/mysql.md index d2f56faea6bd4..64f8de10e4041 100644 --- a/docs/integrations/sources/mysql.md +++ b/docs/integrations/sources/mysql.md @@ -223,10 +223,11 @@ Any database or table encoding combination of charset and collation is supported | Version | Date | Pull Request | Subject | |:--------|:-----------|:-----------------------------------------------------------|:------------------------------------------------------------------------------------------------------------------------------------------------| -| 3.3.4 | 2024-02-08 | [34750](https://github.com/airbytehq/airbyte/pull/34750) | Adopt CDK 0.19.0 | -| 3.3.3 | 2024-01-26 | [34573](https://github.com/airbytehq/airbyte/pull/34573) | Adopt CDK v0.16.0. | -| 3.3.2 | 2024-01-08 | [33005](https://github.com/airbytehq/airbyte/pull/33005) | Adding count stats for incremental sync in AirbyteStateMessage -| 3.3.1 | 2024-01-03 | [33312](https://github.com/airbytehq/airbyte/pull/33312) | Adding count stats in AirbyteStateMessage | +| 3.3.5 | 2024-02-12 | [34580](https://github.com/airbytehq/airbyte/pull/34580) | Support special chars in db name | +| 3.3.4 | 2024-02-08 | [34750](https://github.com/airbytehq/airbyte/pull/34750) | Adopt CDK 0.19.0 | +| 3.3.3 | 2024-01-26 | [34573](https://github.com/airbytehq/airbyte/pull/34573) | Adopt CDK v0.16.0. | +| 3.3.2 | 2024-01-08 | [33005](https://github.com/airbytehq/airbyte/pull/33005) | Adding count stats for incremental sync in AirbyteStateMessage +| 3.3.1 | 2024-01-03 | [33312](https://github.com/airbytehq/airbyte/pull/33312) | Adding count stats in AirbyteStateMessage | | 3.3.0 | 2023-12-19 | [33436](https://github.com/airbytehq/airbyte/pull/33436) | Remove LEGACY state flag | | 3.2.4 | 2023-12-12 | [33356](https://github.com/airbytehq/airbyte/pull/33210) | Support for better debugging tools.. | | 3.2.3 | 2023-12-08 | [33210](https://github.com/airbytehq/airbyte/pull/33210) | Update MySql driver property value for zero date handling. |