[Umbrella][Improve] Migrate Connector/Transform Validation to Declarative OptionRule — Claim & Tracking

We need community help to migrate connector and transform validation from imperative `if/throw` checks to declarative `optionRule()` + `Conditions.*`.

If the `Issue` column shows an existing issue number, please claim in that issue.
If the `Issue` column shows `This issue`, please comment here to claim it.

## Prerequisites / Framework Status

Phase 1 (framework) has been completed:

- Design and scope audit: https://github.com/apache/seatunnel/issues/10976
- Framework implementation: https://github.com/apache/seatunnel/pull/10977

This umbrella issue is for **Phase 2 migration and tracking** only.

## Migration Guide

### A. Numeric range

Use declarative constraints for rules like `port > 0`, `batchSize >= 0`.

```java
import static org.apache.seatunnel.api.configuration.util.Conditions.greaterThan;

OptionRule.builder()
.required(PORT, greaterThan(PORT, 0))
.build();
```

### B1. Required cross-field comparison

Use when both fields are mandatory and have relation constraints.

```java
import static org.apache.seatunnel.api.configuration.util.Conditions.lessThanField;

OptionRule.builder()
.required(START_TIMESTAMP, END_TIMESTAMP, lessThanField(START_TIMESTAMP, END_TIMESTAMP))
.build();
```

### B2. Optional cross-field comparison

Use when fields are not mandatory, but must satisfy relation when present.

```java
import static org.apache.seatunnel.api.configuration.util.Conditions.lessOrEqualField;

OptionRule.builder()
.optional(MIN_VALUE, MAX_VALUE, lessOrEqualField(MIN_VALUE, MAX_VALUE))
.build();
```

Do not accidentally convert optional semantics into required semantics.

### C. Conditional value check

Use when a trigger option enables another constraint.

```java
import static org.apache.seatunnel.api.configuration.util.Conditions.greaterThan;

OptionRule.builder()
.conditional(
IGNORE_NO_LEADER_PARTITION,
true,
greaterThan(PARTITION_DISCOVERY_INTERVAL_MILLIS, 0))
.build();
```

### D1. Deprecated key coexistence (`withFallbackKeys` → `exclusive`)

When migrating a deprecated key to a new key name, replace `withFallbackKeys` + runtime
`sourceMap` mutual-exclusion check with `exclusive()`, so the conflict is caught at
`--check` time rather than silently at runtime.

Before:

```java
public static final Option> KEY_REPLACE_FIELDS =
Options.key("replace_fields").listType().noDefaultValue()
.withFallbackKeys("replace_field");

// runtime sourceMap check, invisible to --check
Map sourceMap = config.getSourceMap();
if (sourceMap.containsKey("replace_field") && sourceMap.containsKey("replace_fields")) {
throw ...;
}

this.replaceFields.addAll(getRequiredOption(config, KEY_REPLACE_FIELDS));
```

After:

```java
@Deprecated
public static final Option> KEY_REPLACE_FIELD =
Options.key("replace_field").listType().noDefaultValue();

public static final Option> KEY_REPLACE_FIELDS =
Options.key("replace_fields").listType().noDefaultValue();

// exclusive() validates at --check time: exactly one must be present
OptionRule.builder()
.exclusive(KEY_REPLACE_FIELD, KEY_REPLACE_FIELDS)
.build();

if (config.getOptional(KEY_REPLACE_FIELDS).isPresent()) {
this.fields = config.get(KEY_REPLACE_FIELDS);
} else {
this.fields = config.get(KEY_REPLACE_FIELD);
}
```
### D2. Exclusive with value constraints

Use when exclusive options also need content validation (e.g. non-empty).

> Framework support:
> - [PR #11010](https://github.com/apache/seatunnel/pull/11010) — Map condition validators: `mapNotEmpty`, `mapContainsKey`, `mapContainsKeys` (merged)
> - [PR #11022](https://github.com/apache/seatunnel/pull/11022) — Allow `exclusive`/`bundled` + `optional(condition)` coexistence (merged)

Before:

```java
// runtime imperative checks — invisible to --check
if (config.get(SCHEMA) != null && config.get(TABLE_CONFIGS) != null) {
throw new IllegalArgumentException("Cannot specify both 'schema' and 'table_configs'");
}
if (config.get(SCHEMA) != null && config.get(SCHEMA).isEmpty()) {
throw new IllegalArgumentException("'schema' must not be empty");
}
```

After:

```java
import static org.apache.seatunnel.api.configuration.util.Conditions.mapNotEmpty;
import static org.apache.seatunnel.api.configuration.util.Conditions.notEmpty;

OptionRule.builder()
.exclusive(SCHEMA, TABLE_CONFIGS)
.optional(SCHEMA, mapNotEmpty(SCHEMA))
.optional(TABLE_CONFIGS, notEmpty(TABLE_CONFIGS))
.build();
```

- `exclusive()` — exactly one must be present
- `optional(option, condition)` — when present, value must satisfy the constraint
- Both can coexist on the same option; constraints are only evaluated when the option is present

### D3. Custom structural validation (`ConditionExtension`)

Use when built-in operators cannot express the validation — for example, validating internal structure of `List>`, or enforcing constraints across nested child configs like `table_configs`.

> Framework support:
>
> - [PR #11048](https://github.com/apache/seatunnel/pull/11048) — `ConditionOperator.EXTENSION` + `ConditionExtension` interface (merged)

Before:

```java
// imperative structural check buried in buildWithConfig() — invisible to --check
for (Map child : tableConfigs) {
if (!child.containsKey("table_name")) {
throw new IllegalArgumentException("each table config must contain 'table_name'");
}
}
// cross-element uniqueness check
if (tableNames.size() != new HashSet<>(tableNames).size()) {
throw new IllegalArgumentException("table names must be unique");
}
```

After:

```java
import org.apache.seatunnel.api.configuration.util.ConditionExtension;
import org.apache.seatunnel.api.configuration.util.Conditions;

static class TableConfigsValidator
implements ConditionExtension>> {
@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> value)
throws OptionValidationException {
if (value.isEmpty()) {
return false;
}
Set seen = new HashSet<>();
for (Map 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;
}
}

OptionRule.builder()
.exclusive(TABLE_CONFIGS, SCHEMA)
.optional(TABLE_CONFIGS,
Conditions.extension(TABLE_CONFIGS, new TableConfigsValidator()))
.build();
```

* `ConditionExtension` — implement `description()` (used in error messages and REST metadata) and `evaluate()` (validation logic, avoid I/O)
* `Conditions.extension(Option, ConditionExtension)` — compile-time type binding between option and extension
* Chains with `.and()` / `.or()` like any built-in operator
* Return `false` for auto-composed error messages, or throw `OptionValidationException` for context-rich details

### E. Keep runtime checks for cases not suitable for declarative rules

Do not migrate these into declarative rules:

* external system/state validation (DB/metastore/network reachability)
* complex parser/semantic validation (SQL parsing, advanced regex semantics)
* runtime context checks (current time, execution topology state)

## Connectors Open for Claim

| Type | Connector | Contributer | Status | PR |
| --- | --- | --- | --- | --- |
| Source | connector-cdc | @nzw921rx | Claimed | #11023 |
| Source | connector-edge-socket | @nzw921rx | Merged | #11384 |
| Source | connector-fake | @Ayushkale11 | In review | #11032 |
| Source | connector-google-sheets | @KaustAbhinand | Completed(no migration needed) | #11943 |
| Source | connector-openmldb | @Nikk8091 | Merged | #12181 |
| Source | connector-web3j | @Nikk8091 | Merged | #12180 |
| Sink | connector-activemq | @Nikk8091 | Merged | #12004 |
| Sink | connector-aerospike | @hyoj-dev | Merged | #12090 |
| Sink | connector-assert | @cyl-uuu | In review | #11625 |
| Sink | connector-bigquery | @yigitcan-ozturk | Merged | #12272 |
| Sink | connector-console | @asrajawat | Merged | #11477 |
| Sink | connector-datahub | @yigitcan-ozturk | Merged | #12174 |
| Sink | connector-dingtalk | @Nikk8091 | Merged | #12170 |
| Sink | connector-druid | @smoggy666 | Merged | #12250 |
| Sink | connector-email | @abolfazlmadanii | Merged | #11817 |
| Sink | connector-fluss | | Todo | |
| Sink | connector-google-firestore | @1328837476-hug | Merged | #12278 |
| Sink | connector-hugegraph | @Nikk8091 | Merged | #12075 |
| Sink | connector-hudi | @zhang-arvin | Completed(no migration needed) | |
| Sink | connector-lance | @ZYZ666-RGB | Merged | #12222 |
| Sink | connector-mqtt | @ealeonraz | In review | #11461 |
| Sink | connector-s3-redshift | @yigitcan-ozturk | Merged | #12274 |
| Sink | connector-selectdb-cloud | @Nikk8091 | Claimed | |
| Sink | connector-sensorsdata | | Todo | |
| Sink | connector-sentry | @yigitcan-ozturk | Merged | #12148 |
| Sink | connector-slack | @1328837476-hug | Merged | #12178 |
| Both | connector-amazondynamodb | @goutamadwant | Merged | #11821 |
| Both | connector-amazonsqs | @xinnyuli | Merged | #12217 |
| Both | connector-cassandra | @Aryadeepta | Merged | #11964 |
| Both | connector-clickhouse | | Todo | |
| Both | connector-databend | @BobSong-dev | Merged | #11831 |
| Both | connector-doris | @zhang-arvin | Merged | #11858 |
| Both | connector-easysearch | | Todo | |
| Both | connector-elasticsearch | @nzw921rx | Merged | #11122 |
| Both | connector-file | @RohanExploit | Merged | #11881 |
| Both | connector-graphql | @KaustAbhinand | Merged | #12273 |
| Both | connector-hbase | @goutamadwant | Merged | #11803 |
| Both | connector-hive | @zhang-arvin | Completed(no migration needed) | |
| Both | connector-http | @junsoo22 | Claimed | |
| Both | connector-iceberg | @ClaireLytt | Merged | #11921 |
| Both | connector-influxdb | | Todo | |
| Both | connector-iotdb | | Todo | |
| Both | connector-iotdb-v2 | @liziing | Merged | #11839 |
| Both | connector-jdbc | @nzw921rx | Merged | #11106 |
| Both | connector-kafka | @nzw921rx | Merged | #11157 |
| Both | connector-kudu | | Todo | |
| Both | connector-maxcompute | | Todo | |
| Both | connector-milvus | @ZYZ666-RGB | Merged | #11504 |
| Both | connector-mongodb | @AmanMishra1996 | Merged | #11886 |
| Both | connector-neo4j | @jjj-n | Merged | #12047 |
| Both | connector-paimon | | Todo | |
| Both | connector-prometheus | @goutamadwant | Merged | #11738 |
| Both | connector-pulsar | @Linz1248 | Merged | #11985 |
| Both | connector-qdrant | @yigitcan-ozturk | Completed(no migration needed) | |
| Both | connector-rabbitmq | @cyl-uuu | Merged | #11795 |
| Both | connector-redis | @ss666 | Merged | #11225|
| Both | connector-rocketmq | @nzw921rx | Merged | #11158 |
| Both | connector-sls | @Nikk8091 | Claimed | |
| Both | connector-socket | @nzw921rx | Merged | #11214 |
| Both | connector-starrocks | @zhang-arvin | Completed(no migration needed) | |
| Both | connector-tablestore | | Todo | |
| Both | connector-tdengine | | Todo | |
| Both | connector-typesense | @yigitcan-ozturk | Merged | #12175 |

## Transforms Open for Claim

| Type | Transform | Contributer | Status | PR |
| --- | --- | --- | --- | --- |
| Transform | CopyField | @nzw921rx | Merged | #11095 |
| Transform | DataValidator | @nzw921rx | Merged | #11095 |
| Transform | DynamicCompile | @nzw921rx | Merged | #11095 |
| Transform | FieldEncrypt | @nzw921rx | Merged | #11095 |
| Transform | FieldMapper | @nzw921rx | Merged | #11095 |
| Transform | FilterField | @nzw921rx | Merged | #11095 |
| Transform | FilterRowKind | @goutamadwant | Merged | #11763 |
| Transform | JsonPath | @nzw921rx | Merged | #11095 |
| Transform | Metadata | @nzw921rx | Merged | #11095 |
| Transform | RegexExtract | @nzw921rx | Merged | #11095 |
| Transform | Replace | @nzw921rx | Merged | #11095 |
| Transform | RowKindExtractor| @nzw921rx | Merged | #11095 |
| Transform | Split | @nzw921rx | Merged | #11095 |
| Transform | SQL | @nzw921rx | Merged | #11095 |
| Transform | TableFilter | @nzw921rx | Merged | #11095 |
| Transform | TableMerge | @nzw921rx | Merged | #11095 |
| Transform | DefineSinkType | @nzw921rx | Merged | #11095 |

## Note

`connector-common` is a shared base module, not a standalone connector plugin, so it is excluded from claim rows.

## How to Contribute

1. **Pick a connector/transform**: Choose one from the lists above.
2. **Claim the task**: Comment on this issue (for example: `I would like to work on connector-kafka` or `I would like to work on FilterRowKind`).
3. **Implement**:
- Migrate declarative-eligible validation rules from imperative `if/throw` to `optionRule()` + `Conditions.*`.
- Keep runtime-only validation in runtime code paths (`*Config.java`, sink/source initialization) when it depends on external state or execution context.
- Distinguish cross-field semantics explicitly:
- required cross-field: `required(A, B, lessThanField(A, B))`
- optional cross-field: `optional(A, B, lessOrEqualField(A, B))`
4. **Reference**:
- Framework scope and audit baseline: [#10976](https://github.com/apache/seatunnel/issues/10976)
- Declarative framework implementation: [#10977](https://github.com/apache/seatunnel/pull/10977)
5. **Submit PR**: Open a Pull Request against `dev`, and link it back to this issue.

Thank you for your contribution.
Please leave a message if you'd like to implement the declarative validation migration for any connector or transform.

## Code of Conduct

I agree to follow this project's Code of Conduct.

Start by selecting an unclaimed Todo connector or transform in the tracking tables, then inspect its existing imperative validation and the OptionRule/Conditions migration patterns in this issue. Use the named OptionRule, Conditions.*, and ConditionExtension entry points; done means suitable validation is declarative, runtime-only checks remain, and the component's validation is verified in a pull request.

Written by the indexing model from the issue text.