[Enhancement][API]Enhance mysql connection properties (#15433)

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-mysql/src/main/java/org/apache/dolphinscheduler/plugin/datasource/mysql/param/MySQLDataSourceProcessor.java

@@ -33,10 +33,9 @@ import org.apache.commons.collections4.MapUtils;
 import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.SQLException;
-import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -55,9 +54,6 @@ public class MySQLDataSourceProcessor extends AbstractDataSourceProcessor {
 
     private static final String ALLOW_URL_IN_LOCAL_IN_FILE_NAME = "allowUrlInLocalInfile";
 
-    private static final String APPEND_PARAMS =
-            "allowLoadLocalInfile=false&autoDeserialize=false&allowLocalInfile=false&allowUrlInLocalInfile=false";
-
     @Override
     public BaseDataSourceParamDTO castDatasourceParamDTO(String paramJson) {
         return JSONUtils.parseObject(paramJson, MySQLDataSourceParamDTO.class);
@@ -119,11 +115,7 @@ public class MySQLDataSourceProcessor extends AbstractDataSourceProcessor {
     @Override
     public String getJdbcUrl(ConnectionParam connectionParam) {
         MySQLConnectionParam mysqlConnectionParam = (MySQLConnectionParam) connectionParam;
-        String jdbcUrl = mysqlConnectionParam.getJdbcUrl();
+        return mysqlConnectionParam.getJdbcUrl();
-        if (MapUtils.isNotEmpty(mysqlConnectionParam.getOther())) {
-            return String.format("%s?%s&%s", jdbcUrl, transformOther(mysqlConnectionParam.getOther()), APPEND_PARAMS);
-        }
-        return String.format("%s?%s", jdbcUrl, APPEND_PARAMS);
     }
 
     @Override
@@ -140,7 +132,32 @@ public class MySQLDataSourceProcessor extends AbstractDataSourceProcessor {
             log.warn("sensitive param : {} in password field is filtered", AUTO_DESERIALIZE);
             password = password.replace(AUTO_DESERIALIZE, "");
         }
-        return DriverManager.getConnection(getJdbcUrl(connectionParam), user, password);
+
+        Properties connectionProperties = getConnectionProperties(mysqlConnectionParam, user, password);
+
+        return DriverManager.getConnection(getJdbcUrl(connectionParam), connectionProperties);
+    }
+
+    private Properties getConnectionProperties(MySQLConnectionParam mysqlConnectionParam, String user,
+                                               String password) {
+        Properties connectionProperties = new Properties();
+        connectionProperties.put("user", user);
+        connectionProperties.put("password", password);
+        Map<String, String> paramMap = mysqlConnectionParam.getOther();
+        if (MapUtils.isNotEmpty(paramMap)) {
+            paramMap.forEach((k, v) -> {
+                if (!checkKeyIsLegitimate(k)) {
+                    log.info("Key `{}` is not legitimate for security reason", k);
+                    return;
+                }
+                connectionProperties.put(k, v);
+            });
+        }
+        connectionProperties.put(AUTO_DESERIALIZE, "false");
+        connectionProperties.put(ALLOW_LOAD_LOCAL_IN_FILE_NAME, "false");
+        connectionProperties.put(ALLOW_LOCAL_IN_FILE_NAME, "false");
+        connectionProperties.put(ALLOW_URL_IN_LOCAL_IN_FILE_NAME, "false");
+        return connectionProperties;
     }
 
     @Override
@@ -158,25 +175,6 @@ public class MySQLDataSourceProcessor extends AbstractDataSourceProcessor {
         return SQLParserUtils.splitAndRemoveComment(sql, com.alibaba.druid.DbType.mysql);
     }
 
-    private String transformOther(Map<String, String> paramMap) {
-        if (MapUtils.isEmpty(paramMap)) {
-            return null;
-        }
-        Map<String, String> otherMap = new HashMap<>();
-        paramMap.forEach((k, v) -> {
-            if (!checkKeyIsLegitimate(k)) {
-                return;
-            }
-            otherMap.put(k, v);
-        });
-        if (MapUtils.isEmpty(otherMap)) {
-            return null;
-        }
-        List<String> otherList = new ArrayList<>();
-        otherMap.forEach((key, value) -> otherList.add(String.format("%s=%s", key, value)));
-        return String.join("&", otherList);
-    }
-
     private static boolean checkKeyIsLegitimate(String key) {
         return !key.contains(ALLOW_LOAD_LOCAL_IN_FILE_NAME)
                 && !key.contains(AUTO_DESERIALIZE)

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-mysql/src/test/java/org/apache/dolphinscheduler/plugin/datasource/mysql/param/MySQLDataSourceProcessorTest.java

@@ -80,7 +80,7 @@ public class MySQLDataSourceProcessorTest {
         MySQLConnectionParam mysqlConnectionParam = new MySQLConnectionParam();
         mysqlConnectionParam.setJdbcUrl("jdbc:mysql://localhost:3306/default");
         Assertions.assertEquals(
-                "jdbc:mysql://localhost:3306/default?allowLoadLocalInfile=false&autoDeserialize=false&allowLocalInfile=false&allowUrlInLocalInfile=false",
+                "jdbc:mysql://localhost:3306/default",
                 mysqlDatasourceProcessor.getJdbcUrl(mysqlConnectionParam));
     }
 

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-mysql/src/test/java/org/apache/dolphinscheduler/plugin/datasource/mysql/utils/DataSourceUtilsTest.java

@@ -117,7 +117,7 @@ public class DataSourceUtilsTest {
         mysqlConnectionParam.setJdbcUrl("jdbc:mysql://localhost:3308");
         String jdbcUrl = DataSourceUtils.getJdbcUrl(DbType.MYSQL, mysqlConnectionParam);
         Assertions.assertEquals(
-                "jdbc:mysql://localhost:3308?allowLoadLocalInfile=false&autoDeserialize=false&allowLocalInfile=false&allowUrlInLocalInfile=false",
+                "jdbc:mysql://localhost:3308",
                 jdbcUrl);
     }