add jdbcProperties options (#146)

* add use SSL Co-authored-by: wudi <>
apache · Jun 9, 2023 · 8d30367 · 8d30367
1 parent 33a9fd8
commit 8d30367
Showing 1 changed file with 20 additions and 7 deletions.
diff --git a/...ris-connector/src/main/java/org/apache/doris/flink/tools/cdc/mysql/MysqlDatabaseSync.java b/...ris-connector/src/main/java/org/apache/doris/flink/tools/cdc/mysql/MysqlDatabaseSync.java
@@ -50,18 +50,19 @@
 public class MysqlDatabaseSync extends DatabaseSync {
     private static final Logger LOG = LoggerFactory.getLogger(MysqlDatabaseSync.class);
 
+    private static String JDBC_URL = "jdbc:mysql://%s:%d?useInformationSchema=true";
+
     public MysqlDatabaseSync() {
     }
 
     @Override
     public Connection getConnection() throws SQLException {
-        return DriverManager.getConnection(
-                String.format(
-                        "jdbc:mysql://%s:%d?useInformationSchema=true",
-                        config.get(MySqlSourceOptions.HOSTNAME),
-                        config.get(MySqlSourceOptions.PORT)),
-                config.get(MySqlSourceOptions.USERNAME),
-                config.get(MySqlSourceOptions.PASSWORD));
+        Properties jdbcProperties = getJdbcProperties();
+        StringBuilder jdbcUrlSb = new StringBuilder(JDBC_URL);
+        jdbcProperties.forEach((key, value) -> jdbcUrlSb.append("&").append(key).append("=").append(value));
+        String jdbcUrl = String.format(jdbcUrlSb.toString(), config.get(MySqlSourceOptions.HOSTNAME), config.get(MySqlSourceOptions.PORT));
+
+        return DriverManager.getConnection(jdbcUrl,config.get(MySqlSourceOptions.USERNAME),config.get(MySqlSourceOptions.PASSWORD));
     }
 
     @Override
@@ -189,4 +190,16 @@ public DataStreamSource<String> buildCdcSource(StreamExecutionEnvironment env) {
                 mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source");
         return streamSource;
     }
+
+    private Properties getJdbcProperties(){
+        Properties jdbcProps = new Properties();
+        for (Map.Entry<String, String> entry : config.toMap().entrySet()) {
+            String key = entry.getKey();
+            String value = entry.getValue();
+            if (key.startsWith(JdbcUrlUtils.PROPERTIES_PREFIX)) {
+                jdbcProps.put(key.substring(JdbcUrlUtils.PROPERTIES_PREFIX.length()), value);
+            }
+        }
+        return jdbcProps;
+    }
 }