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 6b9075f0e6 [Improve][API] Add Map condition validators for
configuration option … (#11010)
6b9075f0e6 is described below
commit 6b9075f0e6b22fcb0cf9ff7f17474faeaabd5245
Author: zhiwei.niu <[email protected]>
AuthorDate: Sun Jun 7 12:28:22 2026 +0800
[Improve][API] Add Map condition validators for configuration option …
(#11010)
Co-authored-by: Doyeon Kim <[email protected]>
---
.../configuration-and-option-system.md | 23 ++++
.../configuration-and-option-system.md | 23 ++++
.../configuration/util/ConditionEvaluators.java | 16 +++
.../api/configuration/util/ConditionOperator.java | 9 +-
.../api/configuration/util/Conditions.java | 20 +++-
.../configuration/util/ConfigValidatorTest.java | 128 +++++++++++++++++++++
6 files changed, 217 insertions(+), 2 deletions(-)
diff --git a/docs/en/architecture/configuration-and-option-system.md
b/docs/en/architecture/configuration-and-option-system.md
index a3d63b4d10..c63fe1fb26 100644
--- a/docs/en/architecture/configuration-and-option-system.md
+++ b/docs/en/architecture/configuration-and-option-system.md
@@ -124,6 +124,9 @@ Available operators (all accessed via the `Conditions`
factory class):
| String | `lowerCase(option)` | string is all lowercase |
| Collection | `notEmpty(option)` | collection is not empty |
| Collection | `unique(option)` | collection has no duplicate elements |
+| Map | `mapNotEmpty(option)` | map is not empty |
+| Map | `mapContainsKey(option, key)` | map contains the specified key |
+| Map | `mapContainsKeys(option, key1, key2, ...)` | map contains all
specified keys |
| Cross-field | `lessThanField(option, other)` | value < another option's
value |
| Cross-field | `lessOrEqualField(option, other)` | value <= another option's
value |
| Cross-field | `greaterThanField(option, other)` | value > another option's
value |
@@ -295,6 +298,26 @@ Lists that must not be empty, or whose elements must be
unique.
.and(Conditions.unique(TABLES)))
```
+### Map constraints
+
+Map must not be empty:
+
+```java
+.required(PROPERTIES, Conditions.mapNotEmpty(PROPERTIES))
+```
+
+Map must contain a specific key:
+
+```java
+.required(KAFKA_CONFIG, Conditions.mapContainsKey(KAFKA_CONFIG,
"bootstrap.servers"))
+```
+
+Map must contain multiple keys simultaneously:
+
+```java
+.required(JDBC_PROPS, Conditions.mapContainsKeys(JDBC_PROPS, "url", "driver",
"user"))
+```
+
### Compound constraints with AND
Multiple conditions combined with `.and(...)`. All conditions must hold.
diff --git a/docs/zh/architecture/configuration-and-option-system.md
b/docs/zh/architecture/configuration-and-option-system.md
index 2c58ec4a77..e9b956a998 100644
--- a/docs/zh/architecture/configuration-and-option-system.md
+++ b/docs/zh/architecture/configuration-and-option-system.md
@@ -124,6 +124,9 @@ public OptionRule optionRule() {
| 字符串 | `lowerCase(option)` | 字符串全部小写 |
| 集合 | `notEmpty(option)` | 集合非空 |
| 集合 | `unique(option)` | 集合元素无重复 |
+| Map | `mapNotEmpty(option)` | Map 非空 |
+| Map | `mapContainsKey(option, key)` | Map 包含指定 key |
+| Map | `mapContainsKeys(option, key1, key2, ...)` | Map 同时包含所有指定 key |
| 跨字段 | `lessThanField(option, other)` | 值 < 另一个配置项的值 |
| 跨字段 | `lessOrEqualField(option, other)` | 值 <= 另一个配置项的值 |
| 跨字段 | `greaterThanField(option, other)` | 值 > 另一个配置项的值 |
@@ -295,6 +298,26 @@ Option validation failed (4 errors):
.and(Conditions.unique(TABLES)))
```
+### Map 约束
+
+Map 必须非空：
+
+```java
+.required(PROPERTIES, Conditions.mapNotEmpty(PROPERTIES))
+```
+
+Map 必须包含指定 key：
+
+```java
+.required(KAFKA_CONFIG, Conditions.mapContainsKey(KAFKA_CONFIG,
"bootstrap.servers"))
+```
+
+Map 必须同时包含多个 key：
+
+```java
+.required(JDBC_PROPS, Conditions.mapContainsKeys(JDBC_PROPS, "url", "driver",
"user"))
+```
+
### AND 复合约束
多个条件通过 `.and(...)` 组合，所有条件必须同时满足。
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/ConditionEvaluators.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/ConditionEvaluators.java
index 1bb231b287..82a72d08e9 100644
---
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/ConditionEvaluators.java
+++
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/ConditionEvaluators.java
@@ -119,6 +119,22 @@ public final class ConditionEvaluators {
return false;
});
+ // Map
+ m.put(
+ ConditionOperator.MAP_NOT_EMPTY,
+ (v, c, cfg) -> v instanceof Map && !((Map) v).isEmpty());
+ m.put(
+ ConditionOperator.MAP_CONTAINS_KEY,
+ (v, c, cfg) -> v instanceof Map && ((Map)
v).containsKey(c.getExpectValue()));
+ m.put(
+ ConditionOperator.MAP_CONTAINS_KEYS,
+ (v, c, cfg) -> {
+ if (!(v instanceof Map)) return false;
+ Object expect = c.getExpectValue();
+ if (!(expect instanceof Collection)) return false;
+ return ((Map) v).keySet().containsAll((Collection) expect);
+ });
+
// Cross-field (null on either side -> false, preserving or()
short-circuit)
m.put(
ConditionOperator.FIELD_LESS_THAN,
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/ConditionOperator.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/ConditionOperator.java
index 965f12843f..8dbd07288f 100644
---
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/ConditionOperator.java
+++
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/ConditionOperator.java
@@ -50,6 +50,12 @@ public enum ConditionOperator {
NOT_EMPTY("is not empty", Category.COLLECTION, Arity.UNARY,
Source.LITERAL),
COLLECTION_UNIQUE("has unique elements", Category.COLLECTION, Arity.UNARY,
Source.LITERAL),
+ // ==================== Map ====================
+
+ MAP_NOT_EMPTY("is not empty", Category.MAP, Arity.UNARY, Source.LITERAL),
+ MAP_CONTAINS_KEY("contains key", Category.MAP, Arity.BINARY,
Source.LITERAL),
+ MAP_CONTAINS_KEYS("contains keys", Category.MAP, Arity.BINARY,
Source.LITERAL),
+
// ==================== Cross-field comparison ====================
FIELD_LESS_THAN("<", Category.NUMERIC, Arity.BINARY, Source.FIELD),
@@ -61,7 +67,8 @@ public enum ConditionOperator {
EQUALITY,
NUMERIC,
STRING,
- COLLECTION
+ COLLECTION,
+ MAP
}
public enum Arity {
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/Conditions.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/Conditions.java
index ff2e550cc4..a510abbf17 100644
---
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/Conditions.java
+++
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/Conditions.java
@@ -19,6 +19,9 @@ package org.apache.seatunnel.api.configuration.util;
import org.apache.seatunnel.api.configuration.Option;
+import java.util.Arrays;
+import java.util.List;
+
/**
* Unified factory for creating {@link Condition} instances.
*
@@ -33,7 +36,7 @@ import org.apache.seatunnel.api.configuration.Option;
* .build();
* }</pre>
*
- * <p>Currently supported operators (17 total, 4 categories):
+ * <p>Currently supported operators (19 total, 5 categories):
*
* <ul>
* <li><b>Numeric</b>: {@code greaterThan}, {@code greaterOrEqual}, {@code
lessThan}, {@code
@@ -41,6 +44,7 @@ import org.apache.seatunnel.api.configuration.Option;
* <li><b>String</b>: {@code notBlank}, {@code startsWith}, {@code
contains}, {@code matches},
* {@code upperCase}, {@code lowerCase}
* <li><b>Collection</b>: {@code notEmpty}, {@code unique}
+ * <li><b>Map</b>: {@code mapNotEmpty}, {@code mapContainsKey}, {@code
mapContainsKeys}
* <li><b>Cross-field</b>: {@code lessThanField}, {@code lessOrEqualField},
{@code
* greaterThanField}, {@code greaterOrEqualField}
* </ul>
@@ -100,6 +104,20 @@ public final class Conditions {
return new Condition<>(option, ConditionOperator.NOT_EMPTY, null,
null);
}
+ // ==================== Map validation ====================
+ public static <T> Condition<T> mapNotEmpty(Option<T> option) {
+ return new Condition<>(option, ConditionOperator.MAP_NOT_EMPTY, null,
null);
+ }
+
+ public static <T> Condition<T> mapContainsKey(Option<T> option, String
key) {
+ return new Condition<>(option, ConditionOperator.MAP_CONTAINS_KEY, (T)
key, null);
+ }
+
+ public static <T> Condition<T> mapContainsKeys(Option<T> option, String...
keys) {
+ List<String> keyList = Arrays.asList(keys);
+ return new Condition<>(option, ConditionOperator.MAP_CONTAINS_KEYS,
(T) keyList, null);
+ }
+
public static <T> Condition<T> unique(Option<T> option) {
return new Condition<>(option, ConditionOperator.COLLECTION_UNIQUE,
null, null);
}
diff --git
a/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/util/ConfigValidatorTest.java
b/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/util/ConfigValidatorTest.java
index 068e1e4d74..ef9536ea41 100644
---
a/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/util/ConfigValidatorTest.java
+++
b/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/util/ConfigValidatorTest.java
@@ -43,6 +43,9 @@ import static
org.apache.seatunnel.api.configuration.util.Conditions.lessOrEqual
import static org.apache.seatunnel.api.configuration.util.Conditions.lessThan;
import static
org.apache.seatunnel.api.configuration.util.Conditions.lessThanField;
import static org.apache.seatunnel.api.configuration.util.Conditions.lowerCase;
+import static
org.apache.seatunnel.api.configuration.util.Conditions.mapContainsKey;
+import static
org.apache.seatunnel.api.configuration.util.Conditions.mapContainsKeys;
+import static
org.apache.seatunnel.api.configuration.util.Conditions.mapNotEmpty;
import static org.apache.seatunnel.api.configuration.util.Conditions.matches;
import static org.apache.seatunnel.api.configuration.util.Conditions.notBlank;
import static org.apache.seatunnel.api.configuration.util.Conditions.notEmpty;
@@ -2588,4 +2591,129 @@ public class ConfigValidatorTest {
ex.getMessage().startsWith("ErrorCode:[API-02]"),
"unified format should still carry standard ErrorCode prefix");
}
+
+ private static final Option<Map<String, String>> MAP_OPTION =
+
Options.key("properties").mapType().noDefaultValue().withDescription("test map
option");
+
+ @Test
+ public void testMapNotEmpty() {
+ OptionRule rule =
+ OptionRule.builder().required(MAP_OPTION,
mapNotEmpty(MAP_OPTION)).build();
+ Map<String, Object> config = new HashMap<>();
+
+ // empty map -> fail
+ config.put(MAP_OPTION.key(), Collections.emptyMap());
+ String msg =
+ assertThrows(OptionValidationException.class, () ->
validate(config, rule))
+ .getMessage();
+ Assertions.assertTrue(msg.contains("properties"));
+ Assertions.assertTrue(msg.contains("is not empty"));
+
+ // non-empty map -> pass
+ Map<String, String> props = new HashMap<>();
+ props.put("k1", "v1");
+ config.put(MAP_OPTION.key(), props);
+ Assertions.assertDoesNotThrow(() -> validate(config, rule));
+ }
+
+ @Test
+ public void testMapContainsKey() {
+ OptionRule rule =
+ OptionRule.builder()
+ .required(MAP_OPTION, mapContainsKey(MAP_OPTION,
"bootstrap.servers"))
+ .build();
+ Map<String, Object> config = new HashMap<>();
+
+ // map without required key -> fail
+ Map<String, String> props = new HashMap<>();
+ props.put("group.id", "test-group");
+ config.put(MAP_OPTION.key(), props);
+ String msg =
+ assertThrows(OptionValidationException.class, () ->
validate(config, rule))
+ .getMessage();
+ Assertions.assertTrue(msg.contains("properties"));
+ Assertions.assertTrue(msg.contains("contains key"));
+
+ // map with required key -> pass
+ props.put("bootstrap.servers", "localhost:9092");
+ config.put(MAP_OPTION.key(), props);
+ Assertions.assertDoesNotThrow(() -> validate(config, rule));
+ }
+
+ @Test
+ public void testMapContainsKeys() {
+ OptionRule rule =
+ OptionRule.builder()
+ .required(
+ MAP_OPTION, mapContainsKeys(MAP_OPTION,
"host", "port", "database"))
+ .build();
+ Map<String, Object> config = new HashMap<>();
+
+ // map missing some keys -> fail
+ Map<String, String> props = new HashMap<>();
+ props.put("host", "localhost");
+ props.put("port", "3306");
+ config.put(MAP_OPTION.key(), props);
+ String msg =
+ assertThrows(OptionValidationException.class, () ->
validate(config, rule))
+ .getMessage();
+ Assertions.assertTrue(msg.contains("properties"));
+ Assertions.assertTrue(msg.contains("contains keys"));
+
+ // map with all required keys -> pass
+ props.put("database", "mydb");
+ config.put(MAP_OPTION.key(), props);
+ Assertions.assertDoesNotThrow(() -> validate(config, rule));
+
+ // map with extra keys beyond required -> still pass
+ props.put("username", "root");
+ config.put(MAP_OPTION.key(), props);
+ Assertions.assertDoesNotThrow(() -> validate(config, rule));
+ }
+
+ @Test
+ public void testMapContainsKeyWithNullValue() {
+ OptionRule rule =
+ OptionRule.builder()
+ .required(MAP_OPTION, mapContainsKey(MAP_OPTION,
"token"))
+ .build();
+ Map<String, Object> config = new HashMap<>();
+
+ // non-map value -> fail
+ config.put(MAP_OPTION.key(), "not-a-map");
+ assertThrows(Exception.class, () -> validate(config, rule));
+
+ // map with the key but null value -> still pass (containsKey only
checks key presence)
+ Map<String, String> props = new HashMap<>();
+ props.put("token", null);
+ config.put(MAP_OPTION.key(), props);
+ Assertions.assertDoesNotThrow(() -> validate(config, rule));
+ }
+
+ @Test
+ public void testMapNotEmptyAndContainsKeyCombined() {
+ OptionRule rule =
+ OptionRule.builder()
+ .required(
+ MAP_OPTION,
+ mapNotEmpty(MAP_OPTION)
+ .and(mapContainsKey(MAP_OPTION,
"bootstrap.servers")))
+ .build();
+ Map<String, Object> config = new HashMap<>();
+
+ // empty map -> fail (mapNotEmpty)
+ config.put(MAP_OPTION.key(), Collections.emptyMap());
+ assertThrows(OptionValidationException.class, () -> validate(config,
rule));
+
+ // non-empty map without required key -> fail (containsKey)
+ Map<String, String> props = new HashMap<>();
+ props.put("group.id", "test");
+ config.put(MAP_OPTION.key(), props);
+ assertThrows(OptionValidationException.class, () -> validate(config,
rule));
+
+ // non-empty map with required key -> pass
+ props.put("bootstrap.servers", "localhost:9092");
+ config.put(MAP_OPTION.key(), props);
+ Assertions.assertDoesNotThrow(() -> validate(config, rule));
+ }
}
