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-12931-07f5011b8506d0bfeca1a23df315a08f45cb3e33 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 22a34f323d825b44fc8219b1eb343a8ee7fe3535 Author: Nikhil kumar <[email protected]> AuthorDate: Wed Oct 28 17:16:33 2026 +0000 [Improve][Connector-V2] Validate DB2 CDC source URL (#12931) Co-authored-by: Nikhil <[email protected]> Co-authored-by: zhiweiniu <[email protected]> --- docs/en/connectors/source/DB2-CDC.md | 2 +- docs/zh/connectors/source/DB2-CDC.md | 2 +- .../db2/source/Db2IncrementalSourceFactory.java | 8 +- .../source/Db2IncrementalSourceFactoryTest.java | 103 +++++++++++++++++++++ 4 files changed, 110 insertions(+), 5 deletions(-) diff --git a/docs/en/connectors/source/DB2-CDC.md b/docs/en/connectors/source/DB2-CDC.md index a4b0b0a07d..8eb7238e04 100644 --- a/docs/en/connectors/source/DB2-CDC.md +++ b/docs/en/connectors/source/DB2-CDC.md @@ -69,7 +69,7 @@ UPDATE and DELETE changes after the initial snapshot is finished. |------|------|----------|---------|-------------| | username | String | Yes | - | User name used to connect to DB2. | | password | String | Yes | - | Password used to connect to DB2. | -| url | String | Yes | - | DB2 JDBC URL. The URL must include a database name, for example `jdbc:db2://127.0.0.1:50000/testdb`. | +| url | String | Yes | - | DB2 JDBC URL. The value must not be empty or whitespace-only. The URL must include a database name, for example `jdbc:db2://127.0.0.1:50000/testdb`. | | database-names | List | No | The database parsed from `url` | Database name to monitor. DB2 CDC captures one database in one source. | | table-names | List | Yes when `table-pattern` is not set | - | Table names to monitor, using `databaseName.schemaName.tableName`, for example `testdb.DB2INST1.CUSTOMERS`. | | table-pattern | String | Yes when `table-names` is not set | - | Regular expression used to discover captured tables. | diff --git a/docs/zh/connectors/source/DB2-CDC.md b/docs/zh/connectors/source/DB2-CDC.md index 2054bedd6e..aa720a889a 100644 --- a/docs/zh/connectors/source/DB2-CDC.md +++ b/docs/zh/connectors/source/DB2-CDC.md @@ -68,7 +68,7 @@ Debezium DB2，在初始快照完成后会继续读取已提交的 INSERT、UPDA |------|------|----------|--------|------| | username | String | 是 | - | 连接 DB2 时使用的用户名。 | | password | String | 是 | - | 连接 DB2 时使用的密码。 | -| url | String | 是 | - | DB2 JDBC URL。URL 必须包含数据库名，例如 `jdbc:db2://127.0.0.1:50000/testdb`。 | +| url | String | 是 | - | DB2 JDBC URL。该值不能为空或仅包含空白字符。URL 必须包含数据库名，例如 `jdbc:db2://127.0.0.1:50000/testdb`。 | | database-names | List | 否 | 从 `url` 中解析出的数据库 | 要监控的数据库名。DB2 CDC 一个 source 监控一个数据库。 | | table-names | List | 未设置 `table-pattern` 时必填 | - | 要监控的表名，格式为 `databaseName.schemaName.tableName`，例如 `testdb.DB2INST1.CUSTOMERS`。 | | table-pattern | String | 未设置 `table-names` 时必填 | - | 用于发现已开启 capture mode 的表的正则表达式。 | diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-db2/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/db2/source/Db2IncrementalSourceFactory.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-db2/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/db2/source/Db2IncrementalSourceFactory.java index cc0ba5509e..51f4490600 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-db2/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/db2/source/Db2IncrementalSourceFactory.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-db2/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/db2/source/Db2IncrementalSourceFactory.java @@ -41,6 +41,8 @@ import java.io.Serializable; import java.util.List; import java.util.Optional; +import static org.apache.seatunnel.api.configuration.util.Conditions.notBlank; + @AutoService(Factory.class) @Slf4j public class Db2IncrementalSourceFactory implements TableSourceFactory { @@ -56,9 +58,9 @@ public class Db2IncrementalSourceFactory implements TableSourceFactory { public OptionRule optionRule() { return Db2IncrementalSourceOptions.getBaseRule() .required( - Db2IncrementalSourceOptions.USERNAME, - Db2IncrementalSourceOptions.PASSWORD, - Db2IncrementalSourceOptions.URL) + Db2IncrementalSourceOptions.USERNAME, Db2IncrementalSourceOptions.PASSWORD) + .required( + Db2IncrementalSourceOptions.URL, notBlank(Db2IncrementalSourceOptions.URL)) .exclusive(ConnectorCommonOptions.TABLE_NAMES, ConnectorCommonOptions.TABLE_PATTERN) .optional( Db2IncrementalSourceOptions.DATABASE_NAMES, diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-db2/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/db2/source/Db2IncrementalSourceFactoryTest.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-db2/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/db2/source/Db2IncrementalSourceFactoryTest.java index a977cb58a1..721cf79aad 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-db2/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/db2/source/Db2IncrementalSourceFactoryTest.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-db2/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/db2/source/Db2IncrementalSourceFactoryTest.java @@ -17,24 +17,127 @@ package org.apache.seatunnel.connectors.seatunnel.cdc.db2.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.TablePath; import org.apache.seatunnel.connectors.seatunnel.cdc.db2.config.Db2SourceConfig; import org.apache.seatunnel.connectors.seatunnel.cdc.db2.config.Db2SourceConfigFactory; 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.mockito.Mockito; import io.debezium.relational.TableId; import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.stream.Stream; class Db2IncrementalSourceFactoryTest { + private static final String VALID_URL = "jdbc:db2://localhost:50000/seatunnel"; + + private final OptionRule optionRule = new Db2IncrementalSourceFactory().optionRule(); + @Test public void testOptionRule() { Assertions.assertNotNull((new Db2IncrementalSourceFactory()).optionRule()); } + @ParameterizedTest + @MethodSource("nonblankUrls") + void testNonblankUrlAcceptedWithoutChangingValue(String key, String value) { + Map<String, Object> config = validConfig(); + config.remove("url"); + config.put(key, value); + + ReadonlyConfig readonlyConfig = Assertions.assertDoesNotThrow(() -> validate(config)); + + // Validation does not trim a supplied URL or check whether it can connect. + Assertions.assertEquals(value, readonlyConfig.get(Db2IncrementalSourceOptions.URL)); + } + + private static Stream<Arguments> nonblankUrls() { + return Stream.of("url", "base-url") + .flatMap( + key -> + Stream.of(VALID_URL, " \t" + VALID_URL + " \r\n") + .map(value -> Arguments.of(key, value))); + } + + @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", " \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", " "); + + ReadonlyConfig readonlyConfig = Assertions.assertDoesNotThrow(() -> validate(config)); + + Assertions.assertEquals(VALID_URL, readonlyConfig.get(Db2IncrementalSourceOptions.URL)); + } + + @Test + void testBlankCanonicalUrlDoesNotUseValidFallback() { + Map<String, Object> config = validConfig(); + config.put("url", " \t "); + config.put("base-url", VALID_URL); + + assertInvalidUrl(config); + } + + private Map<String, Object> validConfig() { + Map<String, Object> config = new HashMap<>(); + config.put("url", VALID_URL); + config.put("username", "test-user"); + config.put("password", "test-password"); + config.put("table-names", Collections.singletonList("seatunnel.DB2INST1.CUSTOMERS")); + return config; + } + + private ReadonlyConfig validate(Map<String, Object> config) { + ReadonlyConfig readonlyConfig = ReadonlyConfig.fromMap(config); + ConfigValidator.validateUnknownKeys(readonlyConfig, optionRule, "DB2-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()); + } + /** Verifies Db2 checkpoint tables use the empty-catalog identifier returned by discovery. */ @Test public void testToTableIdDropsCatalog() {
