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 8b28ff8e7c [Improve][Connector-V2] Validate SelectDB Cloud required
options (#12351)
8b28ff8e7c is described below
commit 8b28ff8e7c6a5105e033909ba53b152df0289255
Author: Nikhil kumar <[email protected]>
AuthorDate: Tue Sep 29 20:46:57 2026 +0530
[Improve][Connector-V2] Validate SelectDB Cloud required options (#12351)
---
docs/en/connectors/sink/SelectDB-Cloud.md | 14 +--
docs/zh/connectors/sink/SelectDB-Cloud.md | 14 +--
.../selectdb/sink/SelectDBSinkFactory.java | 13 ++-
.../selectdb/sink/SelectDBSinkFactoryTest.java | 110 +++++++++++++++++++++
4 files changed, 133 insertions(+), 18 deletions(-)
diff --git a/docs/en/connectors/sink/SelectDB-Cloud.md
b/docs/en/connectors/sink/SelectDB-Cloud.md
index 337288cefe..80a7daa733 100644
--- a/docs/en/connectors/sink/SelectDB-Cloud.md
+++ b/docs/en/connectors/sink/SelectDB-Cloud.md
@@ -35,13 +35,13 @@ Version Supported
| Name | Type | Required | Default |
Description
|
|--------------------|--------|----------|------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| load-url | String | Yes | - | `SelectDB
Cloud` warehouse http address, the format is `warehouse_ip:http_port`
|
-| jdbc-url | String | Yes | - | `SelectDB
Cloud` warehouse jdbc address, the format is `warehouse_ip:mysql_port`
|
-| cluster-name | String | Yes | - | `SelectDB
Cloud` cluster name
|
-| username | String | Yes | - | `SelectDB
Cloud` user username
|
-| password | String | Yes | - | `SelectDB
Cloud` user password
|
+| load-url | String | Yes | - | `SelectDB
Cloud` warehouse HTTP address, in the format `warehouse_ip:http_port`. Must not
be empty or whitespace-only.
|
+| jdbc-url | String | Yes | - | `SelectDB
Cloud` warehouse JDBC address, in the format `warehouse_ip:mysql_port`. Must
not be empty or whitespace-only.
|
+| cluster-name | String | Yes | - | `SelectDB
Cloud` cluster name. Must not be empty or whitespace-only.
|
+| username | String | Yes | - | `SelectDB
Cloud` username. Must not be empty or whitespace-only.
|
+| password | String | No | - | Optional
`SelectDB Cloud` user password.
|
| sink.enable-2pc | bool | No | true | Whether to
enable two-phase commit (2pc), the default is true, to ensure Exactly-Once
semantics. SelectDB uses cache files to load data. When the amount of data is
large, cached data may become invalid (the default expiration time is 1 hour).
If you encounter a large amount of data write loss, please configure
sink.enable-2pc to false. |
-| table.identifier | String | Yes | - | The name
of `SelectDB Cloud` table, the format is `database.table`
|
+| table.identifier | String | Yes | - | The name
of the `SelectDB Cloud` table, in the format `database.table`. Must not be
empty or whitespace-only.
|
| sink.enable-delete | bool | No | false | Whether to
enable deletion. This option requires SelectDB Cloud table to enable batch
delete function, and only supports Unique model.
|
| sink.max-retries | int | No | 3 | the max
retry times if writing records to database failed
|
| sink.buffer-size | int | No | 10 * 1024 * 1024 (10MB) | the
buffer size to cache data for stream load.
|
@@ -176,4 +176,4 @@ sink {
## Changelog
-<ChangeLog />
\ No newline at end of file
+<ChangeLog />
diff --git a/docs/zh/connectors/sink/SelectDB-Cloud.md
b/docs/zh/connectors/sink/SelectDB-Cloud.md
index 83f57ba7e8..23ec152f22 100644
--- a/docs/zh/connectors/sink/SelectDB-Cloud.md
+++ b/docs/zh/connectors/sink/SelectDB-Cloud.md
@@ -36,13 +36,13 @@ SelectDB Cloud 接收器连接器的内部实现是在批量缓存后上传数
| 名称 | 类型 | 是否必填 | 默认值 |
描述
|
|--------------------|--------|----------|------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| load-url | String | 是 | - | `SelectDB
Cloud` 仓库的 HTTP 地址，格式为 `warehouse_ip:http_port`
|
-| jdbc-url | String | 是 | - | `SelectDB
Cloud` 仓库的 JDBC 地址，格式为 `warehouse_ip:mysql_port`
|
-| cluster-name | String | 是 | - | `SelectDB
Cloud` 集群名称
|
-| username | String | 是 | - | `SelectDB
Cloud` 用户名
|
-| password | String | 是 | - | `SelectDB
Cloud` 用户密码
|
+| load-url | String | 是 | - | `SelectDB
Cloud` 仓库的 HTTP 地址，格式为 `warehouse_ip:http_port`。值不能为空或仅包含空白字符。
|
+| jdbc-url | String | 是 | - | `SelectDB
Cloud` 仓库的 JDBC 地址，格式为 `warehouse_ip:mysql_port`。值不能为空或仅包含空白字符。
|
+| cluster-name | String | 是 | - | `SelectDB
Cloud` 集群名称。值不能为空或仅包含空白字符。
|
+| username | String | 是 | - | `SelectDB
Cloud` 用户名。值不能为空或仅包含空白字符。
|
+| password | String | 否 | - | 可选的
`SelectDB Cloud` 用户密码。
|
| sink.enable-2pc | bool | 否 | true |
是否启用两阶段提交（2pc），默认为 true，以确保 Exactly-Once 语义。SelectDB
使用缓存文件加载数据。当数据量较大时，缓存数据可能会失效（默认过期时间为 1 小时）。如果遇到大量数据写入丢失的情况，请将 sink.enable-2pc
配置为 false。
|
-| table.identifier | String | 是 | - | `SelectDB
Cloud` 表的名称，格式为 `database.table`
|
+| table.identifier | String | 是 | - | `SelectDB
Cloud` 表的名称，格式为 `database.table`。值不能为空或仅包含空白字符。
|
| sink.enable-delete | bool | 否 | false |
是否启用删除功能。此选项要求 SelectDB Cloud 表启用批量删除功能，并且仅支持 Unique 模型。
|
| sink.max-retries | int | 否 | 3 |
写入数据库失败时的最大重试次数
|
| sink.buffer-size | int | 否 | 10 * 1024 * 1024 (10MB) |
用于流式加载的数据缓存缓冲区大小
|
@@ -177,4 +177,4 @@ sink {
## 变更日志
-<ChangeLog />
\ No newline at end of file
+<ChangeLog />
diff --git
a/seatunnel-connectors-v2/connector-selectdb-cloud/src/main/java/org/apache/seatunnel/connectors/selectdb/sink/SelectDBSinkFactory.java
b/seatunnel-connectors-v2/connector-selectdb-cloud/src/main/java/org/apache/seatunnel/connectors/selectdb/sink/SelectDBSinkFactory.java
index 289a9406ec..1e08a24743 100644
---
a/seatunnel-connectors-v2/connector-selectdb-cloud/src/main/java/org/apache/seatunnel/connectors/selectdb/sink/SelectDBSinkFactory.java
+++
b/seatunnel-connectors-v2/connector-selectdb-cloud/src/main/java/org/apache/seatunnel/connectors/selectdb/sink/SelectDBSinkFactory.java
@@ -26,6 +26,8 @@ import
org.apache.seatunnel.connectors.selectdb.config.SelectDBSinkOptions;
import com.google.auto.service.AutoService;
+import static org.apache.seatunnel.api.configuration.util.Conditions.notBlank;
+
@AutoService(Factory.class)
public class SelectDBSinkFactory implements TableSinkFactory {
@@ -37,12 +39,15 @@ public class SelectDBSinkFactory implements
TableSinkFactory {
@Override
public OptionRule optionRule() {
return OptionRule.builder()
+ .required(SelectDBSinkOptions.JDBC_URL,
notBlank(SelectDBSinkOptions.JDBC_URL))
+ .required(SelectDBSinkOptions.LOAD_URL,
notBlank(SelectDBSinkOptions.LOAD_URL))
.required(
- SelectDBSinkOptions.JDBC_URL,
- SelectDBSinkOptions.LOAD_URL,
SelectDBSinkOptions.CLUSTER_NAME,
- SelectDBSinkOptions.USERNAME,
- SelectDBSinkOptions.TABLE_IDENTIFIER)
+ notBlank(SelectDBSinkOptions.CLUSTER_NAME))
+ .required(SelectDBSinkOptions.USERNAME,
notBlank(SelectDBSinkOptions.USERNAME))
+ .required(
+ SelectDBSinkOptions.TABLE_IDENTIFIER,
+ notBlank(SelectDBSinkOptions.TABLE_IDENTIFIER))
.optional(
SelectDBSinkOptions.PASSWORD,
SelectDBSinkOptions.SINK_ENABLE_2PC,
diff --git
a/seatunnel-connectors-v2/connector-selectdb-cloud/src/test/java/org/apache/seatunnel/connectors/selectdb/sink/SelectDBSinkFactoryTest.java
b/seatunnel-connectors-v2/connector-selectdb-cloud/src/test/java/org/apache/seatunnel/connectors/selectdb/sink/SelectDBSinkFactoryTest.java
new file mode 100644
index 0000000000..260d7a764a
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-selectdb-cloud/src/test/java/org/apache/seatunnel/connectors/selectdb/sink/SelectDBSinkFactoryTest.java
@@ -0,0 +1,110 @@
+/*
+ * 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.
+ */
+
+package org.apache.seatunnel.connectors.selectdb.sink;
+
+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.connectors.selectdb.config.SelectDBSinkOptions;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+class SelectDBSinkFactoryTest {
+
+ private static final OptionRule OPTION_RULE = new
SelectDBSinkFactory().optionRule();
+ private static final List<String> REQUIRED_STRING_KEYS =
+ Arrays.asList(
+ SelectDBSinkOptions.JDBC_URL.key(),
+ SelectDBSinkOptions.LOAD_URL.key(),
+ SelectDBSinkOptions.CLUSTER_NAME.key(),
+ SelectDBSinkOptions.USERNAME.key(),
+ SelectDBSinkOptions.TABLE_IDENTIFIER.key());
+
+ @Test
+ void testValidConfiguration() {
+ Assertions.assertDoesNotThrow(() -> validate(validConfig()));
+
+ Map<String, Object> valuesWithSurroundingSpaces = validConfig();
+ for (String key : REQUIRED_STRING_KEYS) {
+ valuesWithSurroundingSpaces.put(key, " " +
valuesWithSurroundingSpaces.get(key) + " ");
+ }
+ Assertions.assertDoesNotThrow(() ->
validate(valuesWithSurroundingSpaces));
+ }
+
+ @Test
+ void testMissingRequiredStringsRejected() {
+ for (String key : REQUIRED_STRING_KEYS) {
+ Map<String, Object> config = validConfig();
+ config.remove(key);
+
+ Assertions.assertThrows(OptionValidationException.class, () ->
validate(config), key);
+ }
+ }
+
+ @Test
+ void testEmptyRequiredStringsRejected() {
+ assertRequiredStringsRejected("");
+ }
+
+ @Test
+ void testWhitespaceOnlyRequiredStringsRejected() {
+ assertRequiredStringsRejected(" \t");
+ }
+
+ @Test
+ void testPasswordRemainsOptional() {
+ Assertions.assertDoesNotThrow(() -> validate(validConfig()));
+
+ Map<String, Object> emptyPassword = validConfig();
+ emptyPassword.put(SelectDBSinkOptions.PASSWORD.key(), "");
+ Assertions.assertDoesNotThrow(() -> validate(emptyPassword));
+ }
+
+ private void assertRequiredStringsRejected(String value) {
+ for (String key : REQUIRED_STRING_KEYS) {
+ Map<String, Object> config = validConfig();
+ config.put(key, value);
+
+ Assertions.assertThrows(OptionValidationException.class, () ->
validate(config), key);
+ }
+ }
+
+ private void validate(Map<String, Object> config) {
+ ReadonlyConfig readonlyConfig = ReadonlyConfig.fromMap(config);
+ ConfigValidator.validateUnknownKeys(
+ readonlyConfig, OPTION_RULE, SelectDBSinkOptions.IDENTIFIER);
+ ConfigValidator.of(readonlyConfig).validate(OPTION_RULE);
+ }
+
+ private Map<String, Object> validConfig() {
+ Map<String, Object> config = new HashMap<>();
+ config.put(SelectDBSinkOptions.JDBC_URL.key(),
"selectdb.example.com:9030");
+ config.put(SelectDBSinkOptions.LOAD_URL.key(),
"selectdb.example.com:8080");
+ config.put(SelectDBSinkOptions.CLUSTER_NAME.key(), "selectdb-cluster");
+ config.put(SelectDBSinkOptions.USERNAME.key(), "selectdb-user");
+ config.put(SelectDBSinkOptions.TABLE_IDENTIFIER.key(),
"database.table");
+ return config;
+ }
+}
