This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 51f553cb45 [Improve][Connector-V2] Validate nonblank SQL Server CDC
source URL (#12764)
51f553cb45 is described below
commit 51f553cb45b4b2b37b87f7f654352bc0276daf6b
Author: yuzhou <[email protected]>
AuthorDate: Fri Oct 16 17:31:20 2026 +0000
[Improve][Connector-V2] Validate nonblank SQL Server CDC source URL (#12764)
---
docs/en/connectors/source/SqlServer-CDC.md | 2 +-
docs/zh/connectors/source/SqlServer-CDC.md | 2 +-
.../source/SqlServerIncrementalSourceFactory.java | 7 +-
.../SqlServerIncrementalSourceFactoryTest.java | 127 +++++++++++++++++++++
.../connector/cdc/sqlserver/SqlServerCDCIT.java | 29 +++++
.../src/test/resources/sqlservercdc_blank_url.conf | 35 ++++++
6 files changed, 198 insertions(+), 4 deletions(-)
diff --git a/docs/en/connectors/source/SqlServer-CDC.md
b/docs/en/connectors/source/SqlServer-CDC.md
index 40ca9315dd..cafa736078 100644
--- a/docs/en/connectors/source/SqlServer-CDC.md
+++ b/docs/en/connectors/source/SqlServer-CDC.md
@@ -85,7 +85,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". 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. |
+| url | String | Yes | - |
Empty or whitespace-only values are rejected during configuration validation.
Nonblank values are not trimmed. 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.trustServerCe [...]
| 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 f98fa9118a..ec9c4c0cfe 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 中的 SQL
Server JDBC 连接属性（如 `encrypt`、`trustServerCertificate`）会传递给 CDC 连接；`debezium`
中显式配置的属性（如 `database.trustServerCertificate`）优先。 |
+| 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/source/SqlServerIncrementalSourceFactory.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSourceFactory.java
index 02e40917b7..006d714468 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSourceFactory.java
@@ -17,6 +17,7 @@
package org.apache.seatunnel.connectors.seatunnel.cdc.sqlserver.source;
+import org.apache.seatunnel.api.configuration.util.Conditions;
import org.apache.seatunnel.api.configuration.util.OptionRule;
import org.apache.seatunnel.api.options.ConnectorCommonOptions;
import org.apache.seatunnel.api.source.SeaTunnelSource;
@@ -55,8 +56,10 @@ public class SqlServerIncrementalSourceFactory implements
TableSourceFactory {
return SqlServerIncrementalSourceOptions.getBaseRule()
.required(
SqlServerIncrementalSourceOptions.USERNAME,
- SqlServerIncrementalSourceOptions.PASSWORD,
- SqlServerIncrementalSourceOptions.URL)
+ SqlServerIncrementalSourceOptions.PASSWORD)
+ .required(
+ SqlServerIncrementalSourceOptions.URL,
+
Conditions.notBlank(SqlServerIncrementalSourceOptions.URL))
.exclusive(ConnectorCommonOptions.TABLE_NAMES,
ConnectorCommonOptions.TABLE_PATTERN)
.optional(
SqlServerIncrementalSourceOptions.DATABASE_NAMES,
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 bdf26b9718..4818685dd8 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
@@ -18,6 +18,9 @@
package org.apache.seatunnel.connectors.seatunnel.cdc.sqlserver.source;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConfigValidator;
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
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;
@@ -25,19 +28,143 @@ import
org.apache.seatunnel.connectors.seatunnel.cdc.sqlserver.config.SqlServerS
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
+import java.util.stream.Stream;
class SqlServerIncrementalSourceFactoryTest {
+ private static final String VALID_URL =
+ "jdbc:sqlserver://localhost:1433;databaseName=seatunnel";
+
+ private final OptionRule optionRule = new
SqlServerIncrementalSourceFactory().optionRule();
+
@Test
public void testOptionRule() {
Assertions.assertNotNull((new
SqlServerIncrementalSourceFactory()).optionRule());
}
+ @ParameterizedTest
+ @ValueSource(strings = {"url", "base-url"})
+ void testNonblankUrlAccepted(String key) {
+ Map<String, Object> config = validConfig();
+ config.remove("url");
+ config.put(key, VALID_URL);
+
+ ReadonlyConfig readonlyConfig = Assertions.assertDoesNotThrow(() ->
validate(config));
+
+ Assertions.assertEquals(
+ VALID_URL,
readonlyConfig.get(SqlServerIncrementalSourceOptions.URL));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"url", "base-url"})
+ void testPaddedNonblankUrlAcceptedWithoutTrimming(String key) {
+ Map<String, Object> config = validConfig();
+ config.remove("url");
+ String value = " \t" + VALID_URL + " \r\n";
+ config.put(key, value);
+
+ ReadonlyConfig readonlyConfig = Assertions.assertDoesNotThrow(() ->
validate(config));
+
+ Assertions.assertEquals(value,
readonlyConfig.get(SqlServerIncrementalSourceOptions.URL));
+ }
+
+ @ParameterizedTest
+ @MethodSource("blankUrls")
+ void testBlankUrlRejected(String key, String value) {
+ Map<String, Object> config = validConfig();
+ config.remove("url");
+ config.put(key, value);
+
+ assertInvalidUrl(config);
+ }
+
+ private static Stream<Arguments> blankUrls() {
+ return Stream.of("url", "base-url")
+ .flatMap(
+ key ->
+ Stream.of("", " ", "\t", "\n", "\r", " \t\r\n
")
+ .map(value -> Arguments.of(key,
value)));
+ }
+
+ @Test
+ void testMissingUrlRejected() {
+ Map<String, Object> config = validConfig();
+ config.remove("url");
+
+ assertInvalidUrl(config);
+ }
+
+ @Test
+ void testCanonicalUrlTakesPrecedenceOverFallback() {
+ Map<String, Object> config = validConfig();
+ config.put("base-url",
"jdbc:sqlserver://other-host:1433;databaseName=other");
+
+ ReadonlyConfig readonlyConfig = Assertions.assertDoesNotThrow(() ->
validate(config));
+
+ Assertions.assertEquals(
+ VALID_URL,
readonlyConfig.get(SqlServerIncrementalSourceOptions.URL));
+ }
+
+ @Test
+ void testValidCanonicalUrlIgnoresBlankFallback() {
+ Map<String, Object> config = validConfig();
+ config.put("base-url", " \t ");
+
+ Assertions.assertDoesNotThrow(() -> validate(config));
+ }
+
+ @Test
+ void testBlankCanonicalUrlDoesNotUseValidFallback() {
+ Map<String, Object> config = validConfig();
+ config.put("url", " \t ");
+ config.put("base-url", VALID_URL);
+
+ assertInvalidUrl(config);
+ }
+
+ @Test
+ void testEmptyPasswordRemainsAccepted() {
+ Map<String, Object> config = validConfig();
+ config.put("password", "");
+
+ ReadonlyConfig readonlyConfig = Assertions.assertDoesNotThrow(() ->
validate(config));
+
+ Assertions.assertEquals("",
readonlyConfig.get(SqlServerIncrementalSourceOptions.PASSWORD));
+ }
+
+ private Map<String, Object> validConfig() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("url", VALID_URL);
+ config.put("username", "sa");
+ config.put("password", "Password!");
+ config.put("table-names",
Collections.singletonList("seatunnel.dbo.orders"));
+ return config;
+ }
+
+ private ReadonlyConfig validate(Map<String, Object> config) {
+ ReadonlyConfig readonlyConfig = ReadonlyConfig.fromMap(config);
+ ConfigValidator.validateUnknownKeys(readonlyConfig, optionRule,
"SqlServer-CDC");
+ ConfigValidator.of(readonlyConfig).validate(optionRule);
+ return readonlyConfig;
+ }
+
+ private void assertInvalidUrl(Map<String, Object> config) {
+ OptionValidationException error =
+ Assertions.assertThrows(OptionValidationException.class, () ->
validate(config));
+
+ Assertions.assertTrue(error.getMessage().contains("url"),
error.getMessage());
+ }
+
/**
* SQLServer CDC source creation must accept the driver-specific
databaseName URL syntax during
* the submission-time catalog validation step.
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/java/org/apache/seatunnel/e2e/connector/cdc/sqlserver/SqlServerCDCIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/java/org/apache/seatunnel/e2e/connector/cdc/sqlserver/SqlServerCDCIT.java
index bfd1bc6ae3..5b5bb93da8 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/java/org/apache/seatunnel/e2e/connector/cdc/sqlserver/SqlServerCDCIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/java/org/apache/seatunnel/e2e/connector/cdc/sqlserver/SqlServerCDCIT.java
@@ -28,6 +28,7 @@ import org.apache.seatunnel.e2e.common.container.EngineType;
import org.apache.seatunnel.e2e.common.container.TestContainer;
import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
import org.apache.seatunnel.e2e.common.junit.TestContainerExtension;
+import org.apache.seatunnel.e2e.common.util.ContainerUtil;
import org.apache.seatunnel.e2e.common.util.DependencyJar;
import org.apache.seatunnel.e2e.common.util.JdbcUtil;
import org.apache.seatunnel.e2e.common.util.JobIdGenerator;
@@ -258,6 +259,34 @@ public class SqlServerCDCIT extends TestSuiteBase
implements TestResource {
});
}
+ @TestTemplate
+ @DisabledOnContainer(
+ value = {},
+ type = {EngineType.FLINK},
+ disabledReason = "The configuration-only --check command is
provided by Zeta")
+ public void testBlankSourceUrlRejectedByConfigCheck(TestContainer
container)
+ throws IOException, InterruptedException {
+ String configResource = "/sqlservercdc_blank_url.conf";
+ container.executeExtraCommands(
+ engineContainer ->
+ ContainerUtil.copyConnectorJarToContainer(
+ engineContainer,
+ configResource,
+ "seatunnel-connectors-v2",
+ "connector-",
+ "seatunnel",
+ "/tmp/seatunnel"));
+ String configPath =
"/tmp/seatunnel/config/sqlservercdc_blank_url.conf";
+ container.copyFileToContainer(configResource, configPath);
+
+ Container.ExecResult result =
+ container.executeBaseCommand(new String[] {"--config",
configPath, "--check"});
+ String output = result.getStdout() + result.getStderr();
+
+ Assertions.assertNotEquals(0, result.getExitCode(), output);
+ Assertions.assertTrue(output.contains("url") &&
output.contains("blank"), output);
+ }
+
/**
* Verifies that a single SqlServer CDC source can capture multiple tables
and route them to
* different sink tables in the same database.
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/sqlservercdc_blank_url.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/sqlservercdc_blank_url.conf
new file mode 100644
index 0000000000..993e9b675b
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/sqlservercdc_blank_url.conf
@@ -0,0 +1,35 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+}
+
+source {
+ SqlServer-CDC {
+ username = "sa"
+ password = "Password!"
+ table-names = ["column_type_test.dbo.full_types"]
+ url = " \t\r\n "
+ }
+}
+
+sink {
+ Console {
+ }
+}
