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 e2951f2064 [Feature][API] Add EXTENSION operator for pluggable
validation in ConditionOperator (#11048)
e2951f2064 is described below
commit e2951f2064e568d7880097d2b5d37f72caf5d610
Author: zhiwei.niu <[email protected]>
AuthorDate: Fri Jun 12 16:56:07 2026 +0800
[Feature][API] Add EXTENSION operator for pluggable validation in
ConditionOperator (#11048)
---
.../configuration-and-option-system.md | 68 ++
docs/en/engines/zeta/rest-api-v2.md | 29 +-
.../configuration-and-option-system.md | 68 ++
docs/zh/engines/zeta/rest-api-v2.md | 29 +-
.../api/configuration/util/Condition.java | 20 +-
.../configuration/util/ConditionEvaluators.java | 8 +
.../api/configuration/util/ConditionExtension.java | 63 ++
.../api/configuration/util/ConditionOperator.java | 15 +-
.../api/configuration/util/Conditions.java | 9 +-
.../configuration/util/ConfigValidatorTest.java | 706 ++++++++++++++++++++-
.../api/configuration/util/OptionRuleTest.java | 30 +
.../seatunnel/command/MetadataExportCommand.java | 10 +-
.../command/MetadataExportCommandTest.java | 91 +++
.../server/rest/service/OptionRulesService.java | 6 +-
.../rest/service/OptionRulesServiceTest.java | 43 ++
15 files changed, 1175 insertions(+), 20 deletions(-)
diff --git a/docs/en/architecture/configuration-and-option-system.md
b/docs/en/architecture/configuration-and-option-system.md
index c63fe1fb26..5ba5cc7828 100644
--- a/docs/en/architecture/configuration-and-option-system.md
+++ b/docs/en/architecture/configuration-and-option-system.md
@@ -131,6 +131,7 @@ Available operators (all accessed via the `Conditions`
factory class):
| Cross-field | `lessOrEqualField(option, other)` | value <= another option's
value |
| Cross-field | `greaterThanField(option, other)` | value > another option's
value |
| Cross-field | `greaterOrEqualField(option, other)` | value >= another
option's value |
+| Extension | `Conditions.extension(option, ext)` | delegates to a
`ConditionExtension<T>` implementation |
:::tip
Multiple conditions can be chained with `.and(...)` or `.or(...)` to form
compound constraints. AND binds tighter than OR, so `A.or(B).and(C)` evaluates
as `A || (B && C)`.
@@ -206,6 +207,7 @@ Quick reference:
| Validate value only when trigger matches | `.conditional(trigger, value,
condition...)` |
| Optional field with value check when present | `.optional(opt,
condition...)` |
| Cross-field comparisons | `Conditions.lessThanField/greaterThanField(...)` |
+| Custom / structural validation | `Conditions.extension(opt, ext)` |
### Required fields
@@ -395,6 +397,72 @@ When two optional fields are provided together, their
values must satisfy a cros
Conditions.lessThanField(START_TS, END_TS))
```
+### Custom validation with Extension
+
+When built-in operators are not expressive enough — for example, validating
the internal structure of a `List<Map>` or enforcing cross-key constraints
inside nested configs — use the `EXTENSION` operator.
+
+Implement `ConditionExtension<T>` and wire it via
`Conditions.extension(option, ext)`. The extension plugs into the same
`valueConstraints` pipeline as all built-in operators, so it works with
`.and()` / `.or()`, `required`, `optional`, and `conditional` rules.
+
+Inline anonymous class:
+
+```java
+.optional(API_KEY_ENCODED, Conditions.extension(API_KEY_ENCODED,
+ new ConditionExtension<String>() {
+ @Override
+ public String description() {
+ return "must be Base64-encoded 'id:api_key'";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig cfg, String v)
+ throws OptionValidationException {
+ try {
+ return new
String(Base64.getDecoder().decode(v)).contains(":");
+ } catch (IllegalArgumentException e) {
+ return false;
+ }
+ }
+ }))
+```
+
+Static inner class for complex types:
+
+```java
+static class TableConfigsValidator
+ implements ConditionExtension<List<Map<String, Object>>> {
+ @Override
+ public String description() {
+ return "each entry must contain a non-empty 'table_name', and all
table names must be unique";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, List<Map<String, Object>>
value) throws OptionValidationException {
+ if (value.isEmpty()) {
+ return false;
+ }
+ Set<String> seen = new HashSet<>();
+ for (Map<String, Object> entry : value) {
+ Object name = entry.get("table_name");
+ if (!(name instanceof String) || ((String) name).isEmpty()) {
+ return false;
+ }
+ if (!seen.add((String) name)) {
+ return false;
+ }
+ }
+ return true;
+ }
+}
+
+.exclusive(TABLE_CONFIGS, SCHEMA)
+.optional(TABLE_CONFIGS,
+ Conditions.extension(TABLE_CONFIGS, new TableConfigsValidator()))
+```
+
+:::caution
+`ConditionExtension.evaluate()` runs during job submission validation only.
REST metadata queries only serialize `description()` and do not invoke
`evaluate()`. Implementations should avoid I/O (database connections, HTTP
calls, file access) and only validate structure and values.
+:::
+
## Why It Matters For Operators
This architecture is also what makes the `option-rules` REST endpoint useful.
Tools can inspect the runtime metadata of installed connectors and dynamically
understand:
diff --git a/docs/en/engines/zeta/rest-api-v2.md
b/docs/en/engines/zeta/rest-api-v2.md
index ccde0b04cb..672e3f962d 100644
--- a/docs/en/engines/zeta/rest-api-v2.md
+++ b/docs/en/engines/zeta/rest-api-v2.md
@@ -110,6 +110,10 @@ Please refer [security](security.md)
]
},
"expectValue": "TEMPLATE",
+ "compareOperator": null,
+ "compareOption": null,
+ "conditionOperator": "EQUAL",
+ "conditionOperatorCategory": "EQUALITY",
"operator": null,
"next": null
},
@@ -139,6 +143,26 @@ Please refer [security](security.md)
"operator": null,
"next": null
}
+ },
+ {
+ "expression": "'port' must be between 1 and 65535",
+ "conditionTree": {
+ "option": {
+ "key": "port",
+ "type": "java.lang.Integer",
+ "defaultValue": null,
+ "description": "Server port",
+ "fallbackKeys": [],
+ "optionValues": null
+ },
+ "expectValue": "must be between 1 and 65535",
+ "compareOperator": "extension",
+ "compareOption": null,
+ "conditionOperator": "EXTENSION",
+ "conditionOperatorCategory": "EXTENSION",
+ "operator": null,
+ "next": null
+ }
}
]
}
@@ -151,8 +175,9 @@ Please refer [security](security.md)
- `optionRule.conditionRules` recursively exposes nested conditional option
rules and is an empty array when the connector does not define nested rules.
- For conditional rules, both `expression` and `expressionTree` are returned
for dynamic form rendering.
- `optionRule.valueConstraints` describes value-level validation rules such as
numeric ranges, string patterns, and cross-field comparisons. Each entry
provides a human-readable `expression` string alongside a structured
`conditionTree` for programmatic use. This array is empty when the connector
does not define any value constraints.
-- Within `conditionTree`, the `compareOperator` field (e.g. `>=`, `<`, `>`)
and `compareOption` field are populated for numeric and cross-field
comparisons. For equality checks and other non-comparison conditions, these
fields are `null`.
-- The `conditionOperator` field provides a stable, machine-readable operator
identifier (e.g. `GREATER_OR_EQUAL`, `NOT_BLANK`, `FIELD_LESS_THAN`), while
`conditionOperatorCategory` indicates the operator's category (e.g. `NUMERIC`,
`STRING`, `COLLECTION`, `EQUALITY`). These two fields are designed for
programmatic consumption by frontend applications and automation tools.
+- Within `conditionTree`, the `compareOperator` field is `null` for `EQUAL`
and otherwise uses the operator symbol exposed by the runtime rule (for example
`>=`, `is not blank`, or `extension`). The `compareOption` field is populated
only for cross-field comparisons.
+- `conditionOperator` is a stable operator identifier. Possible values include
`EQUAL`, `GREATER_OR_EQUAL`, `NOT_BLANK`, `FIELD_LESS_THAN`, `EXTENSION`, etc.
`conditionOperatorCategory` indicates the operator category, such as `NUMERIC`,
`STRING`, `COLLECTION`, `EQUALITY`, `EXTENSION`, etc.
+- For `EXTENSION` conditions, `expectValue` carries the rule description text
returned by `ConditionExtension.description()`.
</details>
diff --git a/docs/zh/architecture/configuration-and-option-system.md
b/docs/zh/architecture/configuration-and-option-system.md
index e9b956a998..7f437ff246 100644
--- a/docs/zh/architecture/configuration-and-option-system.md
+++ b/docs/zh/architecture/configuration-and-option-system.md
@@ -131,6 +131,7 @@ public OptionRule optionRule() {
| 跨字段 | `lessOrEqualField(option, other)` | 值 <= 另一个配置项的值 |
| 跨字段 | `greaterThanField(option, other)` | 值 > 另一个配置项的值 |
| 跨字段 | `greaterOrEqualField(option, other)` | 值 >= 另一个配置项的值 |
+| 扩展 | `Conditions.extension(option, ext)` | 委托给 `ConditionExtension<T>`
实现类执行自定义校验 |
:::tip
多个条件可以通过 `.and(...)` 或 `.or(...)` 链式组合成复合约束。AND 优先级高于 OR，因此 `A.or(B).and(C)`
等价于 `A || (B && C)`。
@@ -206,6 +207,7 @@ Option validation failed (4 errors):
| 条件触发的值校验 | `.conditional(trigger, value, condition...)` |
| 可选字段（提供时校验） | `.optional(opt, condition...)` |
| 跨字段比较 | `Conditions.lessThanField/greaterThanField(...)` |
+| 自定义 / 结构化校验 | `Conditions.extension(opt, ext)` |
### 必填字段
@@ -395,6 +397,72 @@ AND 优先级高于 OR，因此 `A.or(B.and(C))` 等价于 `A || (B && C)`。适
Conditions.lessThanField(START_TS, END_TS))
```
+### 自定义校验（Extension 扩展算子）
+
+当内置算子无法满足需求时 — 例如校验 `List<Map>` 内部结构，或对嵌套配置做跨 key 约束 — 使用 `EXTENSION` 算子。
+
+实现 `ConditionExtension<T>` 接口，通过 `Conditions.extension(option, ext)`
接入。扩展算子复用与内置算子完全相同的 `valueConstraints` 管线，支持 `.and()` / `.or()` 链式组合，适用于
`required`、`optional`、`conditional` 等所有规则类型。
+
+匿名内部类写法：
+
+```java
+.optional(API_KEY_ENCODED, Conditions.extension(API_KEY_ENCODED,
+ new ConditionExtension<String>() {
+ @Override
+ public String description() {
+ return "must be Base64-encoded 'id:api_key'";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig cfg, String v)
+ throws OptionValidationException {
+ try {
+ return new
String(Base64.getDecoder().decode(v)).contains(":");
+ } catch (IllegalArgumentException e) {
+ return false;
+ }
+ }
+ }))
+```
+
+静态内部类写法（适合复杂类型）：
+
+```java
+static class TableConfigsValidator
+ implements ConditionExtension<List<Map<String, Object>>> {
+ @Override
+ public String description() {
+ return "each entry must contain a non-empty 'table_name', and all
table names must be unique";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, List<Map<String, Object>>
value) throws OptionValidationException {
+ if (value.isEmpty()) {
+ return false;
+ }
+ Set<String> seen = new HashSet<>();
+ for (Map<String, Object> entry : value) {
+ Object name = entry.get("table_name");
+ if (!(name instanceof String) || ((String) name).isEmpty()) {
+ return false;
+ }
+ if (!seen.add((String) name)) {
+ return false;
+ }
+ }
+ return true;
+ }
+}
+
+.exclusive(TABLE_CONFIGS, SCHEMA)
+.optional(TABLE_CONFIGS,
+ Conditions.extension(TABLE_CONFIGS, new TableConfigsValidator()))
+```
+
+:::caution
+`ConditionExtension.evaluate()` 仅在作业提交校验时执行，REST 元数据查询只序列化
`description()`，不会调用 `evaluate()`。实现时应避免 I/O 操作（如数据库连接、HTTP 请求、文件读写），只做结构和值校验。
+:::
+
## 为什么这对运维也重要
这套设计也是 `option-rules` REST 接口能够成立的原因。运维平台或 UI 可以通过运行时元数据动态获知：
diff --git a/docs/zh/engines/zeta/rest-api-v2.md
b/docs/zh/engines/zeta/rest-api-v2.md
index 7cbc0e43f2..b5f3d13658 100644
--- a/docs/zh/engines/zeta/rest-api-v2.md
+++ b/docs/zh/engines/zeta/rest-api-v2.md
@@ -108,6 +108,10 @@ seatunnel:
]
},
"expectValue": "TEMPLATE",
+ "compareOperator": null,
+ "compareOption": null,
+ "conditionOperator": "EQUAL",
+ "conditionOperatorCategory": "EQUALITY",
"operator": null,
"next": null
},
@@ -137,6 +141,26 @@ seatunnel:
"operator": null,
"next": null
}
+ },
+ {
+ "expression": "'port' must be between 1 and 65535",
+ "conditionTree": {
+ "option": {
+ "key": "port",
+ "type": "java.lang.Integer",
+ "defaultValue": null,
+ "description": "Server port",
+ "fallbackKeys": [],
+ "optionValues": null
+ },
+ "expectValue": "must be between 1 and 65535",
+ "compareOperator": "extension",
+ "compareOption": null,
+ "conditionOperator": "EXTENSION",
+ "conditionOperatorCategory": "EXTENSION",
+ "operator": null,
+ "next": null
+ }
}
]
}
@@ -149,8 +173,9 @@ seatunnel:
- `optionRule.conditionRules` 会递归返回嵌套条件规则；当 connector 未定义嵌套规则时，该字段返回空数组。
- 对于条件规则，会同时返回 `expression` 和 `expressionTree`，便于 Web 做动态表单渲染。
- `optionRule.valueConstraints`
描述值级别的校验规则，包括数值范围、字符串模式匹配以及跨字段比较等。每个条目同时提供人类可读的 `expression` 字符串和便于程序处理的结构化
`conditionTree`。当连接器未定义值约束时，该数组为空。
-- 在 `conditionTree` 中，`compareOperator` 字段（如 `>=`、`<`、`>`）和 `compareOption`
字段用于数值比较和跨字段比较场景；对于等值判断及其他非比较类条件，这两个字段为 `null`。
-- `conditionOperator` 字段提供稳定的、机器可读的操作符标识（如
`GREATER_OR_EQUAL`、`NOT_BLANK`、`FIELD_LESS_THAN`），`conditionOperatorCategory`
字段标明操作符所属分类（如
`NUMERIC`、`STRING`、`COLLECTION`、`EQUALITY`）。这两个字段专为前端应用和自动化工具的程序化消费而设计。
+- 在 `conditionTree` 中，`compareOperator` 字段在 `EQUAL` 场景下为
`null`，其他情况下会返回运行时规则暴露的操作符符号，例如 `>=`、`is not blank` 或
`extension`。`compareOption` 字段仅在跨字段比较场景下有值。
+- `conditionOperator` 是稳定的操作符标识，可选值包括
`EQUAL`、`GREATER_OR_EQUAL`、`NOT_BLANK`、`FIELD_LESS_THAN`、`EXTENSION`
等；`conditionOperatorCategory` 是操作符的分类，可选值包括
`NUMERIC`、`STRING`、`COLLECTION`、`EQUALITY`、`EXTENSION` 等。
+- 对于 `EXTENSION` 条件，`expectValue` 承载的是 `ConditionExtension.description()`
返回的规则说明文本。
</details>
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/Condition.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/Condition.java
index 5511b7ce7d..e94c21b904 100644
---
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/Condition.java
+++
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/Condition.java
@@ -32,15 +32,25 @@ public class Condition<T> {
private final T expectValue;
private final ConditionOperator operator;
private final Option<?> compareOption;
+ private final ConditionExtension<T> extension;
private Boolean and = null;
private Condition<?> next = null;
Condition(Option<T> option, T expectValue) {
- this(option, ConditionOperator.EQUAL, expectValue, null);
+ this(option, ConditionOperator.EQUAL, expectValue, null, null);
}
Condition(
Option<T> option, ConditionOperator operator, T expectValue,
Option<?> compareOption) {
+ this(option, operator, expectValue, compareOption, null);
+ }
+
+ Condition(
+ Option<T> option,
+ ConditionOperator operator,
+ T expectValue,
+ Option<?> compareOption,
+ ConditionExtension<T> extension) {
if (option == null) {
throw new IllegalArgumentException("Condition option must not be
null");
}
@@ -61,10 +71,15 @@ public class Condition<T> {
"Operator %s requires an expectValue, but
expectValue is null",
operator.name()));
}
+ if (operator == ConditionOperator.EXTENSION && extension == null) {
+ throw new IllegalArgumentException(
+ "Operator EXTENSION requires a non-null
ConditionExtension");
+ }
this.option = option;
this.operator = operator;
this.expectValue = expectValue;
this.compareOption = compareOption;
+ this.extension = extension;
}
public static <T> Condition<T> of(Option<T> option, T expectValue) {
@@ -219,6 +234,9 @@ public class Condition<T> {
ConditionOperator op = cond.operator;
String key = "'" + cond.option.key() + "'";
+ if (op == ConditionOperator.EXTENSION) {
+ return key + " " + cond.extension.description();
+ }
if (op.getSource() == ConditionOperator.Source.FIELD) {
return key + " " + op.getSymbol() + " '" +
cond.compareOption.key() + "'";
}
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 82a72d08e9..eacb9566e6 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
@@ -169,6 +169,14 @@ public final class ConditionEvaluators {
return compareNumbers(v, other) >= 0;
});
+ // Extension (custom logic delegated to ConditionExtension)
+ m.put(
+ ConditionOperator.EXTENSION,
+ (v, c, cfg) -> {
+ ConditionExtension<Object> ext =
(ConditionExtension<Object>) c.getExtension();
+ return ext.evaluate(cfg, v);
+ });
+
for (ConditionOperator op : ConditionOperator.values()) {
if (!m.containsKey(op)) {
throw new IllegalStateException(
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/ConditionExtension.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/ConditionExtension.java
new file mode 100644
index 0000000000..2048059678
--- /dev/null
+++
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/ConditionExtension.java
@@ -0,0 +1,63 @@
+/*
+ * 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.api.configuration.util;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+
+/**
+ * Pluggable validation extension for cases where built-in {@link
ConditionOperator} operators are
+ * not expressive enough — for example, validating the internal structure of a
{@code List<Map>} or
+ * enforcing cross-key constraints inside nested configs.
+ *
+ * <p>Wire an implementation via {@link
+ * Conditions#extension(org.apache.seatunnel.api.configuration.Option,
ConditionExtension)}. The
+ * extension plugs into the same {@code valueConstraints} pipeline as all
built-in operators and
+ * supports {@code .and()} / {@code .or()} chaining, {@code required}, {@code
optional}, and {@code
+ * conditional} rules.
+ *
+ * <p>Implementations should avoid I/O (database connections, HTTP calls, file
access) and only
+ * validate structure and values. {@link #evaluate} runs only during job
submission validation; REST
+ * metadata queries only serialize {@link #description()}.
+ *
+ * @param <T> the option value type
+ */
+public interface ConditionExtension<T> {
+
+ /**
+ * Rule description used in error messages ({@link Condition#toString()})
and metadata
+ * serialization (REST {@code /option-rules} and CLI metadata export).
+ *
+ * @return non-null description, e.g. {@code "must be between 1 and 65535"}
+ */
+ String description();
+
+ /**
+ * Evaluates whether {@code value} passes this validation rule.
+ *
+ * <p>Return {@code false} for simple failure — the framework composes the
error from {@link
+ * #description()} automatically. Throw {@link OptionValidationException}
when a richer,
+ * context-specific message is needed. Avoid other unchecked exceptions —
they propagate
+ * unwrapped.
+ *
+ * @param config full configuration context (read-only), available for
cross-field checks
+ * @param value the resolved option value; may be {@code null}
+ * @return {@code true} if valid
+ * @throws OptionValidationException for detailed error reporting
+ */
+ boolean evaluate(ReadonlyConfig config, T value) throws
OptionValidationException;
+}
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 8dbd07288f..632c444a0b 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
@@ -61,24 +61,31 @@ public enum ConditionOperator {
FIELD_LESS_THAN("<", Category.NUMERIC, Arity.BINARY, Source.FIELD),
FIELD_LESS_OR_EQUAL("<=", Category.NUMERIC, Arity.BINARY, Source.FIELD),
FIELD_GREATER_THAN(">", Category.NUMERIC, Arity.BINARY, Source.FIELD),
- FIELD_GREATER_OR_EQUAL(">=", Category.NUMERIC, Arity.BINARY, Source.FIELD);
+ FIELD_GREATER_OR_EQUAL(">=", Category.NUMERIC, Arity.BINARY, Source.FIELD),
+
+ // ==================== Extension ====================
+
+ EXTENSION("extension", Category.EXTENSION, Arity.EXTENSION,
Source.EXTENSION);
public enum Category {
EQUALITY,
NUMERIC,
STRING,
COLLECTION,
- MAP
+ MAP,
+ EXTENSION
}
public enum Arity {
UNARY,
- BINARY
+ BINARY,
+ EXTENSION
}
public enum Source {
LITERAL,
- FIELD
+ FIELD,
+ EXTENSION
}
private final String symbol;
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 a510abbf17..90ad9beda1 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
@@ -36,8 +36,6 @@ import java.util.List;
* .build();
* }</pre>
*
- * <p>Currently supported operators (19 total, 5 categories):
- *
* <ul>
* <li><b>Numeric</b>: {@code greaterThan}, {@code greaterOrEqual}, {@code
lessThan}, {@code
* lessOrEqual}
@@ -47,6 +45,7 @@ import java.util.List;
* <li><b>Map</b>: {@code mapNotEmpty}, {@code mapContainsKey}, {@code
mapContainsKeys}
* <li><b>Cross-field</b>: {@code lessThanField}, {@code lessOrEqualField},
{@code
* greaterThanField}, {@code greaterOrEqualField}
+ * <li><b>Extension</b>: {@code extension} (custom logic via {@link
ConditionExtension})
* </ul>
*
* <p>Additionally, equality checks are available via {@link
Condition#of(Option, Object)} (EQUAL)
@@ -139,4 +138,10 @@ public final class Conditions {
public static <T> Condition<T> greaterOrEqualField(Option<T> option,
Option<T> other) {
return new Condition<>(option,
ConditionOperator.FIELD_GREATER_OR_EQUAL, null, other);
}
+
+ // ==================== Extension (pluggable validation)
====================
+
+ public static <T> Condition<T> extension(Option<T> option,
ConditionExtension<T> ext) {
+ return new Condition<>(option, ConditionOperator.EXTENSION, null,
null, ext);
+ }
}
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 fec4b5e05b..2b859bb059 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
@@ -17,6 +17,8 @@
package org.apache.seatunnel.api.configuration.util;
+import
org.apache.seatunnel.shade.com.fasterxml.jackson.core.type.TypeReference;
+
import org.apache.seatunnel.api.configuration.Option;
import org.apache.seatunnel.api.configuration.OptionTest;
import org.apache.seatunnel.api.configuration.Options;
@@ -407,7 +409,6 @@ public class ConfigValidatorTest {
@Test
public void testMultipleValueNestedRule() {
OptionRule subOption1 = OptionRule.builder().required(KEY_USERNAME,
KEY_PASSWORD).build();
- OptionRule subOption2 =
OptionRule.builder().required(KEY_BEARER_TOKEN).build();
OptionRule optionRule =
OptionRule.builder()
.optional(SINGLE_CHOICE_VALUE_TEST)
@@ -453,9 +454,6 @@ public class ConfigValidatorTest {
public static final Option<String> DB_NAME =
Options.key("db_name").stringType().noDefaultValue().withDescription("database
name");
- public static final Option<String> DELIMITER =
-
Options.key("delimiter").stringType().noDefaultValue().withDescription("delimiter");
-
public static final Option<Long> START_TS =
Options.key("start_ts").longType().noDefaultValue().withDescription("start
timestamp");
@@ -2753,7 +2751,7 @@ public class ConfigValidatorTest {
// list option present with valid list -> pass
Map<String, Object> config6 = new HashMap<>();
- config6.put(TEST_TOPIC.key(), Arrays.asList("topic1"));
+ config6.put(TEST_TOPIC.key(), Collections.singletonList("topic1"));
Assertions.assertDoesNotThrow(() -> validate(config6, rule));
}
@@ -2787,4 +2785,702 @@ public class ConfigValidatorTest {
config4.put(TEST_TOPIC.key(), Collections.emptyList());
assertThrows(OptionValidationException.class, () -> validate(config4,
rule));
}
+
+ static final Option<Integer> EXT_PORT =
+
Options.key("ext.port").intType().noDefaultValue().withDescription("port");
+
+ static final Option<Integer> OPT_PORT =
+
Options.key("opt.port").intType().defaultValue(8080).withDescription("optional
port");
+
+ static class PortRangeExtension implements ConditionExtension<Integer> {
+ @Override
+ public String description() {
+ return "must be between 1 and 65535";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, Integer value)
+ throws OptionValidationException {
+ return value != null && value >= 1 && value <= 65535;
+ }
+ }
+
+ @Test
+ public void testExtensionRequiredPass() {
+ OptionRule rule =
+ OptionRule.builder()
+ .required(
+ EXT_PORT, Conditions.extension(EXT_PORT, new
PortRangeExtension()))
+ .build();
+ Map<String, Object> config = new HashMap<>();
+ config.put("ext.port", 8080);
+ Assertions.assertDoesNotThrow(() -> validate(config, rule));
+ }
+
+ @Test
+ public void testExtensionRequiredFail() {
+ OptionRule rule =
+ OptionRule.builder()
+ .required(
+ EXT_PORT, Conditions.extension(EXT_PORT, new
PortRangeExtension()))
+ .build();
+ Map<String, Object> config = new HashMap<>();
+ config.put("ext.port", 99999);
+ assertThrows(OptionValidationException.class, () -> validate(config,
rule));
+ }
+
+ @Test
+ public void testExtensionOptionalAbsentSkip() {
+ OptionRule rule =
+ OptionRule.builder()
+ .optional(
+ OPT_PORT, Conditions.extension(OPT_PORT, new
PortRangeExtension()))
+ .build();
+ Map<String, Object> config = new HashMap<>();
+ Assertions.assertDoesNotThrow(() -> validate(config, rule));
+ }
+
+ @Test
+ public void testExtensionOptionalPresentPass() {
+ OptionRule rule =
+ OptionRule.builder()
+ .optional(
+ OPT_PORT, Conditions.extension(OPT_PORT, new
PortRangeExtension()))
+ .build();
+ Map<String, Object> config = new HashMap<>();
+ config.put("opt.port", 443);
+ Assertions.assertDoesNotThrow(() -> validate(config, rule));
+ }
+
+ @Test
+ public void testExtensionOptionalPresentFail() {
+ OptionRule rule =
+ OptionRule.builder()
+ .optional(
+ OPT_PORT, Conditions.extension(OPT_PORT, new
PortRangeExtension()))
+ .build();
+ Map<String, Object> config = new HashMap<>();
+ config.put("opt.port", 0);
+ assertThrows(OptionValidationException.class, () -> validate(config,
rule));
+ }
+
+ @Test
+ public void testExtensionAndBuiltinCombined() {
+ Condition<Integer> combined =
+ greaterOrEqual(EXT_PORT, 1)
+ .and(Conditions.extension(EXT_PORT, new
PortRangeExtension()));
+ OptionRule rule = OptionRule.builder().required(EXT_PORT,
combined).build();
+
+ Map<String, Object> pass = new HashMap<>();
+ pass.put("ext.port", 80);
+ Assertions.assertDoesNotThrow(() -> validate(pass, rule));
+
+ Map<String, Object> fail = new HashMap<>();
+ fail.put("ext.port", 70000);
+ assertThrows(OptionValidationException.class, () -> validate(fail,
rule));
+ }
+
+ @Test
+ public void testExtensionOrBuiltinCombined() {
+ ConditionExtension<Integer> isWellKnown =
+ new ConditionExtension<Integer>() {
+ @Override
+ public String description() {
+ return "is a well-known port (1-1023)";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, Integer
value)
+ throws OptionValidationException {
+ return value != null && value >= 1 && value <= 1023;
+ }
+ };
+ Condition<Integer> combined =
+ Conditions.extension(EXT_PORT, isWellKnown)
+ .or(Condition.of(EXT_PORT, ConditionOperator.EQUAL,
8080));
+ OptionRule rule = OptionRule.builder().required(EXT_PORT,
combined).build();
+
+ Map<String, Object> passWellKnown = new HashMap<>();
+ passWellKnown.put("ext.port", 443);
+ Assertions.assertDoesNotThrow(() -> validate(passWellKnown, rule));
+
+ Map<String, Object> passExact = new HashMap<>();
+ passExact.put("ext.port", 8080);
+ Assertions.assertDoesNotThrow(() -> validate(passExact, rule));
+
+ Map<String, Object> fail = new HashMap<>();
+ fail.put("ext.port", 5000);
+ assertThrows(OptionValidationException.class, () -> validate(fail,
rule));
+ }
+
+ @Test
+ public void testExtensionConditionToString() {
+ Condition<Integer> cond = Conditions.extension(EXT_PORT, new
PortRangeExtension());
+ assertEquals("'ext.port' must be between 1 and 65535",
cond.toString());
+ }
+
+ @Test
+ public void testExtensionConditionEquals() {
+ Condition<Integer> a = Conditions.extension(EXT_PORT, new
PortRangeExtension());
+ Condition<Integer> b = Conditions.extension(EXT_PORT, new
PortRangeExtension());
+ assertEquals(a, b);
+ assertEquals(a.hashCode(), b.hashCode());
+
+ Condition<Integer> c =
+ Conditions.extension(
+ EXT_PORT,
+ new ConditionExtension<Integer>() {
+ @Override
+ public String description() {
+ return "different impl";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config,
Integer value) {
+ return value != null;
+ }
+ });
+ assertEquals(a, c);
+ assertEquals(a.hashCode(), c.hashCode());
+ }
+
+ static final Option<List<Map<String, Object>>> LIST_MAP_OPTION =
+ Options.key("rules")
+ .type(new TypeReference<List<Map<String, Object>>>() {})
+ .noDefaultValue()
+ .withDescription("list of rule maps");
+
+ static final Option<Map<String, List<Map<String, Object>>>>
NESTED_MAP_OPTION =
+ Options.key("nested.config")
+ .type(new TypeReference<Map<String, List<Map<String,
Object>>>>() {})
+ .noDefaultValue()
+ .withDescription("nested map of list of maps");
+
+ static class ListMapStructureExtension
+ implements ConditionExtension<List<Map<String, Object>>> {
+ @Override
+ public String description() {
+ return "each rule must contain 'field' and 'type' keys";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, List<Map<String,
Object>> value)
+ throws OptionValidationException {
+ if (value == null || value.isEmpty()) return false;
+ for (Map<String, Object> rule : value) {
+ if (!rule.containsKey("field") || !rule.containsKey("type")) {
+ return false;
+ }
+ }
+ return true;
+ }
+ }
+
+ static class NestedMapExtension
+ implements ConditionExtension<Map<String, List<Map<String,
Object>>>> {
+ @Override
+ public String description() {
+ return "each group must have non-empty rules with 'name' key";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, Map<String,
List<Map<String, Object>>> value)
+ throws OptionValidationException {
+ if (value == null || value.isEmpty()) return false;
+ for (Map.Entry<String, List<Map<String, Object>>> entry :
value.entrySet()) {
+ List<Map<String, Object>> rules = entry.getValue();
+ if (rules == null || rules.isEmpty()) return false;
+ for (Map<String, Object> rule : rules) {
+ if (!rule.containsKey("name")) return false;
+ }
+ }
+ return true;
+ }
+ }
+
+ @Test
+ public void testExtensionListMapValidStructure() {
+ OptionRule rule =
+ OptionRule.builder()
+ .required(
+ LIST_MAP_OPTION,
+ Conditions.extension(
+ LIST_MAP_OPTION, new
ListMapStructureExtension()))
+ .build();
+
+ Map<String, Object> ruleItem1 = new HashMap<>();
+ ruleItem1.put("field", "name");
+ ruleItem1.put("type", "string");
+ Map<String, Object> ruleItem2 = new HashMap<>();
+ ruleItem2.put("field", "age");
+ ruleItem2.put("type", "int");
+
+ Map<String, Object> config = new HashMap<>();
+ config.put("rules", Arrays.asList(ruleItem1, ruleItem2));
+ Assertions.assertDoesNotThrow(() -> validate(config, rule));
+ }
+
+ @Test
+ public void testExtensionListMapMissingKey() {
+ OptionRule rule =
+ OptionRule.builder()
+ .required(
+ LIST_MAP_OPTION,
+ Conditions.extension(
+ LIST_MAP_OPTION, new
ListMapStructureExtension()))
+ .build();
+
+ Map<String, Object> ruleItem = new HashMap<>();
+ ruleItem.put("field", "name");
+
+ Map<String, Object> config = new HashMap<>();
+ config.put("rules", Collections.singletonList(ruleItem));
+ assertThrows(OptionValidationException.class, () -> validate(config,
rule));
+ }
+
+ @Test
+ public void testExtensionListMapEmptyList() {
+ OptionRule rule =
+ OptionRule.builder()
+ .required(
+ LIST_MAP_OPTION,
+ Conditions.extension(
+ LIST_MAP_OPTION, new
ListMapStructureExtension()))
+ .build();
+
+ Map<String, Object> config = new HashMap<>();
+ config.put("rules", Collections.emptyList());
+ assertThrows(OptionValidationException.class, () -> validate(config,
rule));
+ }
+
+ @Test
+ public void testExtensionListMapNullValue() {
+ OptionRule rule =
+ OptionRule.builder()
+ .required(
+ LIST_MAP_OPTION,
+ Conditions.extension(
+ LIST_MAP_OPTION, new
ListMapStructureExtension()))
+ .build();
+
+ Map<String, Object> config = new HashMap<>();
+ config.put("rules", null);
+ assertThrows(OptionValidationException.class, () -> validate(config,
rule));
+ }
+
+ @Test
+ public void testExtensionListMapSingleItemValid() {
+ OptionRule rule =
+ OptionRule.builder()
+ .required(
+ LIST_MAP_OPTION,
+ Conditions.extension(
+ LIST_MAP_OPTION, new
ListMapStructureExtension()))
+ .build();
+
+ Map<String, Object> item = new HashMap<>();
+ item.put("field", "id");
+ item.put("type", "long");
+
+ Map<String, Object> config = new HashMap<>();
+ config.put("rules", Collections.singletonList(item));
+ Assertions.assertDoesNotThrow(() -> validate(config, rule));
+ }
+
+ @Test
+ public void testExtensionListMapPartialInvalid() {
+ OptionRule rule =
+ OptionRule.builder()
+ .required(
+ LIST_MAP_OPTION,
+ Conditions.extension(
+ LIST_MAP_OPTION, new
ListMapStructureExtension()))
+ .build();
+
+ Map<String, Object> good = new HashMap<>();
+ good.put("field", "id");
+ good.put("type", "long");
+ Map<String, Object> bad = new HashMap<>();
+ bad.put("field", "name");
+
+ Map<String, Object> config = new HashMap<>();
+ config.put("rules", Arrays.asList(good, bad));
+ assertThrows(OptionValidationException.class, () -> validate(config,
rule));
+ }
+
+ @Test
+ public void testExtensionNestedMapValid() {
+ OptionRule rule =
+ OptionRule.builder()
+ .required(
+ NESTED_MAP_OPTION,
+ Conditions.extension(NESTED_MAP_OPTION, new
NestedMapExtension()))
+ .build();
+
+ Map<String, Object> item1 = new HashMap<>();
+ item1.put("name", "rule1");
+ Map<String, Object> item2 = new HashMap<>();
+ item2.put("name", "rule2");
+
+ Map<String, Object> nested = new HashMap<>();
+ nested.put("group_a", Arrays.asList(item1, item2));
+
+ Map<String, Object> config = new HashMap<>();
+ config.put("nested.config", nested);
+ Assertions.assertDoesNotThrow(() -> validate(config, rule));
+ }
+
+ @Test
+ public void testExtensionNestedMapEmptyGroup() {
+ OptionRule rule =
+ OptionRule.builder()
+ .required(
+ NESTED_MAP_OPTION,
+ Conditions.extension(NESTED_MAP_OPTION, new
NestedMapExtension()))
+ .build();
+
+ Map<String, Object> nested = new HashMap<>();
+ nested.put("group_a", Collections.emptyList());
+
+ Map<String, Object> config = new HashMap<>();
+ config.put("nested.config", nested);
+ assertThrows(OptionValidationException.class, () -> validate(config,
rule));
+ }
+
+ @Test
+ public void testExtensionNestedMapMissingNameKey() {
+ OptionRule rule =
+ OptionRule.builder()
+ .required(
+ NESTED_MAP_OPTION,
+ Conditions.extension(NESTED_MAP_OPTION, new
NestedMapExtension()))
+ .build();
+
+ Map<String, Object> item = new HashMap<>();
+ item.put("value", "something");
+ Map<String, Object> nested = new HashMap<>();
+ nested.put("group_a", Collections.singletonList(item));
+
+ Map<String, Object> config = new HashMap<>();
+ config.put("nested.config", nested);
+ assertThrows(OptionValidationException.class, () -> validate(config,
rule));
+ }
+
+ @Test
+ public void testExtensionNestedMapEmptyOuter() {
+ OptionRule rule =
+ OptionRule.builder()
+ .required(
+ NESTED_MAP_OPTION,
+ Conditions.extension(NESTED_MAP_OPTION, new
NestedMapExtension()))
+ .build();
+
+ Map<String, Object> config = new HashMap<>();
+ config.put("nested.config", Collections.emptyMap());
+ assertThrows(OptionValidationException.class, () -> validate(config,
rule));
+ }
+
+ @Test
+ public void testExtensionNestedMapMultiGroupPartialInvalid() {
+ OptionRule rule =
+ OptionRule.builder()
+ .required(
+ NESTED_MAP_OPTION,
+ Conditions.extension(NESTED_MAP_OPTION, new
NestedMapExtension()))
+ .build();
+
+ Map<String, Object> good = new HashMap<>();
+ good.put("name", "ok");
+ Map<String, Object> bad = new HashMap<>();
+ bad.put("other", "missing name");
+ Map<String, Object> nested = new HashMap<>();
+ nested.put("group_a", Collections.singletonList(good));
+ nested.put("group_b", Collections.singletonList(bad));
+
+ Map<String, Object> config = new HashMap<>();
+ config.put("nested.config", nested);
+ assertThrows(OptionValidationException.class, () -> validate(config,
rule));
+ }
+
+ @Test
+ public void testExtensionThrowsOptionValidationException() {
+ OptionRule rule =
+ OptionRule.builder()
+ .required(
+ EXT_PORT,
+ Conditions.extension(
+ EXT_PORT,
+ new ConditionExtension<Integer>() {
+ @Override
+ public String description() {
+ return "must be even";
+ }
+
+ @Override
+ public boolean evaluate(
+ ReadonlyConfig config,
Integer value)
+ throws
OptionValidationException {
+ if (value != null && value % 2
!= 0) {
+ throw new
OptionValidationException(
+ "Value %d is odd,
must be even", value);
+ }
+ return value != null;
+ }
+ }))
+ .build();
+
+ Map<String, Object> pass = new HashMap<>();
+ pass.put("ext.port", 80);
+ Assertions.assertDoesNotThrow(() -> validate(pass, rule));
+
+ Map<String, Object> fail = new HashMap<>();
+ fail.put("ext.port", 81);
+ OptionValidationException ex =
+ assertThrows(OptionValidationException.class, () ->
validate(fail, rule));
+ Assertions.assertTrue(ex.getMessage().contains("odd"));
+ }
+
+ @Test
+ public void testExtensionListMapOptionalAbsent() {
+ OptionRule rule =
+ OptionRule.builder()
+ .optional(
+ LIST_MAP_OPTION,
+ Conditions.extension(
+ LIST_MAP_OPTION, new
ListMapStructureExtension()))
+ .build();
+
+ Map<String, Object> config = new HashMap<>();
+ Assertions.assertDoesNotThrow(() -> validate(config, rule));
+ }
+
+ @Test
+ public void testExtensionListMapToString() {
+ Condition<List<Map<String, Object>>> cond =
+ Conditions.extension(LIST_MAP_OPTION, new
ListMapStructureExtension());
+ assertEquals("'rules' each rule must contain 'field' and 'type' keys",
cond.toString());
+ }
+
+ @Test
+ public void testExtensionNestedMapToString() {
+ Condition<Map<String, List<Map<String, Object>>>> cond =
+ Conditions.extension(NESTED_MAP_OPTION, new
NestedMapExtension());
+ assertEquals(
+ "'nested.config' each group must have non-empty rules with
'name' key",
+ cond.toString());
+ }
+
+ @Test
+ public void testExtensionListMapAndBuiltinChain() {
+ Condition<List<Map<String, Object>>> combined =
+ notEmpty(LIST_MAP_OPTION)
+ .and(
+ Conditions.extension(
+ LIST_MAP_OPTION, new
ListMapStructureExtension()));
+ OptionRule rule = OptionRule.builder().required(LIST_MAP_OPTION,
combined).build();
+
+ Map<String, Object> config = new HashMap<>();
+ config.put("rules", Collections.emptyList());
+ assertThrows(OptionValidationException.class, () -> validate(config,
rule));
+
+ Map<String, Object> good = new HashMap<>();
+ good.put("field", "x");
+ good.put("type", "y");
+ Map<String, Object> config2 = new HashMap<>();
+ config2.put("rules", Collections.singletonList(good));
+ Assertions.assertDoesNotThrow(() -> validate(config2, rule));
+ }
+
+ @Test
+ public void testExtensionAndExtensionChain() {
+ ConditionExtension<Integer> positiveExt =
+ new ConditionExtension<Integer>() {
+ @Override
+ public String description() {
+ return "must be positive";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, Integer
value) {
+ return value != null && value > 0;
+ }
+ };
+ ConditionExtension<Integer> evenExt =
+ new ConditionExtension<Integer>() {
+ @Override
+ public String description() {
+ return "must be even";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, Integer
value) {
+ return value != null && value % 2 == 0;
+ }
+ };
+ Condition<Integer> combined =
+ Conditions.extension(EXT_PORT, positiveExt)
+ .and(Conditions.extension(EXT_PORT, evenExt));
+ OptionRule rule = OptionRule.builder().required(EXT_PORT,
combined).build();
+
+ Map<String, Object> pass = new HashMap<>();
+ pass.put("ext.port", 80);
+ Assertions.assertDoesNotThrow(() -> validate(pass, rule));
+
+ // positive but odd -> fail
+ Map<String, Object> failOdd = new HashMap<>();
+ failOdd.put("ext.port", 81);
+ assertThrows(OptionValidationException.class, () -> validate(failOdd,
rule));
+
+ // even but negative -> fail
+ Map<String, Object> failNeg = new HashMap<>();
+ failNeg.put("ext.port", -2);
+ assertThrows(OptionValidationException.class, () -> validate(failNeg,
rule));
+ }
+
+ @Test
+ public void testExtensionOrExtensionChain() {
+ ConditionExtension<Integer> wellKnownExt =
+ new ConditionExtension<Integer>() {
+ @Override
+ public String description() {
+ return "well-known port (1-1023)";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, Integer
value) {
+ return value != null && value >= 1 && value <= 1023;
+ }
+ };
+ ConditionExtension<Integer> highPortExt =
+ new ConditionExtension<Integer>() {
+ @Override
+ public String description() {
+ return "high port (49152-65535)";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, Integer
value) {
+ return value != null && value >= 49152 && value <=
65535;
+ }
+ };
+ Condition<Integer> combined =
+ Conditions.extension(EXT_PORT, wellKnownExt)
+ .or(Conditions.extension(EXT_PORT, highPortExt));
+ OptionRule rule = OptionRule.builder().required(EXT_PORT,
combined).build();
+
+ Map<String, Object> passWellKnown = new HashMap<>();
+ passWellKnown.put("ext.port", 443);
+ Assertions.assertDoesNotThrow(() -> validate(passWellKnown, rule));
+
+ Map<String, Object> passHigh = new HashMap<>();
+ passHigh.put("ext.port", 50000);
+ Assertions.assertDoesNotThrow(() -> validate(passHigh, rule));
+
+ // middle range -> fail both
+ Map<String, Object> fail = new HashMap<>();
+ fail.put("ext.port", 8080);
+ assertThrows(OptionValidationException.class, () -> validate(fail,
rule));
+ }
+
+ @Test
+ public void testExtensionWithExclusiveOptional() {
+ ConditionExtension<String> patternExt =
+ new ConditionExtension<String>() {
+ @Override
+ public String description() {
+ return "must start with a letter";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, String
value) {
+ return value != null
+ && !value.isEmpty()
+ && Character.isLetter(value.charAt(0));
+ }
+ };
+ OptionRule rule =
+ OptionRule.builder()
+ .exclusive(TEST_TOPIC_PATTERN, TEST_TOPIC)
+ .optional(
+ TEST_TOPIC_PATTERN,
+ Conditions.extension(TEST_TOPIC_PATTERN,
patternExt))
+ .optional(TEST_TOPIC, notEmpty(TEST_TOPIC))
+ .build();
+
+ // neither present -> fails exclusive
+ Map<String, Object> config1 = new HashMap<>();
+ assertThrows(OptionValidationException.class, () -> validate(config1,
rule));
+
+ // pattern present with valid value -> pass
+ Map<String, Object> config2 = new HashMap<>();
+ config2.put(TEST_TOPIC_PATTERN.key(), "topic.*");
+ Assertions.assertDoesNotThrow(() -> validate(config2, rule));
+
+ // pattern present but starts with digit -> fails extension
+ Map<String, Object> config3 = new HashMap<>();
+ config3.put(TEST_TOPIC_PATTERN.key(), "123topic");
+ assertThrows(OptionValidationException.class, () -> validate(config3,
rule));
+
+ // both present -> fails exclusive
+ Map<String, Object> config4 = new HashMap<>();
+ config4.put(TEST_TOPIC_PATTERN.key(), "topic.*");
+ config4.put(TEST_TOPIC.key(), Collections.singletonList("t1"));
+ assertThrows(OptionValidationException.class, () -> validate(config4,
rule));
+
+ // topic present with valid list -> pass
+ Map<String, Object> config5 = new HashMap<>();
+ config5.put(TEST_TOPIC.key(), Arrays.asList("t1", "t2"));
+ Assertions.assertDoesNotThrow(() -> validate(config5, rule));
+ }
+
+ @Test
+ public void testExtensionWithBundledOptional() {
+ ConditionExtension<String> patternExt =
+ new ConditionExtension<String>() {
+ @Override
+ public String description() {
+ return "must contain a wildcard";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, String
value) {
+ return value != null && value.contains("*");
+ }
+ };
+ OptionRule rule =
+ OptionRule.builder()
+ .bundled(TEST_TOPIC_PATTERN, TEST_TOPIC)
+ .optional(
+ TEST_TOPIC_PATTERN,
+ Conditions.extension(TEST_TOPIC_PATTERN,
patternExt))
+ .optional(TEST_TOPIC, notEmpty(TEST_TOPIC))
+ .build();
+
+ // neither present -> pass (bundled group absent)
+ Map<String, Object> config1 = new HashMap<>();
+ Assertions.assertDoesNotThrow(() -> validate(config1, rule));
+
+ // both present with valid values -> pass
+ Map<String, Object> config2 = new HashMap<>();
+ config2.put(TEST_TOPIC_PATTERN.key(), "topic.*");
+ config2.put(TEST_TOPIC.key(), Collections.singletonList("t1"));
+ Assertions.assertDoesNotThrow(() -> validate(config2, rule));
+
+ // only one present -> fails bundled
+ Map<String, Object> config3 = new HashMap<>();
+ config3.put(TEST_TOPIC_PATTERN.key(), "topic.*");
+ assertThrows(OptionValidationException.class, () -> validate(config3,
rule));
+
+ // both present but pattern has no wildcard -> fails extension
+ Map<String, Object> config4 = new HashMap<>();
+ config4.put(TEST_TOPIC_PATTERN.key(), "topic-fixed");
+ config4.put(TEST_TOPIC.key(), Collections.singletonList("t1"));
+ assertThrows(OptionValidationException.class, () -> validate(config4,
rule));
+
+ // both present but topic is empty -> fails notEmpty
+ Map<String, Object> config5 = new HashMap<>();
+ config5.put(TEST_TOPIC_PATTERN.key(), "topic.*");
+ config5.put(TEST_TOPIC.key(), Collections.emptyList());
+ assertThrows(OptionValidationException.class, () -> validate(config5,
rule));
+ }
}
diff --git
a/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/util/OptionRuleTest.java
b/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/util/OptionRuleTest.java
index 1c3db5bdfc..38b9a47adc 100644
---
a/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/util/OptionRuleTest.java
+++
b/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/util/OptionRuleTest.java
@@ -22,6 +22,7 @@ import
org.apache.seatunnel.shade.com.fasterxml.jackson.core.type.TypeReference;
import org.apache.seatunnel.api.configuration.Option;
import org.apache.seatunnel.api.configuration.OptionTest;
import org.apache.seatunnel.api.configuration.Options;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -350,6 +351,35 @@ public class OptionRuleTest {
TEST_TOPIC_PATTERN,
Conditions.notBlank(TEST_TOPIC_PATTERN))
.build();
assertThrows(OptionValidationException.class, executable);
+
+ // test extension condition builds correctly
+ ConditionExtension<Integer> positiveExt =
+ new ConditionExtension<Integer>() {
+ @Override
+ public String description() {
+ return "must be positive";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, Integer
value) {
+ return value != null && value > 0;
+ }
+ };
+ OptionRule extRule =
+ OptionRule.builder()
+ .required(TEST_PORTS)
+ .optional(TEST_NUM, Conditions.extension(TEST_NUM,
positiveExt))
+ .build();
+ Assertions.assertNotNull(extRule);
+ assertEquals(1, extRule.getValueConstraints().size());
+ assertEquals(
+ ConditionOperator.EXTENSION,
extRule.getValueConstraints().get(0).getOperator());
+
+ // test extension with null extension throws
+ assertThrows(IllegalArgumentException.class, () ->
Conditions.extension(TEST_NUM, null));
+
+ // test extension with null option throws
+ assertThrows(IllegalArgumentException.class, () ->
Conditions.extension(null, positiveExt));
}
@Test
diff --git
a/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/MetadataExportCommand.java
b/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/MetadataExportCommand.java
index 69d6f78b6b..ff47c66aeb 100644
---
a/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/MetadataExportCommand.java
+++
b/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/MetadataExportCommand.java
@@ -420,10 +420,14 @@ public class MetadataExportCommand implements
Command<MetadataExportCommandArgs>
}
ObjectNode node = mapper.createObjectNode();
node.put("key", condition.getOption().key());
- if (condition.getExpectValue() != null) {
- node.put("expectValue",
String.valueOf(condition.getExpectValue()));
- }
ConditionOperator op = condition.getOperator();
+ Object expectValue = condition.getExpectValue();
+ if (op == ConditionOperator.EXTENSION && condition.getExtension() !=
null) {
+ expectValue = condition.getExtension().description();
+ }
+ if (expectValue != null) {
+ node.put("expectValue", String.valueOf(expectValue));
+ }
if (op != null) {
node.put("conditionOperator", op.name());
node.put("conditionOperatorCategory", op.getCategory().name());
diff --git
a/seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/command/MetadataExportCommandTest.java
b/seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/command/MetadataExportCommandTest.java
new file mode 100644
index 0000000000..19eb432fb7
--- /dev/null
+++
b/seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/command/MetadataExportCommandTest.java
@@ -0,0 +1,91 @@
+/*
+ * 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.core.starter.seatunnel.command;
+
+import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode;
+import
org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ObjectNode;
+
+import org.apache.seatunnel.api.common.PluginIdentifier;
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConditionExtension;
+import org.apache.seatunnel.api.configuration.util.Conditions;
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.common.constants.PluginType;
+import
org.apache.seatunnel.core.starter.seatunnel.args.MetadataExportCommandArgs;
+
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Method;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class MetadataExportCommandTest {
+
+ @Test
+ void shouldExportExtensionConditionDescription() throws Exception {
+ Option<Integer> port =
+
Options.key("port").intType().noDefaultValue().withDescription("Port number");
+ ConditionExtension<Integer> portRangeExtension =
+ new ConditionExtension<Integer>() {
+ @Override
+ public String description() {
+ return "must be between 1 and 65535";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, Integer
value) {
+ return value != null && value >= 1 && value <= 65535;
+ }
+ };
+ OptionRule optionRule =
+ OptionRule.builder()
+ .required(port, Conditions.extension(port,
portRangeExtension))
+ .build();
+
+ MetadataExportCommand command = new MetadataExportCommand(new
MetadataExportCommandArgs());
+ Method exportConnector =
+ MetadataExportCommand.class.getDeclaredMethod(
+ "exportConnector",
+ PluginIdentifier.class,
+ OptionRule.class,
+ PluginType.class);
+ exportConnector.setAccessible(true);
+
+ ObjectNode connectorNode =
+ (ObjectNode)
+ exportConnector.invoke(
+ command,
+ PluginIdentifier.of("seatunnel", "source",
"ExtensionSource"),
+ optionRule,
+ PluginType.SOURCE);
+
+ JsonNode valueConstraint =
connectorNode.get("valueConstraints").get(0);
+ assertTrue(
+ valueConstraint.get("expression").asText().contains("must be
between 1 and 65535"));
+
+ JsonNode conditionTree = valueConstraint.get("conditionTree");
+ assertEquals("port", conditionTree.get("key").asText());
+ assertEquals("must be between 1 and 65535",
conditionTree.get("expectValue").asText());
+ assertEquals("extension",
conditionTree.get("compareOperator").asText());
+ assertEquals("EXTENSION",
conditionTree.get("conditionOperator").asText());
+ assertEquals("EXTENSION",
conditionTree.get("conditionOperatorCategory").asText());
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/OptionRulesService.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/OptionRulesService.java
index 002a953003..a2f8c7ce57 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/OptionRulesService.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/OptionRulesService.java
@@ -282,9 +282,13 @@ public class OptionRulesService extends BaseService {
: null;
String conditionOperator = (op != null) ? op.name() : null;
String conditionOperatorCategory = (op != null) ?
op.getCategory().name() : null;
+ Object expectValue = condition.getExpectValue();
+ if (op == ConditionOperator.EXTENSION && condition.getExtension() !=
null) {
+ expectValue = condition.getExtension().description();
+ }
return new OptionRuleResponse.ConditionNode(
toOptionMetadata(condition.getOption()),
- condition.getExpectValue(),
+ expectValue,
compareOperatorSymbol,
compareOptionMeta,
conditionOperator,
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/service/OptionRulesServiceTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/service/OptionRulesServiceTest.java
index 151c3ec3e9..ac5c7bd4ab 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/service/OptionRulesServiceTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/service/OptionRulesServiceTest.java
@@ -20,7 +20,9 @@ package org.apache.seatunnel.engine.server.rest.service;
import org.apache.seatunnel.api.common.PluginIdentifier;
import org.apache.seatunnel.api.configuration.Option;
import org.apache.seatunnel.api.configuration.Options;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.configuration.SingleChoiceOption;
+import org.apache.seatunnel.api.configuration.util.ConditionExtension;
import org.apache.seatunnel.api.configuration.util.Conditions;
import org.apache.seatunnel.api.configuration.util.OptionRule;
import org.apache.seatunnel.engine.server.rest.response.OptionRuleResponse;
@@ -360,6 +362,47 @@ class OptionRulesServiceTest {
assertEquals("end_ts", tree.getCompareOption().getKey());
}
+ @Test
+ void shouldPreserveExtensionConstraintMetadata() {
+ Option<Integer> port =
+
Options.key("port").intType().noDefaultValue().withDescription("Port number");
+ ConditionExtension<Integer> portRangeExtension =
+ new ConditionExtension<Integer>() {
+ @Override
+ public String description() {
+ return "must be between 1 and 65535";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, Integer
value) {
+ return value != null && value >= 1 && value <= 65535;
+ }
+ };
+
+ OptionRule optionRule =
+ OptionRule.builder()
+ .required(port, Conditions.extension(port,
portRangeExtension))
+ .build();
+
+ OptionRuleResponse response =
+ service.buildResponse(
+ PluginIdentifier.of("seatunnel", "source",
"ExtensionSource"), optionRule);
+
+ List<OptionRuleResponse.ValueConstraintMetadata> constraints =
+ response.getOptionRule().getValueConstraints();
+ assertEquals(1, constraints.size());
+
+ OptionRuleResponse.ValueConstraintMetadata constraint =
constraints.get(0);
+ assertTrue(constraint.getExpression().contains("must be between 1 and
65535"));
+
+ OptionRuleResponse.ConditionNode tree = constraint.getConditionTree();
+ assertNotNull(tree);
+ assertEquals("must be between 1 and 65535", tree.getExpectValue());
+ assertEquals("extension", tree.getCompareOperator());
+ assertEquals("EXTENSION", tree.getConditionOperator());
+ assertEquals("EXTENSION", tree.getConditionOperatorCategory());
+ }
+
private enum AuthMode {
PASSWORD,
TOKEN
