This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 5a87a77a52 [Fix][Connector-V2] Forward SQL Server JDBC URL properties
(#11674)
5a87a77a52 is described below
commit 5a87a77a52bb5b3184dc006ea36158f1be865d8a
Author: Jast <[email protected]>
AuthorDate: Sun Aug 23 09:16:38 2026 +0800
[Fix][Connector-V2] Forward SQL Server JDBC URL properties (#11674)
Co-authored-by: zhangshenghang <[email protected]>
---
docs/en/connectors/source/SqlServer-CDC.md | 2 +-
docs/zh/connectors/source/SqlServer-CDC.md | 2 +-
.../config/SqlServerSourceConfigFactory.java | 18 ++++++
.../source/SqlServerIncrementalSource.java | 40 +++++++++++++
.../SqlServerIncrementalSourceFactoryTest.java | 65 ++++++++++++++++++++++
5 files changed, 125 insertions(+), 2 deletions(-)
diff --git a/docs/en/connectors/source/SqlServer-CDC.md
b/docs/en/connectors/source/SqlServer-CDC.md
index bda3c1005f..9819dac8e7 100644
--- a/docs/en/connectors/source/SqlServer-CDC.md
+++ b/docs/en/connectors/source/SqlServer-CDC.md
@@ -81,7 +81,7 @@ case-sensitive databases, make sure the configured identifier
case matches the d
| table-names | List | Yes | - |
Tables to monitor. Use the full table identifier:
`databaseName.schemaName.tableName`. This option is mutually exclusive with
`table-pattern`.
[...]
| table-pattern | String | Yes | - |
Regular expression for tables to monitor. Use the full table identifier in the
pattern, for example `column_type_test\\.dbo\\..*`. This option is mutually
exclusive with `table-names`.
[...]
| table-names-config | List | No | - |
Table config list. for example: [{"table": "db1.schema1.table1","primaryKeys":
["key1"],"snapshotSplitColumn": "key2"}]. The snapshotSplitColumn option must
be configured with a unique key. If a non-unique column is provided, the
configuration is ignored and SeaTunnel automatically selects an appropriate
split column internally.
[...]
-| url | String | Yes | - |
URL has to be with database, like
"jdbc:sqlserver://localhost:1433;databaseName=test".
[...]
+| url | String | Yes | - |
URL has to be with database, like
"jdbc:sqlserver://localhost:1433;databaseName=test". SQL Server JDBC connection
properties in this URL, such as `encrypt` and `trustServerCertificate`, are
forwarded to the CDC connection. An explicitly configured property in
`debezium`, such as `database.trustServerCertificate`, takes precedence. |
| startup.mode | Enum | No | INITIAL |
Optional startup mode for SqlServer CDC consumer. Valid values are `initial`,
`earliest`, `latest`, and `timestamp`.<br/>`initial`: read the table snapshot
first and then continue with incremental changes.<br/>`earliest`: start from
the earliest available CDC LSN.<br/>`latest`: only read changes after the job
starts.<br/>`timestamp`: start from the LSN resolved from `startup.timestamp`.
[...]
| startup.timestamp | Long | No | - |
Start from the specified epoch timestamp (in milliseconds). This timestamp is
converted with `server-time-zone` when `startup.mode = timestamp`.<br/> **Note,
This option is required when** the **"startup.mode" option used
`'timestamp'`.**
[...]
| stop.mode | Enum | No | NEVER |
Optional stop mode for SqlServer CDC consumer, valid enumerations are "never".
[...]
diff --git a/docs/zh/connectors/source/SqlServer-CDC.md
b/docs/zh/connectors/source/SqlServer-CDC.md
index f321463460..f98fa9118a 100644
--- a/docs/zh/connectors/source/SqlServer-CDC.md
+++ b/docs/zh/connectors/source/SqlServer-CDC.md
@@ -79,7 +79,7 @@ Sql Server CDC 连接器允许从 SqlServer 数据库读取快照数据和增量
| table-names | List | 是 | -
| 要监控的表，使用完整表标识：`databaseName.schemaName.tableName`。该参数与 `table-pattern` 互斥。
|
| table-pattern | String | 是 | -
| 要监控的表名正则表达式，需要匹配完整表标识，例如 `column_type_test\\.dbo\\..*`。该参数与 `table-names`
互斥。
|
| table-names-config | List | 否 | -
| 表配置列表。例如：[{"table": "db1.schema1.table1","primaryKeys":
["key1"],"snapshotSplitColumn": "key2"}]。 snapshotSplitColumn
选项必须配置为唯一键(主键或唯一索引)。 如果指定了非唯一列，该配置将被忽略，SeaTunnel 会在内部自动选择合适的拆分列。
|
-| url | String | 是 | -
| URL 必须包含数据库，如 "jdbc:sqlserver://localhost:1433;databaseName=test"。
|
+| url | String | 是 | -
| URL 必须包含数据库，如 "jdbc:sqlserver://localhost:1433;databaseName=test"。URL 中的 SQL
Server JDBC 连接属性（如 `encrypt`、`trustServerCertificate`）会传递给 CDC 连接；`debezium`
中显式配置的属性（如 `database.trustServerCertificate`）优先。 |
| startup.mode | Enum | 否 |
INITIAL | SqlServer CDC 消费者的可选启动模式，有效值为 `initial`、`earliest`、`latest` 和
`timestamp`。<br/>`initial`：先读取表快照，再继续读取增量变更。<br/>`earliest`：从最早可用的 CDC LSN
开始读取。<br/>`latest`：只读取作业启动后的变更。<br/>`timestamp`：从 `startup.timestamp` 解析出的 LSN
开始读取。
|
| startup.timestamp | Long | 否 | -
| 从指定的纪元时间戳（以毫秒为单位）开始。当 `startup.mode = timestamp` 时，该时间戳会按 `server-time-zone`
转换。<br/> **注意，当 "startup.mode" 选项使用 `'timestamp'` 时，此选项是必需的。**
|
| stop.mode | Enum | 否 | NEVER
| SqlServer CDC 消费者的可选停止模式，有效枚举为 "never"。
|
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/config/SqlServerSourceConfigFactory.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/config/SqlServerSourceConfigFactory.java
index 1f57099abe..cbc99d8d30 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/config/SqlServerSourceConfigFactory.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/config/SqlServerSourceConfigFactory.java
@@ -37,6 +37,20 @@ public class SqlServerSourceConfigFactory extends
JdbcSourceConfigFactory {
private static final String DRIVER_CLASS_NAME =
"com.microsoft.sqlserver.jdbc.SQLServerDriver";
public static final String SCHEMA_CHANGE_KEY = "include.schema.changes";
+ // Holds JDBC properties forwarded from the connector URL; applied before
dbzProperties in
+ // create() so an explicit debezium.* setting still wins over URL-derived
ones.
+ private Properties jdbcUrlProperties;
+
+ /**
+ * Adds SQL Server JDBC properties parsed from the connector URL.
+ *
+ * <p>Explicit Debezium properties are applied afterwards and therefore
take precedence.
+ */
+ public SqlServerSourceConfigFactory jdbcUrlProperties(Properties
jdbcUrlProperties) {
+ this.jdbcUrlProperties = jdbcUrlProperties;
+ return this;
+ }
+
@Override
public SqlServerSourceConfig create(int subtask) {
Properties props = new Properties();
@@ -76,6 +90,10 @@ public class SqlServerSourceConfigFactory extends
JdbcSourceConfigFactory {
props.setProperty("table.include.list", tableIncludeList);
}
+ if (jdbcUrlProperties != null) {
+ jdbcUrlProperties.forEach(props::put);
+ }
+
if (dbzProperties != null) {
dbzProperties.forEach(props::put);
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSource.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSource.java
index 2945d26777..c744dd0730 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSource.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSource.java
@@ -56,6 +56,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Properties;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -93,9 +94,48 @@ public class SqlServerIncrementalSource<T> extends
IncrementalSource<T, JdbcSour
configFactory.originUrl(urlInfo.getOrigin());
configFactory.hostname(urlInfo.getHost());
configFactory.port(urlInfo.getPort());
+
configFactory.jdbcUrlProperties(toDebeziumJdbcProperties(urlInfo.getSuffix()));
return configFactory;
}
+ /**
+ * Converts SQL Server JDBC URL properties to Debezium connection
properties.
+ *
+ * <p>The Debezium SQL Server connector rebuilds its JDBC connection from
{@code database.*}
+ * properties. Carrying the URL suffix into that namespace preserves
driver options such as
+ * {@code encrypt} and {@code trustServerCertificate}.
+ */
+ static Properties toDebeziumJdbcProperties(String suffix) {
+ Properties properties = new Properties();
+ for (String property : suffix.split(";")) {
+ int separator = property.indexOf('=');
+ if (separator <= 0) {
+ continue;
+ }
+ String key = property.substring(0, separator).trim();
+ if (isConnectionAddressProperty(key)) {
+ continue;
+ }
+ properties.setProperty("database." + key,
property.substring(separator + 1).trim());
+ }
+ return properties;
+ }
+
+ private static boolean isConnectionAddressProperty(String key) {
+ // Exclude URL properties that are identity-related: the SQL Server
JDBC driver accepts
+ // these inside the connection-string but SeaTunnel sources them from
dedicated options
+ // (URL/host/port/username/password). Forwarding them here would
silently override the
+ // operator-configured values and create a credential-handling
surprise.
+ return "database".equalsIgnoreCase(key)
+ || "databaseName".equalsIgnoreCase(key)
+ || "serverName".equalsIgnoreCase(key)
+ || "port".equalsIgnoreCase(key)
+ || "portNumber".equalsIgnoreCase(key)
+ || "instanceName".equalsIgnoreCase(key)
+ || "user".equalsIgnoreCase(key)
+ || "password".equalsIgnoreCase(key);
+ }
+
@SuppressWarnings("unchecked")
@Override
public DebeziumDeserializationSchema<T>
createDebeziumDeserializationSchema(
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSourceFactoryTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSourceFactoryTest.java
index e814f76ee5..bdf26b9718 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSourceFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSourceFactoryTest.java
@@ -20,6 +20,8 @@ package
org.apache.seatunnel.connectors.seatunnel.cdc.sqlserver.source;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.table.catalog.Catalog;
import org.apache.seatunnel.api.table.factory.FactoryUtil;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.sqlserver.config.SqlServerSourceConfig;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.sqlserver.config.SqlServerSourceConfigFactory;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -28,6 +30,7 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
+import java.util.Properties;
class SqlServerIncrementalSourceFactoryTest {
@Test
@@ -58,4 +61,66 @@ class SqlServerIncrementalSourceFactoryTest {
Assertions.assertTrue(catalog.isPresent());
catalog.ifPresent(Catalog::close);
}
+
+ @Test
+ public void testForwardJdbcUrlPropertiesToDebeziumConnection() {
+ Properties debeziumProperties = new Properties();
+ debeziumProperties.setProperty("database.encrypt", "false");
+ SqlServerSourceConfigFactory configFactory =
+ (SqlServerSourceConfigFactory)
+ new SqlServerSourceConfigFactory()
+ .hostname("localhost")
+ .port(1433)
+ .username("sa")
+ .password("Password!")
+ .databaseList("seatunnel");
+ configFactory.jdbcUrlProperties(
+ SqlServerIncrementalSource.toDebeziumJdbcProperties(
+
"databaseName=seatunnel;encrypt=true;trustServerCertificate=true"));
+ configFactory.debeziumProperties(debeziumProperties);
+ SqlServerSourceConfig sourceConfig = configFactory.create(0);
+
+ Assertions.assertEquals(
+ "false",
sourceConfig.getDbzConfiguration().getString("database.encrypt"));
+ Assertions.assertEquals(
+ "true",
+
sourceConfig.getDbzConfiguration().getString("database.trustServerCertificate"));
+ Assertions.assertNull(
+
sourceConfig.getDbzConfiguration().getString("database.databaseName"));
+ }
+
+ /**
+ * URL-embedded {@code user} / {@code password} segments must never
override the
+ * operator-configured {@code username} / {@code password} options — doing
so would silently
+ * change the credentials the job connects with.
+ */
+ @Test
+ public void testJdbcUrlCredentialsDoNotOverrideConfiguredCredentials() {
+ SqlServerSourceConfigFactory configFactory =
+ (SqlServerSourceConfigFactory)
+ new SqlServerSourceConfigFactory()
+ .hostname("localhost")
+ .port(1433)
+ .username("configured-user")
+ .password("configured-password")
+ .databaseList("seatunnel");
+ configFactory.jdbcUrlProperties(
+ SqlServerIncrementalSource.toDebeziumJdbcProperties(
+ "user=url-user;password=url-password;encrypt=true"));
+ SqlServerSourceConfig sourceConfig = configFactory.create(0);
+
+ // The configured username/password must win — URL-embedded
user/password must be
+ // filtered out before they are forwarded into the Debezium Properties
bag.
+ Assertions.assertEquals(
+ "configured-user",
sourceConfig.getDbzConfiguration().getString("database.user"));
+ Assertions.assertEquals(
+ "configured-password",
+
sourceConfig.getDbzConfiguration().getString("database.password"));
+
Assertions.assertNull(sourceConfig.getDbzConfiguration().getString("database.url-user"));
+ Assertions.assertNull(
+
sourceConfig.getDbzConfiguration().getString("database.url-password"));
+ // Forwarding of unrelated URL properties is still in effect.
+ Assertions.assertEquals(
+ "true",
sourceConfig.getDbzConfiguration().getString("database.encrypt"));
+ }
}
