Carl-Zhou-CN commented on code in PR #5388:
URL: https://github.com/apache/seatunnel/pull/5388#discussion_r1360548401
##########
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlIT.java:
##########
@@ -299,4 +320,153 @@ protected void initCatalog() {
jdbcCase.getJdbcUrl().replace(HOST,
dbServer.getHost())));
catalog.open();
}
+
+ private String getUrl() {
+ return URL.replace("HOST", dbServer.getHost());
+ }
+
+ @Test
+ public void parametersTest() throws SQLException, IOException,
ClassNotFoundException {
+ defaultSinkParametersTest();
+ defaultSourceParametersTest();
+ }
+
+ void defaultSinkParametersTest() throws IOException, SQLException,
ClassNotFoundException {
+ // case1 url not contains parameters and properties not contains
parameters
+ JdbcSink jdbcSink1 = new JdbcSink();
+ HashMap<String, Object> map1 = getMap();
+ map1.put("url", getUrl());
+ Config config1 = ConfigFactory.parseMap(map1);
+ Properties connectionProperties1 = getSinkProperties(jdbcSink1,
config1);
+
Assertions.assertEquals(connectionProperties1.get("rewriteBatchedStatements"),
"true");
+
+ // case2 url contains parameters and properties not contains parameters
+ JdbcSink jdbcSink2 = new JdbcSink();
+ HashMap<String, Object> map2 = getMap();
+ map2.put("url", getUrl() + "?rewriteBatchedStatements=false");
+ Config config2 = ConfigFactory.parseMap(map2);
+ Properties connectionProperties2 = getSinkProperties(jdbcSink2,
config2);
+
Assertions.assertEquals(connectionProperties2.get("rewriteBatchedStatements"),
"true");
+
+ // case3 url not contains parameters and properties not contains
parameters
+ JdbcSink jdbcSink3 = new JdbcSink();
+ HashMap<String, Object> map3 = getMap();
+ HashMap<String, String> properties3 = new HashMap<>();
+ properties3.put("rewriteBatchedStatements", "false");
+ map3.put("properties", properties3);
+ map3.put("url", getUrl());
+ Config config3 = ConfigFactory.parseMap(map3);
+ Properties connectionProperties3 = getSinkProperties(jdbcSink3,
config3);
+
Assertions.assertEquals(connectionProperties3.get("rewriteBatchedStatements"),
"false");
+
+ // case4 url contains parameters and properties contains parameters
+ JdbcSink jdbcSink4 = new JdbcSink();
+ HashMap<String, Object> map4 = getMap();
+ HashMap<String, String> properties4 = new HashMap<>();
+ properties4.put("useSSL", "true");
+ properties4.put("rewriteBatchedStatements", "false");
+ map4.put("properties", properties4);
+ map4.put("url", getUrl() +
"?useSSL=false&rewriteBatchedStatements=true");
+ Config config4 = ConfigFactory.parseMap(map4);
+ Properties connectionProperties4 = getSinkProperties(jdbcSink4,
config4);
+ Assertions.assertEquals(connectionProperties4.get("useSSL"), "true");
+
Assertions.assertEquals(connectionProperties4.get("rewriteBatchedStatements"),
"false");
+ }
+
+ void defaultSourceParametersTest() throws IOException, SQLException,
ClassNotFoundException {
+ // case1 url not contains parameters and properties not contains
parameters
+ JdbcSource jdbcSource1 = new JdbcSource();
+ HashMap<String, Object> map1 = getMap();
+ map1.put("url", getUrl());
+ map1.put("query", SQL);
+ Config config1 = ConfigFactory.parseMap(map1);
+ Properties connectionProperties1 = getSourceProperties(jdbcSource1,
config1);
+
Assertions.assertEquals(connectionProperties1.get("rewriteBatchedStatements"),
"true");
+
+ // case2 url contains parameters and properties not contains parameters
+ JdbcSource jdbcSource2 = new JdbcSource();
+ HashMap<String, Object> map2 = getMap();
+ map2.put("url", getUrl() + "?rewriteBatchedStatements=false");
+ map2.put("query", SQL);
+ Config config2 = ConfigFactory.parseMap(map2);
+ Properties connectionProperties2 = getSourceProperties(jdbcSource2,
config2);
+
Assertions.assertEquals(connectionProperties2.get("rewriteBatchedStatements"),
"true");
+
+ // case3 url not contains parameters and properties not contains
parameters
+ JdbcSource jdbcSource3 = new JdbcSource();
+ HashMap<String, Object> map3 = getMap();
Review Comment:
done
##########
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlIT.java:
##########
@@ -299,4 +320,153 @@ protected void initCatalog() {
jdbcCase.getJdbcUrl().replace(HOST,
dbServer.getHost())));
catalog.open();
}
+
+ private String getUrl() {
+ return URL.replace("HOST", dbServer.getHost());
+ }
+
+ @Test
+ public void parametersTest() throws SQLException, IOException,
ClassNotFoundException {
+ defaultSinkParametersTest();
+ defaultSourceParametersTest();
+ }
+
+ void defaultSinkParametersTest() throws IOException, SQLException,
ClassNotFoundException {
+ // case1 url not contains parameters and properties not contains
parameters
+ JdbcSink jdbcSink1 = new JdbcSink();
+ HashMap<String, Object> map1 = getMap();
+ map1.put("url", getUrl());
+ Config config1 = ConfigFactory.parseMap(map1);
+ Properties connectionProperties1 = getSinkProperties(jdbcSink1,
config1);
+
Assertions.assertEquals(connectionProperties1.get("rewriteBatchedStatements"),
"true");
+
+ // case2 url contains parameters and properties not contains parameters
+ JdbcSink jdbcSink2 = new JdbcSink();
+ HashMap<String, Object> map2 = getMap();
+ map2.put("url", getUrl() + "?rewriteBatchedStatements=false");
+ Config config2 = ConfigFactory.parseMap(map2);
+ Properties connectionProperties2 = getSinkProperties(jdbcSink2,
config2);
+
Assertions.assertEquals(connectionProperties2.get("rewriteBatchedStatements"),
"true");
+
+ // case3 url not contains parameters and properties not contains
parameters
+ JdbcSink jdbcSink3 = new JdbcSink();
+ HashMap<String, Object> map3 = getMap();
+ HashMap<String, String> properties3 = new HashMap<>();
+ properties3.put("rewriteBatchedStatements", "false");
+ map3.put("properties", properties3);
+ map3.put("url", getUrl());
+ Config config3 = ConfigFactory.parseMap(map3);
+ Properties connectionProperties3 = getSinkProperties(jdbcSink3,
config3);
+
Assertions.assertEquals(connectionProperties3.get("rewriteBatchedStatements"),
"false");
+
+ // case4 url contains parameters and properties contains parameters
+ JdbcSink jdbcSink4 = new JdbcSink();
+ HashMap<String, Object> map4 = getMap();
+ HashMap<String, String> properties4 = new HashMap<>();
+ properties4.put("useSSL", "true");
+ properties4.put("rewriteBatchedStatements", "false");
+ map4.put("properties", properties4);
+ map4.put("url", getUrl() +
"?useSSL=false&rewriteBatchedStatements=true");
+ Config config4 = ConfigFactory.parseMap(map4);
+ Properties connectionProperties4 = getSinkProperties(jdbcSink4,
config4);
+ Assertions.assertEquals(connectionProperties4.get("useSSL"), "true");
+
Assertions.assertEquals(connectionProperties4.get("rewriteBatchedStatements"),
"false");
+ }
+
+ void defaultSourceParametersTest() throws IOException, SQLException,
ClassNotFoundException {
+ // case1 url not contains parameters and properties not contains
parameters
+ JdbcSource jdbcSource1 = new JdbcSource();
+ HashMap<String, Object> map1 = getMap();
+ map1.put("url", getUrl());
+ map1.put("query", SQL);
+ Config config1 = ConfigFactory.parseMap(map1);
+ Properties connectionProperties1 = getSourceProperties(jdbcSource1,
config1);
+
Assertions.assertEquals(connectionProperties1.get("rewriteBatchedStatements"),
"true");
+
+ // case2 url contains parameters and properties not contains parameters
+ JdbcSource jdbcSource2 = new JdbcSource();
+ HashMap<String, Object> map2 = getMap();
+ map2.put("url", getUrl() + "?rewriteBatchedStatements=false");
+ map2.put("query", SQL);
+ Config config2 = ConfigFactory.parseMap(map2);
+ Properties connectionProperties2 = getSourceProperties(jdbcSource2,
config2);
+
Assertions.assertEquals(connectionProperties2.get("rewriteBatchedStatements"),
"true");
+
+ // case3 url not contains parameters and properties not contains
parameters
+ JdbcSource jdbcSource3 = new JdbcSource();
+ HashMap<String, Object> map3 = getMap();
+ HashMap<String, String> properties3 = new HashMap<>();
+ properties3.put("rewriteBatchedStatements", "false");
+ map3.put("properties", properties3);
+ map3.put("url", getUrl());
+ map3.put("query", SQL);
+ Config config3 = ConfigFactory.parseMap(map3);
+ Properties connectionProperties3 = getSourceProperties(jdbcSource3,
config3);
+
Assertions.assertEquals(connectionProperties3.get("rewriteBatchedStatements"),
"false");
+
+ // case4 url contains parameters and properties contains parameters
+ JdbcSource jdbcSource4 = new JdbcSource();
+ HashMap<String, Object> map4 = getMap();
+ HashMap<String, String> properties4 = new HashMap<>();
+ properties4.put("useSSL", "true");
+ properties4.put("rewriteBatchedStatements", "false");
+ map4.put("properties", properties4);
+ map4.put("url", getUrl() +
"?useSSL=false&rewriteBatchedStatements=true");
+ map4.put("query", SQL);
+ Config config4 = ConfigFactory.parseMap(map4);
+ Properties connectionProperties4 = getSourceProperties(jdbcSource4,
config4);
+ Assertions.assertEquals(connectionProperties4.get("useSSL"), "true");
+
Assertions.assertEquals(connectionProperties4.get("rewriteBatchedStatements"),
"false");
+ }
+
+ @NotNull private HashMap<String, Object> getMap() {
+ HashMap<String, Object> map = new HashMap<>();
+ map.put("driver", "com.mysql.cj.jdbc.Driver");
+ map.put("user", MYSQL_USERNAME);
+ map.put("password", MYSQL_PASSWORD);
+ return map;
+ }
+
+ private Properties getSinkProperties(JdbcSink jdbcSink, Config config)
+ throws IOException, SQLException, ClassNotFoundException {
+ jdbcSink.setTypeInfo(
+ new SeaTunnelRowType(
+ new String[] {"id"}, new SeaTunnelDataType<?>[]
{BasicType.INT_TYPE}));
+ jdbcSink.prepare(config);
+ JdbcSinkWriter jdbcSinkWriter = (JdbcSinkWriter)
jdbcSink.createWriter(null);
+ JdbcConnectionProvider connectionProvider =
+ (JdbcConnectionProvider) getFieldValue(jdbcSinkWriter,
"connectionProvider");
+ ConnectionImpl connection = (ConnectionImpl)
connectionProvider.getOrEstablishConnection();
+ Properties connectionProperties = connection.getProperties();
+ return connectionProperties;
+ }
+
+ private Properties getSourceProperties(JdbcSource jdbcSource, Config
config)
+ throws IOException, SQLException, ClassNotFoundException {
+ jdbcSource.prepare(config);
+ JdbcConnectionProvider connectionProvider =
+ (JdbcConnectionProvider) getFieldValue(jdbcSource,
"jdbcConnectionProvider");
+ ConnectionImpl connection = (ConnectionImpl)
connectionProvider.getOrEstablishConnection();
+ Properties connectionProperties = connection.getProperties();
+ return connectionProperties;
+ }
+
+ private static Object getFieldValue(Object object, String name) {
+ Class objClass = object.getClass();
+ Field[] fields = objClass.getDeclaredFields();
+ for (Field field : fields) {
+ try {
+ String fieldName = field.getName();
+ if (fieldName.equalsIgnoreCase(name)) {
+ field.setAccessible(true);
+ return field.get(object);
+ }
+ } catch (SecurityException e) {
+
+ } catch (IllegalAccessException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ return null;
+ }
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
