This is an automated email from the ASF dual-hosted git repository.
nzw921rx 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 4ae4fee301 [Improve][Connector-V2][Paimon] Add declarative notBlank
validation to required options (#12573)
4ae4fee301 is described below
commit 4ae4fee30165e9f082f80b05ec5a8f3bb810b474
Author: Karthik Thota <[email protected]>
AuthorDate: Mon Oct 12 17:16:34 2026 +0530
[Improve][Connector-V2][Paimon] Add declarative notBlank validation to
required options (#12573)
Co-authored-by: zhiweiniu <[email protected]>
---
docs/en/connectors/sink/Paimon.md | 6 +-
docs/en/connectors/source/Paimon.md | 2 +-
docs/zh/connectors/sink/Paimon.md | 6 +-
docs/zh/connectors/source/Paimon.md | 2 +-
.../paimon/catalog/PaimonCatalogFactory.java | 6 +-
.../seatunnel/paimon/sink/PaimonSinkFactory.java | 6 +-
.../paimon/source/PaimonSourceFactory.java | 5 +-
.../config/PaimonNotBlankValidationTest.java | 278 +++++++++++++++++++++
8 files changed, 298 insertions(+), 13 deletions(-)
diff --git a/docs/en/connectors/sink/Paimon.md
b/docs/en/connectors/sink/Paimon.md
index db9b2c40a6..91721b9a82 100644
--- a/docs/en/connectors/sink/Paimon.md
+++ b/docs/en/connectors/sink/Paimon.md
@@ -70,12 +70,12 @@ libfb303-xxx.jar
| name | type | required | default value
| Description
|
|------------------------------|---------|----------|------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| warehouse | String | Yes | -
| Paimon warehouse path
|
+| warehouse | String | Yes | -
| Paimon warehouse path. Must not be blank.
|
| catalog_name | String | No | paimon
| The name of Paimon catalog
|
| catalog_type | String | No | filesystem
| Catalog type of Paimon, support filesystem and hive
|
| catalog_uri | String | Yes when `catalog_type` is `hive` |
- | Catalog URI of Paimon. Required when
`catalog_type` is `hive`.
|
-| database | String | Yes | -
| The database you want to access
|
-| table | String | Yes | -
| The table you want to access
|
+| database | String | Yes | -
| The database you want to access. Must not be blank.
|
+| table | String | Yes | -
| The table you want to access. Must not be blank.
|
| user | String | No | -
| Paimon user to access table
|
| password | String | No | -
| Paimon user password to access table
|
| hdfs_site_path | String | No | -
| Deprecated. The path of hdfs-site.xml. Prefer `paimon.hadoop.conf` or
`paimon.hadoop.conf-path` for new jobs
|
diff --git a/docs/en/connectors/source/Paimon.md
b/docs/en/connectors/source/Paimon.md
index a850ee7488..beddf6ae5e 100644
--- a/docs/en/connectors/source/Paimon.md
+++ b/docs/en/connectors/source/Paimon.md
@@ -70,7 +70,7 @@ Read data from Apache Paimon.
### warehouse [string]
-Paimon warehouse path
+Paimon warehouse path. Must not be blank.
### catalog_type [string]
diff --git a/docs/zh/connectors/sink/Paimon.md
b/docs/zh/connectors/sink/Paimon.md
index 9773f4d2ae..ddbae6f261 100644
--- a/docs/zh/connectors/sink/Paimon.md
+++ b/docs/zh/connectors/sink/Paimon.md
@@ -70,12 +70,12 @@ libfb303-xxx.jar
| 名称 | 类型 | 是否必须 | 默认值 |
描述
|
|------------------------------|------|------|------------------------------|------------------------------------------------------------------------------------------------------|
-| warehouse | 字符串 | 是 | - |
Paimon warehouse路径
|
+| warehouse | 字符串 | 是 | - |
Paimon warehouse路径，不能为空白字符串。
|
| catalog_name | 字符串 | 否 | paimon |
Paimon catalog名称
|
| catalog_type | 字符串 | 否 | filesystem |
Paimon的catalog类型，目前支持filesystem和hive
|
| catalog_uri | 字符串 | 当 `catalog_type` 为 `hive` 时是 | -
| Paimon catalog 的 URI。当 `catalog_type` 为 `hive` 时必须配置。
|
-| database | 字符串 | 是 | - |
数据库名称
|
-| table | 字符串 | 是 | - |
表名
|
+| database | 字符串 | 是 | - |
数据库名称，不能为空白字符串。
|
+| table | 字符串 | 是 | - |
表名，不能为空白字符串。
|
| user | 字符串 | 否 | - |
paimon开启权限后，用户名
|
| password | 字符串 | 否 | - |
paimon开启权限后，用户名对应密码
|
| hdfs_site_path | 字符串 | 否 | - |
已废弃。hdfs-site.xml 文件路径。新作业建议使用 `paimon.hadoop.conf` 或 `paimon.hadoop.conf-path`
|
diff --git a/docs/zh/connectors/source/Paimon.md
b/docs/zh/connectors/source/Paimon.md
index 4d06ca9e08..e753440962 100644
--- a/docs/zh/connectors/source/Paimon.md
+++ b/docs/zh/connectors/source/Paimon.md
@@ -70,7 +70,7 @@ import ChangeLog from '../changelog/connector-paimon.md';
### warehouse [string]
-Paimon warehouse 路径
+Paimon warehouse 路径，不能为空白字符串。
### catalog_type [string]
diff --git
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/catalog/PaimonCatalogFactory.java
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/catalog/PaimonCatalogFactory.java
index 20b67b39d7..35629fa7d6 100644
---
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/catalog/PaimonCatalogFactory.java
+++
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/catalog/PaimonCatalogFactory.java
@@ -47,8 +47,10 @@ public class PaimonCatalogFactory implements CatalogFactory {
return OptionRule.builder()
.required(
PaimonBaseOptions.WAREHOUSE,
- PaimonBaseOptions.DATABASE,
- PaimonBaseOptions.TABLE)
+ Conditions.notBlank(PaimonBaseOptions.WAREHOUSE))
+ .required(
+ PaimonBaseOptions.DATABASE,
Conditions.notBlank(PaimonBaseOptions.DATABASE))
+ .required(PaimonBaseOptions.TABLE,
Conditions.notBlank(PaimonBaseOptions.TABLE))
.optional(
PaimonBaseOptions.CATALOG_NAME,
PaimonBaseOptions.HDFS_SITE_PATH,
diff --git
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/PaimonSinkFactory.java
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/PaimonSinkFactory.java
index 49d4e90c00..d578c48435 100644
---
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/PaimonSinkFactory.java
+++
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/PaimonSinkFactory.java
@@ -48,8 +48,10 @@ public class PaimonSinkFactory implements TableSinkFactory {
return OptionRule.builder()
.required(
PaimonSinkOptions.WAREHOUSE,
- PaimonSinkOptions.DATABASE,
- PaimonSinkOptions.TABLE)
+ Conditions.notBlank(PaimonSinkOptions.WAREHOUSE))
+ .required(
+ PaimonSinkOptions.DATABASE,
Conditions.notBlank(PaimonSinkOptions.DATABASE))
+ .required(PaimonSinkOptions.TABLE,
Conditions.notBlank(PaimonSinkOptions.TABLE))
.optional(
PaimonSinkOptions.CATALOG_NAME,
PaimonSinkOptions.HDFS_SITE_PATH,
diff --git
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/PaimonSourceFactory.java
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/PaimonSourceFactory.java
index 339a634568..60d9da2a46 100644
---
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/PaimonSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/PaimonSourceFactory.java
@@ -18,6 +18,7 @@
package org.apache.seatunnel.connectors.seatunnel.paimon.source;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.Conditions;
import org.apache.seatunnel.api.configuration.util.OptionRule;
import org.apache.seatunnel.api.options.table.CatalogOptions;
import org.apache.seatunnel.api.source.SeaTunnelSource;
@@ -47,7 +48,9 @@ public class PaimonSourceFactory implements
TableSourceFactory {
@Override
public OptionRule optionRule() {
return OptionRule.builder()
- .required(PaimonSourceOptions.WAREHOUSE)
+ .required(
+ PaimonSourceOptions.WAREHOUSE,
+ Conditions.notBlank(PaimonSourceOptions.WAREHOUSE))
.optional(
PaimonSourceOptions.CATALOG_NAME,
PaimonSourceOptions.DATABASE,
diff --git
a/seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/config/PaimonNotBlankValidationTest.java
b/seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/config/PaimonNotBlankValidationTest.java
new file mode 100644
index 0000000000..07a88b37c4
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/config/PaimonNotBlankValidationTest.java
@@ -0,0 +1,278 @@
+/*
+ * 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.seatunnel.paimon.config;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConfigValidator;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
+import
org.apache.seatunnel.connectors.seatunnel.paimon.catalog.PaimonCatalogFactory;
+import org.apache.seatunnel.connectors.seatunnel.paimon.sink.PaimonSinkFactory;
+import
org.apache.seatunnel.connectors.seatunnel.paimon.source.PaimonSourceFactory;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Validates that {@link PaimonSourceFactory}, {@link PaimonSinkFactory}, and
{@link
+ * PaimonCatalogFactory} reject blank values for their required string
options, per #11007.
+ */
+class PaimonNotBlankValidationTest {
+
+ private static void validateSink(Map<String, Object> cfg) {
+ ConfigValidator.of(ReadonlyConfig.fromMap(cfg))
+ .validate(new PaimonSinkFactory().optionRule());
+ }
+
+ private static void validateSource(Map<String, Object> cfg) {
+ ConfigValidator.of(ReadonlyConfig.fromMap(cfg))
+ .validate(new PaimonSourceFactory().optionRule());
+ }
+
+ private static void validateCatalog(Map<String, Object> cfg) {
+ ConfigValidator.of(ReadonlyConfig.fromMap(cfg))
+ .validate(new PaimonCatalogFactory().optionRule());
+ }
+
+ private static Map<String, Object> validSink() {
+ Map<String, Object> m = new HashMap<>();
+ m.put(PaimonSinkOptions.WAREHOUSE.key(), "file:///tmp/paimon");
+ m.put(PaimonSinkOptions.DATABASE.key(), "db");
+ m.put(PaimonSinkOptions.TABLE.key(), "t");
+ return m;
+ }
+
+ private static Map<String, Object> validSource() {
+ Map<String, Object> m = new HashMap<>();
+ m.put(PaimonBaseOptions.WAREHOUSE.key(), "file:///tmp/paimon");
+ m.put(PaimonBaseOptions.TABLE.key(), "t");
+ return m;
+ }
+
+ @Test
+ void notBlankSinkValidConfig() {
+ Assertions.assertDoesNotThrow(
+ () -> validateSink(validSink()),
+ "Valid sink config must pass after notBlank change");
+ }
+
+ @Test
+ void notBlankSinkMissingWarehouse() {
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put(PaimonSinkOptions.DATABASE.key(), "db");
+ cfg.put(PaimonSinkOptions.TABLE.key(), "t");
+ OptionValidationException ex =
+ Assertions.assertThrows(OptionValidationException.class, () ->
validateSink(cfg));
+ Assertions.assertTrue(
+ ex.getMessage().contains("warehouse"),
+ "Expected failure to mention 'warehouse', got: " +
ex.getMessage());
+ }
+
+ @Test
+ void notBlankSinkEmptyWarehouse() {
+ Map<String, Object> cfg = validSink();
+ cfg.put(PaimonSinkOptions.WAREHOUSE.key(), "");
+ OptionValidationException ex =
+ Assertions.assertThrows(OptionValidationException.class, () ->
validateSink(cfg));
+ Assertions.assertTrue(
+ ex.getMessage().contains("warehouse"),
+ "Expected failure to mention 'warehouse', got: " +
ex.getMessage());
+ }
+
+ @Test
+ void notBlankSinkWhitespaceWarehouse() {
+ Map<String, Object> cfg = validSink();
+ cfg.put(PaimonSinkOptions.WAREHOUSE.key(), " ");
+ OptionValidationException ex =
+ Assertions.assertThrows(OptionValidationException.class, () ->
validateSink(cfg));
+ Assertions.assertTrue(
+ ex.getMessage().contains("warehouse"),
+ "Expected failure to mention 'warehouse', got: " +
ex.getMessage());
+ }
+
+ @Test
+ void notBlankSinkPaddedWarehousePasses() {
+ Map<String, Object> cfg = validSink();
+ cfg.put(PaimonSinkOptions.WAREHOUSE.key(), " file:///tmp/paimon ");
+ Assertions.assertDoesNotThrow(
+ () -> validateSink(cfg),
+ "Padded warehouse (' file:///tmp/paimon ') must pass —
notBlank trims internally");
+ }
+
+ @Test
+ void notBlankSinkEmptyDatabase() {
+ Map<String, Object> cfg = validSink();
+ cfg.put(PaimonSinkOptions.DATABASE.key(), "");
+ OptionValidationException ex =
+ Assertions.assertThrows(OptionValidationException.class, () ->
validateSink(cfg));
+ Assertions.assertTrue(
+ ex.getMessage().contains("database"),
+ "Expected failure to mention 'database', got: " +
ex.getMessage());
+ }
+
+ @Test
+ void notBlankSinkWhitespaceDatabase() {
+ Map<String, Object> cfg = validSink();
+ cfg.put(PaimonSinkOptions.DATABASE.key(), " \t ");
+ OptionValidationException ex =
+ Assertions.assertThrows(OptionValidationException.class, () ->
validateSink(cfg));
+ Assertions.assertTrue(
+ ex.getMessage().contains("database"),
+ "Expected failure to mention 'database', got: " +
ex.getMessage());
+ }
+
+ @Test
+ void notBlankSinkPaddedDatabasePasses() {
+ Map<String, Object> cfg = validSink();
+ cfg.put(PaimonSinkOptions.DATABASE.key(), " db ");
+ Assertions.assertDoesNotThrow(
+ () -> validateSink(cfg), "Padded database ' db ' must pass
notBlank");
+ }
+
+ @Test
+ void notBlankSinkEmptyTable() {
+ Map<String, Object> cfg = validSink();
+ cfg.put(PaimonSinkOptions.TABLE.key(), "");
+ OptionValidationException ex =
+ Assertions.assertThrows(OptionValidationException.class, () ->
validateSink(cfg));
+ Assertions.assertTrue(
+ ex.getMessage().contains("table"),
+ "Expected failure to mention 'table', got: " +
ex.getMessage());
+ }
+
+ @Test
+ void notBlankSinkWhitespaceTable() {
+ Map<String, Object> cfg = validSink();
+ cfg.put(PaimonSinkOptions.TABLE.key(), " ");
+ OptionValidationException ex =
+ Assertions.assertThrows(OptionValidationException.class, () ->
validateSink(cfg));
+ Assertions.assertTrue(
+ ex.getMessage().contains("table"),
+ "Expected failure to mention 'table', got: " +
ex.getMessage());
+ }
+
+ @Test
+ void notBlankSinkPaddedTablePasses() {
+ Map<String, Object> cfg = validSink();
+ cfg.put(PaimonSinkOptions.TABLE.key(), " t ");
+ Assertions.assertDoesNotThrow(
+ () -> validateSink(cfg), "Padded table ' t ' must pass
notBlank");
+ }
+
+ @Test
+ void notBlankSourceValidConfig() {
+ Assertions.assertDoesNotThrow(
+ () -> validateSource(validSource()),
+ "Valid source config must pass after notBlank change");
+ }
+
+ @Test
+ void notBlankSourceMissingWarehouse() {
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put(PaimonBaseOptions.TABLE.key(), "t");
+ OptionValidationException ex =
+ Assertions.assertThrows(OptionValidationException.class, () ->
validateSource(cfg));
+ Assertions.assertTrue(
+ ex.getMessage().contains("warehouse"),
+ "Expected failure to mention 'warehouse', got: " +
ex.getMessage());
+ }
+
+ @Test
+ void notBlankSourceEmptyWarehouse() {
+ Map<String, Object> cfg = validSource();
+ cfg.put(PaimonBaseOptions.WAREHOUSE.key(), "");
+ OptionValidationException ex =
+ Assertions.assertThrows(OptionValidationException.class, () ->
validateSource(cfg));
+ Assertions.assertTrue(
+ ex.getMessage().contains("warehouse"),
+ "Expected failure to mention 'warehouse', got: " +
ex.getMessage());
+ }
+
+ @Test
+ void notBlankSourceWhitespaceWarehouse() {
+ Map<String, Object> cfg = validSource();
+ cfg.put(PaimonBaseOptions.WAREHOUSE.key(), " ");
+ OptionValidationException ex =
+ Assertions.assertThrows(OptionValidationException.class, () ->
validateSource(cfg));
+ Assertions.assertTrue(
+ ex.getMessage().contains("warehouse"),
+ "Expected failure to mention 'warehouse', got: " +
ex.getMessage());
+ }
+
+ @Test
+ void notBlankSourcePaddedWarehousePasses() {
+ Map<String, Object> cfg = validSource();
+ cfg.put(PaimonBaseOptions.WAREHOUSE.key(), " file:///tmp/paimon ");
+ Assertions.assertDoesNotThrow(
+ () -> validateSource(cfg), "Padded source warehouse must pass
notBlank");
+ }
+
+ @Test
+ void notBlankCatalogValidConfig() {
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put(PaimonBaseOptions.WAREHOUSE.key(), "file:///tmp/paimon");
+ cfg.put(PaimonBaseOptions.DATABASE.key(), "db");
+ cfg.put(PaimonBaseOptions.TABLE.key(), "t");
+ Assertions.assertDoesNotThrow(
+ () -> validateCatalog(cfg), "Valid catalog config must pass
after notBlank change");
+ }
+
+ @Test
+ void notBlankCatalogEmptyWarehouse() {
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put(PaimonBaseOptions.WAREHOUSE.key(), "");
+ cfg.put(PaimonBaseOptions.DATABASE.key(), "db");
+ cfg.put(PaimonBaseOptions.TABLE.key(), "t");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class, () ->
validateCatalog(cfg));
+ Assertions.assertTrue(
+ ex.getMessage().contains("warehouse"),
+ "Expected failure to mention 'warehouse', got: " +
ex.getMessage());
+ }
+
+ @Test
+ void notBlankCatalogWhitespaceDatabase() {
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put(PaimonBaseOptions.WAREHOUSE.key(), "file:///tmp/paimon");
+ cfg.put(PaimonBaseOptions.DATABASE.key(), " ");
+ cfg.put(PaimonBaseOptions.TABLE.key(), "t");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class, () ->
validateCatalog(cfg));
+ Assertions.assertTrue(
+ ex.getMessage().contains("database"),
+ "Expected failure to mention 'database', got: " +
ex.getMessage());
+ }
+
+ @Test
+ void notBlankCatalogEmptyTable() {
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put(PaimonBaseOptions.WAREHOUSE.key(), "file:///tmp/paimon");
+ cfg.put(PaimonBaseOptions.DATABASE.key(), "db");
+ cfg.put(PaimonBaseOptions.TABLE.key(), "");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class, () ->
validateCatalog(cfg));
+ Assertions.assertTrue(
+ ex.getMessage().contains("table"),
+ "Expected failure to mention 'table', got: " +
ex.getMessage());
+ }
+}
