This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12764-be8a310b251251e5dc76b9d49c488b06e38c0159 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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 { + } +}
