Skip to content

Commit f85eb78

Browse files
Carl-Zhou-CNCarl-Zhou-CN
authored
[Fix][JDBC] fix jdbc default connection parameter invalid (#8185)
1 parent f41e882 commit f85eb78

File tree

11 files changed

+249
-60
lines changed

11 files changed

+249
-60
lines changed

‎seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/ImportClassCheckTest.java

+8
Original file line numberDiff line numberDiff line change
@@ -112,6 +112,14 @@ public void jettyShadeCheck() {
112112
log.info("check jetty shade successfully");
113113
}
114114

115+
@Test
116+
public void hikariShadeCheck() {
117+
Map<String, List<String>> errorMap =
118+
checkImportClassPrefixWithAll(Collections.singletonList("com.zaxxer.hikari"));
119+
Assertions.assertEquals(0, errorMap.size(), shadeErrorMsg("hikari", errorMap));
120+
log.info("check hikari shade successfully");
121+
}
122+
115123
@Test
116124
public void janinoShadeCheck() {
117125
Map<String, List<String>> errorMap =

‎seatunnel-connectors-v2/connector-cdc/connector-cdc-base/pom.xml

+7-10
Original file line numberDiff line numberDiff line change
@@ -34,12 +34,6 @@
3434

3535
<dependencyManagement>
3636
<dependencies>
37-
<dependency>
38-
<groupId>com.zaxxer</groupId>
39-
<artifactId>HikariCP</artifactId>
40-
<version>${hikaricp.version}</version>
41-
</dependency>
42-
4337
<dependency>
4438
<groupId>org.apache.seatunnel</groupId>
4539
<artifactId>connector-common</artifactId>
@@ -98,10 +92,6 @@
9892
<groupId>com.github.luben</groupId>
9993
<artifactId>zstd-jni</artifactId>
10094
</dependency>
101-
<dependency>
102-
<groupId>com.zaxxer</groupId>
103-
<artifactId>HikariCP</artifactId>
104-
</dependency>
10595
<dependency>
10696
<groupId>org.apache.seatunnel</groupId>
10797
<artifactId>connector-common</artifactId>
@@ -116,6 +106,13 @@
116106
<version>${commons-lang3.version}</version>
117107
</dependency>
118108

109+
<dependency>
110+
<groupId>org.apache.seatunnel</groupId>
111+
<artifactId>seatunnel-hikari</artifactId>
112+
<version>${project.version}</version>
113+
<classifier>optional</classifier>
114+
</dependency>
115+
119116
<dependency>
120117
<groupId>junit</groupId>
121118
<artifactId>junit</artifactId>

‎seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/relational/connection/JdbcConnectionFactory.java

+2-1
Original file line numberDiff line numberDiff line change
@@ -17,13 +17,14 @@
1717

1818
package org.apache.seatunnel.connectors.cdc.base.relational.connection;
1919

20+
import org.apache.seatunnel.shade.com.zaxxer.hikari.HikariDataSource;
21+
2022
import org.apache.seatunnel.common.utils.SeaTunnelException;
2123
import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig;
2224

2325
import org.slf4j.Logger;
2426
import org.slf4j.LoggerFactory;
2527

26-
import com.zaxxer.hikari.HikariDataSource;
2728
import io.debezium.jdbc.JdbcConfiguration;
2829
import io.debezium.jdbc.JdbcConnection;
2930

‎seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/relational/connection/JdbcConnectionPoolFactory.java

+3-3
Original file line numberDiff line numberDiff line change
@@ -17,10 +17,10 @@
1717

1818
package org.apache.seatunnel.connectors.cdc.base.relational.connection;
1919

20-
import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig;
20+
import org.apache.seatunnel.shade.com.zaxxer.hikari.HikariConfig;
21+
import org.apache.seatunnel.shade.com.zaxxer.hikari.HikariDataSource;
2122

22-
import com.zaxxer.hikari.HikariConfig;
23-
import com.zaxxer.hikari.HikariDataSource;
23+
import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig;
2424

2525
/** A connection pool factory to create pooled DataSource {@link HikariDataSource}. */
2626
public abstract class JdbcConnectionPoolFactory {

‎seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/relational/connection/JdbcConnectionPools.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -17,13 +17,13 @@
1717

1818
package org.apache.seatunnel.connectors.cdc.base.relational.connection;
1919

20+
import org.apache.seatunnel.shade.com.zaxxer.hikari.HikariDataSource;
21+
2022
import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig;
2123

2224
import org.slf4j.Logger;
2325
import org.slf4j.LoggerFactory;
2426

25-
import com.zaxxer.hikari.HikariDataSource;
26-
2727
import java.util.HashMap;
2828
import java.util.Map;
2929

‎seatunnel-connectors-v2/connector-jdbc/pom.xml

+4-42
Original file line numberDiff line numberDiff line change
@@ -240,9 +240,10 @@
240240
</dependency>
241241

242242
<dependency>
243-
<groupId>com.zaxxer</groupId>
244-
<artifactId>HikariCP</artifactId>
245-
<version>${hikari.version}</version>
243+
<groupId>org.apache.seatunnel</groupId>
244+
<artifactId>seatunnel-hikari</artifactId>
245+
<version>${project.version}</version>
246+
<classifier>optional</classifier>
246247
</dependency>
247248

248249
<dependency>
@@ -353,43 +354,4 @@
353354
</dependency>
354355
</dependencies>
355356

356-
<build>
357-
<plugins>
358-
<plugin>
359-
<groupId>org.apache.maven.plugins</groupId>
360-
<artifactId>maven-shade-plugin</artifactId>
361-
<executions>
362-
<execution>
363-
<goals>
364-
<goal>shade</goal>
365-
</goals>
366-
<phase>package</phase>
367-
<configuration>
368-
<createSourcesJar>false</createSourcesJar>
369-
<shadeSourcesContent>true</shadeSourcesContent>
370-
<shadedArtifactAttached>false</shadedArtifactAttached>
371-
<createDependencyReducedPom>false</createDependencyReducedPom>
372-
<filters>
373-
<filter>
374-
<artifact>*:*</artifact>
375-
<excludes>
376-
<exclude>META-INF/*.SF</exclude>
377-
<exclude>META-INF/*.DSA</exclude>
378-
<exclude>META-INF/*.RSA</exclude>
379-
</excludes>
380-
</filter>
381-
</filters>
382-
<relocations>
383-
<!-- rename hikari to avoid jar conflict from spark -->
384-
<relocation>
385-
<pattern>com.zaxxer.hikari</pattern>
386-
<shadedPattern>${seatunnel.shade.package}.com.zaxxer.hikari</shadedPattern>
387-
</relocation>
388-
</relocations>
389-
</configuration>
390-
</execution>
391-
</executions>
392-
</plugin>
393-
</plugins>
394-
</build>
395357
</project>

‎seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/ConnectionPoolManager.java

+2-1
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,8 @@
1717

1818
package org.apache.seatunnel.connectors.seatunnel.jdbc.sink;
1919

20-
import com.zaxxer.hikari.HikariDataSource;
20+
import org.apache.seatunnel.shade.com.zaxxer.hikari.HikariDataSource;
21+
2122
import lombok.Getter;
2223

2324
import java.sql.Connection;

‎seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkWriter.java

+3-1
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@
1717

1818
package org.apache.seatunnel.connectors.seatunnel.jdbc.sink;
1919

20+
import org.apache.seatunnel.shade.com.zaxxer.hikari.HikariDataSource;
21+
2022
import org.apache.seatunnel.api.sink.MultiTableResourceManager;
2123
import org.apache.seatunnel.api.table.catalog.TablePath;
2224
import org.apache.seatunnel.api.table.catalog.TableSchema;
@@ -31,7 +33,6 @@
3133
import org.apache.seatunnel.connectors.seatunnel.jdbc.state.JdbcSinkState;
3234
import org.apache.seatunnel.connectors.seatunnel.jdbc.state.XidInfo;
3335

34-
import com.zaxxer.hikari.HikariDataSource;
3536
import lombok.extern.slf4j.Slf4j;
3637

3738
import java.io.IOException;
@@ -88,6 +89,7 @@ public MultiTableResourceManager<ConnectionPoolManager> initMultiTableResourceMa
8889
ds.setPassword(jdbcSinkConfig.getJdbcConnectionConfig().getPassword().get());
8990
}
9091
ds.setAutoCommit(jdbcSinkConfig.getJdbcConnectionConfig().isAutoCommit());
92+
jdbcSinkConfig.getJdbcConnectionConfig().getProperties().forEach(ds::addDataSourceProperty);
9193
return new JdbcMultiTableResourceManager(new ConnectionPoolManager(ds));
9294
}
9395

‎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

+115
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
package org.apache.seatunnel.connectors.seatunnel.jdbc;
2020

2121
import org.apache.seatunnel.shade.com.google.common.collect.Lists;
22+
import org.apache.seatunnel.shade.com.zaxxer.hikari.pool.HikariProxyConnection;
2223

2324
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
2425
import org.apache.seatunnel.api.table.catalog.CatalogTable;
@@ -33,6 +34,7 @@
3334
import org.apache.seatunnel.common.utils.ReflectionUtils;
3435
import org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.mysql.MySqlCatalog;
3536
import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.connection.JdbcConnectionProvider;
37+
import org.apache.seatunnel.connectors.seatunnel.jdbc.sink.JdbcMultiTableResourceManager;
3638
import org.apache.seatunnel.connectors.seatunnel.jdbc.sink.JdbcSink;
3739
import org.apache.seatunnel.connectors.seatunnel.jdbc.sink.JdbcSinkFactory;
3840
import org.apache.seatunnel.connectors.seatunnel.jdbc.sink.JdbcSinkWriter;
@@ -459,6 +461,7 @@ private String getUrl() {
459461
public void parametersTest() throws Exception {
460462
defaultSinkParametersTest();
461463
defaultSourceParametersTest();
464+
defaultMultiSinkParametersTest();
462465
}
463466

464467
void defaultSinkParametersTest() throws IOException, SQLException, ClassNotFoundException {
@@ -546,6 +549,118 @@ void defaultSinkParametersTest() throws IOException, SQLException, ClassNotFound
546549
Assertions.assertEquals(connectionProperties4.get("rewriteBatchedStatements"), "false");
547550
}
548551

552+
void defaultMultiSinkParametersTest() throws IOException, SQLException, ClassNotFoundException {
553+
TableSchema tableSchema =
554+
TableSchema.builder()
555+
.column(
556+
PhysicalColumn.of(
557+
"c_bigint",
558+
BasicType.LONG_TYPE,
559+
22,
560+
false,
561+
null,
562+
"c_bigint"))
563+
.build();
564+
CatalogTable catalogTable =
565+
CatalogTable.of(
566+
TableIdentifier.of("test_catalog", "seatunnel", "source"),
567+
tableSchema,
568+
new HashMap<>(),
569+
new ArrayList<>(),
570+
"User table");
571+
572+
// case1 url not contains parameters and properties not contains parameters
573+
Map<String, Object> map1 = getDefaultConfigMap();
574+
map1.put("url", getUrl());
575+
ReadonlyConfig config1 = ReadonlyConfig.fromMap(map1);
576+
TableSinkFactoryContext context1 =
577+
TableSinkFactoryContext.replacePlaceholderAndCreate(
578+
catalogTable,
579+
config1,
580+
Thread.currentThread().getContextClassLoader(),
581+
Collections.emptyList());
582+
JdbcSink jdbcSink1 = (JdbcSink) new JdbcSinkFactory().createSink(context1).createSink();
583+
JdbcMultiTableResourceManager multiTableResourceManager1 =
584+
(JdbcMultiTableResourceManager)
585+
jdbcSink1.createWriter(null).initMultiTableResourceManager(1, 1);
586+
Properties connectionProperties1 = getMultiSinkProperties(multiTableResourceManager1);
587+
Assertions.assertEquals(connectionProperties1.get("rewriteBatchedStatements"), "true");
588+
589+
// case2 url contains parameters and properties not contains parameters
590+
Map<String, Object> map2 = getDefaultConfigMap();
591+
map2.put("url", getUrl() + "?rewriteBatchedStatements=false");
592+
ReadonlyConfig config2 = ReadonlyConfig.fromMap(map2);
593+
TableSinkFactoryContext context2 =
594+
TableSinkFactoryContext.replacePlaceholderAndCreate(
595+
catalogTable,
596+
config2,
597+
Thread.currentThread().getContextClassLoader(),
598+
Collections.emptyList());
599+
JdbcSink jdbcSink2 = (JdbcSink) new JdbcSinkFactory().createSink(context2).createSink();
600+
JdbcMultiTableResourceManager multiTableResourceManager2 =
601+
(JdbcMultiTableResourceManager)
602+
jdbcSink2.createWriter(null).initMultiTableResourceManager(1, 1);
603+
Properties connectionProperties2 = getMultiSinkProperties(multiTableResourceManager2);
604+
Assertions.assertEquals(connectionProperties2.get("rewriteBatchedStatements"), "false");
605+
606+
// case3 url not contains parameters and properties not contains parameters
607+
Map<String, Object> map3 = getDefaultConfigMap();
608+
Map<String, String> properties3 = new HashMap<>();
609+
properties3.put("rewriteBatchedStatements", "false");
610+
map3.put("properties", properties3);
611+
map3.put("url", getUrl());
612+
ReadonlyConfig config3 = ReadonlyConfig.fromMap(map3);
613+
TableSinkFactoryContext context3 =
614+
TableSinkFactoryContext.replacePlaceholderAndCreate(
615+
catalogTable,
616+
config3,
617+
Thread.currentThread().getContextClassLoader(),
618+
Collections.emptyList());
619+
JdbcSink jdbcSink3 = (JdbcSink) new JdbcSinkFactory().createSink(context3).createSink();
620+
JdbcMultiTableResourceManager multiTableResourceManager3 =
621+
(JdbcMultiTableResourceManager)
622+
jdbcSink3.createWriter(null).initMultiTableResourceManager(1, 1);
623+
Properties connectionProperties3 = getMultiSinkProperties(multiTableResourceManager3);
624+
Assertions.assertEquals(connectionProperties3.get("rewriteBatchedStatements"), "false");
625+
626+
// case4 url contains parameters and properties contains parameters
627+
Map<String, Object> map4 = getDefaultConfigMap();
628+
Map<String, String> properties4 = new HashMap<>();
629+
properties4.put("useSSL", "true");
630+
properties4.put("rewriteBatchedStatements", "false");
631+
map4.put("properties", properties4);
632+
map4.put("url", getUrl() + "?useSSL=false&rewriteBatchedStatements=true");
633+
ReadonlyConfig config4 = ReadonlyConfig.fromMap(map4);
634+
TableSinkFactoryContext context4 =
635+
TableSinkFactoryContext.replacePlaceholderAndCreate(
636+
catalogTable,
637+
config4,
638+
Thread.currentThread().getContextClassLoader(),
639+
Collections.emptyList());
640+
JdbcSink jdbcSink4 = (JdbcSink) new JdbcSinkFactory().createSink(context4).createSink();
641+
JdbcMultiTableResourceManager multiTableResourceManager4 =
642+
(JdbcMultiTableResourceManager)
643+
jdbcSink4.createWriter(null).initMultiTableResourceManager(1, 1);
644+
Properties connectionProperties4 = getMultiSinkProperties(multiTableResourceManager4);
645+
Assertions.assertEquals(connectionProperties4.get("useSSL"), "true");
646+
Assertions.assertEquals(connectionProperties4.get("rewriteBatchedStatements"), "false");
647+
}
648+
649+
private Properties getMultiSinkProperties(
650+
JdbcMultiTableResourceManager multiTableResourceManager) throws SQLException {
651+
HikariProxyConnection hikariProxyConnection =
652+
(HikariProxyConnection)
653+
multiTableResourceManager
654+
.getSharedResource()
655+
.get()
656+
.getConnectionPool()
657+
.getConnection();
658+
Properties connectionProperties =
659+
((ConnectionImpl) ReflectionUtils.getField(hikariProxyConnection, "delegate").get())
660+
.getProperties();
661+
return connectionProperties;
662+
}
663+
549664
void defaultSourceParametersTest() throws Exception {
550665
// case1 url not contains parameters and properties not contains parameters
551666
Map<String, Object> map1 = getDefaultConfigMap();

‎seatunnel-shade/pom.xml

+1
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
<module>seatunnel-jetty9-9.4.56</module>
3737
<module>seatunnel-hadoop-aws</module>
3838
<module>seatunnel-arrow</module>
39+
<module>seatunnel-hikari</module>
3940
</modules>
4041

4142
<build>

0 commit comments

Comments
 (0)