(flink-connector-jdbc) 02/02: [FLINK-35176][Connector/JDBC] Support property authentication connection for dynamic jdbc source table

[fa...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

fanrui pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-jdbc.git

commit a9135f98bd708525c1474d76da98dfce149fb4d3
Author: Roc Marshal <fl...@126.com>
AuthorDate: Sat Apr 20 16:27:43 2024 +0800

    [FLINK-35176][Connector/JDBC] Support property authentication connection for dynamic jdbc source table
---
 .../org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactory.java | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactory.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactory.java
index 3eecfa15..58363a52 100644
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactory.java
+++ b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactory.java
@@ -71,6 +71,7 @@ import static org.apache.flink.connector.jdbc.table.JdbcConnectorOptions.SINK_PA
 import static org.apache.flink.connector.jdbc.table.JdbcConnectorOptions.TABLE_NAME;
 import static org.apache.flink.connector.jdbc.table.JdbcConnectorOptions.URL;
 import static org.apache.flink.connector.jdbc.table.JdbcConnectorOptions.USERNAME;
+import static org.apache.flink.connector.jdbc.utils.JdbcUtils.getConnectionProperties;
 
 /**
  * Factory for creating configured instances of {@link JdbcDynamicTableSource} and {@link
@@ -153,6 +154,8 @@ public class JdbcDynamicTableFactory implements DynamicTableSourceFactory, Dynam
         readableConfig.getOptional(USERNAME).ifPresent(builder::setUsername);
         readableConfig.getOptional(PASSWORD).ifPresent(builder::setPassword);
         readableConfig.getOptional(COMPATIBLE_MODE).ifPresent(builder::setCompatibleMode);
+        getConnectionProperties(readableConfig)
+                .forEach((key, value) -> builder.setProperty(key.toString(), value.toString()));
         return builder.build();
     }