This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 57e8dc3f5 [cdc] In schema building phase of mysql cdc, the connection
should handle jdbc properties (#2507)
57e8dc3f5 is described below
commit 57e8dc3f51c66c22f419a5493bed22e5191283fd
Author: yuzelin <[email protected]>
AuthorDate: Thu Dec 14 15:50:13 2023 +0800
[cdc] In schema building phase of mysql cdc, the connection should handle
jdbc properties (#2507)
---
.../flink/action/cdc/mysql/MySqlActionUtils.java | 66 +++++++++++++++++-----
.../action/cdc/mysql/MySqlSyncDatabaseAction.java | 3 +-
.../action/cdc/mysql/MySqlSyncTableAction.java | 2 +-
.../cdc/mysql/MySqlCdcTypeMappingITCase.java | 29 ++++++++++
4 files changed, 85 insertions(+), 15 deletions(-)
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
index b44291fc4..ba01acd36 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
@@ -51,6 +51,7 @@ import java.util.Properties;
import java.util.function.Predicate;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import java.util.stream.Collectors;
import static
org.apache.paimon.flink.action.cdc.TypeMapping.TypeMappingMode.TINYINT1_NOT_BOOL;
@@ -66,17 +67,25 @@ public class MySqlActionUtils {
.withDescription(
"Whether capture the scan the newly added tables
or not, by default is true.");
- static Connection getConnection(Configuration mySqlConfig, boolean
tinyint1NotBool)
+ static Connection getConnection(Configuration mySqlConfig, Map<String,
String> jdbcProperties)
throws Exception {
+ String paramString = "";
+ if (!jdbcProperties.isEmpty()) {
+ paramString =
+ "?"
+ + jdbcProperties.entrySet().stream()
+ .map(e -> e.getKey() + "=" + e.getValue())
+ .collect(Collectors.joining("&"));
+ }
+
String url =
String.format(
"jdbc:mysql://%s:%d%s",
mySqlConfig.get(MySqlSourceOptions.HOSTNAME),
mySqlConfig.get(MySqlSourceOptions.PORT),
- // we need to add the `tinyInt1isBit` parameter to the
connection url to
- // make sure the tinyint(1) in MySQL is converted to
bits or not. Refer to
- //
https://dev.mysql.com/doc/connector-j/8.0/en/connector-j-connp-props-result-sets.html#cj-conn-prop_tinyInt1isBit
- tinyint1NotBool ? "?tinyInt1isBit=false" : "");
+ paramString);
+
+ LOG.info("Connect to MySQL server using url: {}", url);
return DriverManager.getConnection(
url,
@@ -93,9 +102,9 @@ public class MySqlActionUtils {
Pattern databasePattern =
Pattern.compile(mySqlConfig.get(MySqlSourceOptions.DATABASE_NAME));
MySqlSchemasInfo mySqlSchemasInfo = new MySqlSchemasInfo();
- try (Connection conn =
- MySqlActionUtils.getConnection(
- mySqlConfig,
typeMapping.containsMode(TINYINT1_NOT_BOOL))) {
+ Map<String, String> jdbcProperties = getJdbcProperties(typeMapping,
mySqlConfig);
+
+ try (Connection conn = MySqlActionUtils.getConnection(mySqlConfig,
jdbcProperties)) {
DatabaseMetaData metaData = conn.getMetaData();
try (ResultSet schemas = metaData.getCatalogs()) {
while (schemas.next()) {
@@ -129,7 +138,7 @@ public class MySqlActionUtils {
}
public static MySqlSource<String> buildMySqlSource(
- Configuration mySqlConfig, String tableList) {
+ Configuration mySqlConfig, String tableList, TypeMapping
typeMapping) {
MySqlSourceBuilder<String> sourceBuilder = MySqlSource.builder();
sourceBuilder
@@ -195,18 +204,18 @@ public class MySqlActionUtils {
}
Properties jdbcProperties = new Properties();
+ jdbcProperties.putAll(getJdbcProperties(typeMapping, mySqlConfig));
+ sourceBuilder.jdbcProperties(jdbcProperties);
+
Properties debeziumProperties = new Properties();
for (Map.Entry<String, String> entry : mySqlConfig.toMap().entrySet())
{
String key = entry.getKey();
String value = entry.getValue();
- if (key.startsWith(JdbcUrlUtils.PROPERTIES_PREFIX)) {
-
jdbcProperties.put(key.substring(JdbcUrlUtils.PROPERTIES_PREFIX.length()),
value);
- } else if
(key.startsWith(DebeziumOptions.DEBEZIUM_OPTIONS_PREFIX)) {
+ if (key.startsWith(DebeziumOptions.DEBEZIUM_OPTIONS_PREFIX)) {
debeziumProperties.put(
key.substring(DebeziumOptions.DEBEZIUM_OPTIONS_PREFIX.length()), value);
}
}
- sourceBuilder.jdbcProperties(jdbcProperties);
sourceBuilder.debeziumProperties(debeziumProperties);
Map<String, Object> customConverterConfigs = new HashMap<>();
@@ -223,6 +232,37 @@ public class MySqlActionUtils {
.build();
}
+ // see
+ //
https://ververica.github.io/flink-cdc-connectors/master/content/connectors/mysql-cdc.html#connector-options
+ //
https://dev.mysql.com/doc/connectors/en/connector-j-reference-configuration-properties.html
+ private static Map<String, String> getJdbcProperties(
+ TypeMapping typeMapping, Configuration mySqlConfig) {
+ Map<String, String> jdbcProperties =
+ mySqlConfig.toMap().entrySet().stream()
+ .filter(e ->
e.getKey().startsWith(JdbcUrlUtils.PROPERTIES_PREFIX))
+ .collect(
+ Collectors.toMap(
+ e ->
+ e.getKey()
+ .substring(
+
JdbcUrlUtils.PROPERTIES_PREFIX
+
.length()),
+ Map.Entry::getValue));
+
+ if (typeMapping.containsMode(TINYINT1_NOT_BOOL)) {
+ String tinyInt1isBit = jdbcProperties.get("tinyInt1isBit");
+ if (tinyInt1isBit == null) {
+ jdbcProperties.put("tinyInt1isBit", "false");
+ } else if ("true".equals(jdbcProperties.get("tinyInt1isBit"))) {
+ throw new IllegalArgumentException(
+ "Type mapping option 'tinyint1-not-bool' conflicts
with jdbc properties 'jdbc.properties.tinyInt1isBit=true'. "
+ + "Option 'tinyint1-not-bool' is equal to
'jdbc.properties.tinyInt1isBit=false'.");
+ }
+ }
+
+ return jdbcProperties;
+ }
+
public static void registerJdbcDriver() {
try {
Class.forName("com.mysql.cj.jdbc.Driver");
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
index ca7678b41..c469f6d1d 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
@@ -194,7 +194,8 @@ public class MySqlSyncDatabaseAction extends
SyncDatabaseActionBase {
cdcSourceConfig.get(MySqlSourceOptions.DATABASE_NAME),
includingTables,
monitoredTables,
- excludedTables));
+ excludedTables),
+ typeMapping);
}
private void logNonPkTables(List<Identifier> nonPkTables) {
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
index f8551a48b..9bf18308a 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
@@ -111,7 +111,7 @@ public class MySqlSyncTableAction extends
SyncTableActionBase {
mySqlSchemasInfo.pkTables().stream()
.map(i -> i.getDatabaseName() + "\\." +
i.getObjectName())
.collect(Collectors.joining("|"));
- return MySqlActionUtils.buildMySqlSource(cdcSourceConfig, tableList);
+ return MySqlActionUtils.buildMySqlSource(cdcSourceConfig, tableList,
typeMapping);
}
private void validateMySqlTableInfos(MySqlSchemasInfo mySqlSchemasInfo) {
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlCdcTypeMappingITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlCdcTypeMappingITCase.java
index 7eb7de979..237e6c262 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlCdcTypeMappingITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlCdcTypeMappingITCase.java
@@ -20,10 +20,12 @@ package org.apache.paimon.flink.action.cdc.mysql;
import org.apache.paimon.flink.action.cdc.TypeMapping;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.testutils.assertj.AssertionUtils;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import com.ververica.cdc.connectors.mysql.table.JdbcUrlUtils;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
@@ -43,6 +45,7 @@ import static
org.apache.paimon.flink.action.cdc.TypeMapping.TypeMappingMode.TIN
import static
org.apache.paimon.flink.action.cdc.TypeMapping.TypeMappingMode.TO_NULLABLE;
import static
org.apache.paimon.flink.action.cdc.TypeMapping.TypeMappingMode.TO_STRING;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** IT test for {@link TypeMapping} in MySQL CDC. */
public class MySqlCdcTypeMappingITCase extends MySqlActionITCaseBase {
@@ -61,6 +64,10 @@ public class MySqlCdcTypeMappingITCase extends
MySqlActionITCaseBase {
Map<String, String> mySqlConfig = getBasicMySqlConfig();
mySqlConfig.put("database-name", "tinyint1_not_bool_test");
+ // test tinyInt1isBit compatibility and url building
+ mySqlConfig.put(JdbcUrlUtils.PROPERTIES_PREFIX + "tinyInt1isBit",
"false");
+ mySqlConfig.put(JdbcUrlUtils.PROPERTIES_PREFIX + "useSSL", "false");
+
MySqlSyncDatabaseAction action =
syncDatabaseActionBuilder(mySqlConfig)
.withTableConfig(getBasicTableConfig())
@@ -113,6 +120,28 @@ public class MySqlCdcTypeMappingITCase extends
MySqlActionITCaseBase {
}
}
+ @Test
+ public void testConflictTinyInt1NotBool() {
+ Map<String, String> mySqlConfig = getBasicMySqlConfig();
+ mySqlConfig.put("database-name", "tinyint1_not_bool_test");
+ mySqlConfig.put(JdbcUrlUtils.PROPERTIES_PREFIX + "tinyInt1isBit",
"true");
+
+ MySqlSyncDatabaseAction action =
+ syncDatabaseActionBuilder(mySqlConfig)
+ .withTableConfig(getBasicTableConfig())
+ .withMode(COMBINED.configString())
+ .withTypeMappingModes(TINYINT1_NOT_BOOL.configString())
+ .build();
+
+ assertThatThrownBy(action::run)
+ .satisfies(
+ AssertionUtils.anyCauseMatches(
+ IllegalArgumentException.class,
+ "Type mapping option 'tinyint1-not-bool'
conflicts with "
+ + "jdbc properties
'jdbc.properties.tinyInt1isBit=true'. "
+ + "Option 'tinyint1-not-bool' is equal
to 'jdbc.properties.tinyInt1isBit=false'."));
+ }
+
// --------------------------------------- all-to-string
---------------------------------------
@Test
