This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12404-463adf8c47aab1f9da8f135121917281870d9b97 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 4cf17541df318472b5fffdfcbaab2b1a33ec31c8 Author: TianHengZhuang <[email protected]> AuthorDate: Tue Sep 29 13:31:39 2026 +0000 [Improve][Connector-V2][Fake] Declarative optionRule value constraints for fake source (#12404) --- .../introduction/concepts/incompatible-changes.md | 12 ++ .../introduction/concepts/incompatible-changes.md | 9 + .../seatunnel/fake/source/FakeSourceFactory.java | 53 ++++- .../seatunnel/fake/source/FakeFactoryTest.java | 228 ++++++++++++++++++++- 4 files changed, 293 insertions(+), 9 deletions(-) diff --git a/docs/en/introduction/concepts/incompatible-changes.md b/docs/en/introduction/concepts/incompatible-changes.md index 3d1e871d57..5719f6fa40 100644 --- a/docs/en/introduction/concepts/incompatible-changes.md +++ b/docs/en/introduction/concepts/incompatible-changes.md @@ -52,6 +52,18 @@ You need to check this document before you upgrade to related version. trust store of the SeaTunnel runtime, or switch to the `host`/`port` + `ssl = true` configuration with a properly configured trust store. +### FakeSource (connector-fake) + +- Declarative option constraints are now enforced at factory validation time instead of + silently passing and failing only at runtime. Affected options: `split.num`, + `vector.dimension` and `binary.vector.dimension` must be > 0; `row.num`, + `split.read-interval`, `map.size`, `array.size`, `bytes.length` and `string.length` must be + >= 0; `tinyint.min/max`, `smallint.min/max`, `int.min/max`, `bigint.min/max`, + `float.min/max`, `double.min/max` and `vector.float.min/max` must satisfy min <= max. + Note that `row.num = 0` (empty source) is still valid. Existing jobs that set invalid + values and previously ran successfully will now fail fast at startup with a validation + error. + ### Zeta REST Pagination Parameter Validation - **Behavior change: `page` and `rows` are validated on paginated endpoints** diff --git a/docs/zh/introduction/concepts/incompatible-changes.md b/docs/zh/introduction/concepts/incompatible-changes.md index 90384b9c00..17912d0650 100644 --- a/docs/zh/introduction/concepts/incompatible-changes.md +++ b/docs/zh/introduction/concepts/incompatible-changes.md @@ -26,6 +26,15 @@ - **迁移指南**：将 Broker 证书（或私有 CA 证书链）导入 SeaTunnel 运行时的 JVM 信任库，或改用 `host`/`port` + `ssl = true` 配置并正确设置信任库。 +### FakeSource (connector-fake) + +- 声明式选项约束现在在工厂校验阶段即强制生效，而不再静默放行、直到运行时才失败。受影响选项：`split.num`、 + `vector.dimension` 和 `binary.vector.dimension` 必须 > 0；`row.num`、`split.read-interval`、`map.size`、 + `array.size`、`bytes.length` 和 `string.length` 必须 >= 0；`tinyint.min/max`、`smallint.min/max`、 + `int.min/max`、`bigint.min/max`、`float.min/max`、`double.min/max` 和 `vector.float.min/max` + 必须满足 min <= max。注意 `row.num = 0`（空 Source）仍然有效。此前设置了无效值且成功运行的现有作业， + 将在启动时快速抛出校验错误并失败。 + ### Zeta REST 分页参数校验 - **行为变更：分页接口开始校验 `page` 与 `rows`** diff --git a/seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceFactory.java b/seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceFactory.java index 556b5bf5a2..5bf75cfc07 100644 --- a/seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceFactory.java +++ b/seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceFactory.java @@ -17,6 +17,7 @@ package org.apache.seatunnel.connectors.seatunnel.fake.source; +import org.apache.seatunnel.api.configuration.util.Conditions; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.options.ConnectorCommonOptions; import org.apache.seatunnel.api.source.SeaTunnelSource; @@ -39,6 +40,8 @@ import java.util.stream.Collectors; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.ARRAY_SIZE; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.BIGINT_FAKE_MODE; +import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.BIGINT_MAX; +import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.BIGINT_MIN; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.BIGINT_TEMPLATE; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.BINARY_VECTOR_DIMENSION; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.BYTES_LENGTH; @@ -46,26 +49,39 @@ import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOp import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.DATE_MONTH_TEMPLATE; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.DATE_YEAR_TEMPLATE; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.DOUBLE_FAKE_MODE; +import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.DOUBLE_MAX; +import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.DOUBLE_MIN; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.DOUBLE_TEMPLATE; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.FLOAT_FAKE_MODE; +import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.FLOAT_MAX; +import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.FLOAT_MIN; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.FLOAT_TEMPLATE; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.INT_FAKE_MODE; +import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.INT_MAX; +import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.INT_MIN; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.INT_TEMPLATE; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.MAP_SIZE; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.ROWS; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.ROW_NUM; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.SMALLINT_FAKE_MODE; +import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.SMALLINT_MAX; +import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.SMALLINT_MIN; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.SMALLINT_TEMPLATE; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.SPLIT_NUM; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.SPLIT_READ_INTERVAL; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.STRING_FAKE_MODE; +import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.STRING_LENGTH; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.STRING_TEMPLATE; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.TIME_HOUR_TEMPLATE; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.TIME_MINUTE_TEMPLATE; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.TIME_SECOND_TEMPLATE; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.TINYINT_FAKE_MODE; +import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.TINYINT_MAX; +import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.TINYINT_MIN; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.TINYINT_TEMPLATE; import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.VECTOR_DIMENSION; +import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.VECTOR_FLOAT_MAX; +import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.VECTOR_FLOAT_MIN; @AutoService(Factory.class) public class FakeSourceFactory implements TableSourceFactory, SupportSourceDryRunValidation { @@ -78,6 +94,35 @@ public class FakeSourceFactory implements TableSourceFactory, SupportSourceDryRu public OptionRule optionRule() { return OptionRule.builder() .exclusive(ConnectorCommonOptions.TABLE_CONFIGS, ConnectorCommonOptions.SCHEMA) + // row.num=0 stays valid (empty source); split.num must be positive + .optional(ROW_NUM, Conditions.greaterOrEqual(ROW_NUM, 0)) + .optional(SPLIT_NUM, Conditions.greaterThan(SPLIT_NUM, 0)) + .optional(SPLIT_READ_INTERVAL, Conditions.greaterOrEqual(SPLIT_READ_INTERVAL, 0)) + .optional(MAP_SIZE, Conditions.greaterOrEqual(MAP_SIZE, 0)) + .optional(ARRAY_SIZE, Conditions.greaterOrEqual(ARRAY_SIZE, 0)) + .optional(BYTES_LENGTH, Conditions.greaterOrEqual(BYTES_LENGTH, 0)) + .optional(STRING_LENGTH, Conditions.greaterOrEqual(STRING_LENGTH, 0)) + .optional(VECTOR_DIMENSION, Conditions.greaterThan(VECTOR_DIMENSION, 0)) + .optional( + BINARY_VECTOR_DIMENSION, Conditions.greaterThan(BINARY_VECTOR_DIMENSION, 0)) + .optional( + TINYINT_MIN, + TINYINT_MAX, + Conditions.lessOrEqualField(TINYINT_MIN, TINYINT_MAX)) + .optional( + SMALLINT_MIN, + SMALLINT_MAX, + Conditions.lessOrEqualField(SMALLINT_MIN, SMALLINT_MAX)) + .optional(INT_MIN, INT_MAX, Conditions.lessOrEqualField(INT_MIN, INT_MAX)) + .optional( + BIGINT_MIN, BIGINT_MAX, Conditions.lessOrEqualField(BIGINT_MIN, BIGINT_MAX)) + .optional(FLOAT_MIN, FLOAT_MAX, Conditions.lessOrEqualField(FLOAT_MIN, FLOAT_MAX)) + .optional( + DOUBLE_MIN, DOUBLE_MAX, Conditions.lessOrEqualField(DOUBLE_MIN, DOUBLE_MAX)) + .optional( + VECTOR_FLOAT_MIN, + VECTOR_FLOAT_MAX, + Conditions.lessOrEqualField(VECTOR_FLOAT_MIN, VECTOR_FLOAT_MAX)) .optional( STRING_FAKE_MODE, TINYINT_FAKE_MODE, @@ -87,14 +132,6 @@ public class FakeSourceFactory implements TableSourceFactory, SupportSourceDryRu FLOAT_FAKE_MODE, DOUBLE_FAKE_MODE, ROWS, - ROW_NUM, - SPLIT_NUM, - SPLIT_READ_INTERVAL, - MAP_SIZE, - ARRAY_SIZE, - BYTES_LENGTH, - VECTOR_DIMENSION, - BINARY_VECTOR_DIMENSION, DATE_YEAR_TEMPLATE, DATE_MONTH_TEMPLATE, DATE_DAY_TEMPLATE, diff --git a/seatunnel-connectors-v2/connector-fake/src/test/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeFactoryTest.java b/seatunnel-connectors-v2/connector-fake/src/test/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeFactoryTest.java index 6dd59e7c39..c35c6d1807 100644 --- a/seatunnel-connectors-v2/connector-fake/src/test/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeFactoryTest.java +++ b/seatunnel-connectors-v2/connector-fake/src/test/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeFactoryTest.java @@ -17,13 +17,239 @@ package org.apache.seatunnel.connectors.seatunnel.fake.source; +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.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + public class FakeFactoryTest { + private final FakeSourceFactory factory = new FakeSourceFactory(); + + private static Map<String, Object> validBaseConfig() { + Map<String, Object> map = new HashMap<>(); + Map<String, Object> schema = new HashMap<>(); + schema.put("fields", new HashMap<String, Object>()); + map.put("schema", schema); + return map; + } + + private void validate(Map<String, Object> map) { + ConfigValidator.of(ReadonlyConfig.fromMap(map)).validate(factory.optionRule()); + } + @Test void optionRule() { - Assertions.assertNotNull((new FakeSourceFactory()).optionRule()); + Assertions.assertNotNull(factory.optionRule()); + } + + @Test + void validConfigPassesValidation() { + validate(validBaseConfig()); + } + + @Test + void validNumericBoundsPassValidation() { + Map<String, Object> map = validBaseConfig(); + map.put("row.num", 1); + map.put("split.num", 1); + map.put("split.read-interval", 0); + map.put("map.size", 0); + map.put("array.size", 0); + map.put("bytes.length", 0); + map.put("string.length", 0); + map.put("vector.dimension", 1); + map.put("binary.vector.dimension", 8); + validate(map); + } + + @Test + void rowNumZeroPassesValidation() { + Map<String, Object> map = validBaseConfig(); + map.put("row.num", 0); + validate(map); + } + + @Test + void minEqualsMaxPassesValidation() { + Map<String, Object> map = validBaseConfig(); + map.put("int.min", 5); + map.put("int.max", 5); + map.put("tinyint.min", 1); + map.put("tinyint.max", 1); + map.put("smallint.min", 2); + map.put("smallint.max", 2); + map.put("bigint.min", 3L); + map.put("bigint.max", 3L); + map.put("float.min", 1.5f); + map.put("float.max", 1.5f); + map.put("double.min", 2.5d); + map.put("double.max", 2.5d); + map.put("vector.float.min", 0.5f); + map.put("vector.float.max", 0.5f); + validate(map); + } + + @Test + void oneSidedMinOrMaxPassesValidation() { + Map<String, Object> onlyMin = validBaseConfig(); + onlyMin.put("int.min", 0); + validate(onlyMin); + + Map<String, Object> onlyMax = validBaseConfig(); + onlyMax.put("int.max", 100); + validate(onlyMax); + } + + @Test + void exclusiveUsesListForTableConfigs() { + Map<String, Object> map = validBaseConfig(); + List<Map<String, Object>> tables = Collections.singletonList(new HashMap<>()); + map.put("tables_configs", tables); + OptionValidationException ex = + Assertions.assertThrows(OptionValidationException.class, () -> validate(map)); + String message = ex.getMessage(); + Assertions.assertTrue( + message.contains("schema") || message.contains("tables_configs"), + () -> "expected exclusive message, got: " + message); + } + + @Test + void neitherSchemaNorTableConfigsRejected() { + Assertions.assertThrows(OptionValidationException.class, () -> validate(new HashMap<>())); + } + + @Test + void nonPositiveSplitNumIsRejected() { + Map<String, Object> map = validBaseConfig(); + map.put("split.num", 0); + Assertions.assertThrows(OptionValidationException.class, () -> validate(map)); + } + + @Test + void negativeMapSizeIsRejected() { + Map<String, Object> map = validBaseConfig(); + map.put("map.size", -1); + Assertions.assertThrows(OptionValidationException.class, () -> validate(map)); + } + + @Test + void negativeArraySizeIsRejected() { + Map<String, Object> map = validBaseConfig(); + map.put("array.size", -1); + Assertions.assertThrows(OptionValidationException.class, () -> validate(map)); + } + + @Test + void negativeBytesLengthIsRejected() { + Map<String, Object> map = validBaseConfig(); + map.put("bytes.length", -1); + Assertions.assertThrows(OptionValidationException.class, () -> validate(map)); + } + + @Test + void negativeStringLengthIsRejected() { + Map<String, Object> map = validBaseConfig(); + map.put("string.length", -1); + Assertions.assertThrows(OptionValidationException.class, () -> validate(map)); + } + + @Test + void negativeSplitReadIntervalIsRejected() { + Map<String, Object> map = validBaseConfig(); + map.put("split.read-interval", -1); + Assertions.assertThrows(OptionValidationException.class, () -> validate(map)); + } + + @Test + void nonPositiveVectorDimensionIsRejected() { + Map<String, Object> map = validBaseConfig(); + map.put("vector.dimension", 0); + Assertions.assertThrows(OptionValidationException.class, () -> validate(map)); + } + + @Test + void nonPositiveBinaryVectorDimensionIsRejected() { + Map<String, Object> map = validBaseConfig(); + map.put("binary.vector.dimension", 0); + Assertions.assertThrows(OptionValidationException.class, () -> validate(map)); + } + + @Test + void tinyintMinGreaterThanMaxIsRejected() { + Map<String, Object> map = validBaseConfig(); + map.put("tinyint.min", 10); + map.put("tinyint.max", 5); + Assertions.assertThrows(OptionValidationException.class, () -> validate(map)); + } + + @Test + void smallintMinGreaterThanMaxIsRejected() { + Map<String, Object> map = validBaseConfig(); + map.put("smallint.min", 10); + map.put("smallint.max", 5); + Assertions.assertThrows(OptionValidationException.class, () -> validate(map)); + } + + @Test + void intMinGreaterThanIntMaxIsRejected() { + Map<String, Object> map = validBaseConfig(); + map.put("int.min", 10); + map.put("int.max", 5); + Assertions.assertThrows(OptionValidationException.class, () -> validate(map)); + } + + @Test + void bigintMinGreaterThanMaxIsRejected() { + Map<String, Object> map = validBaseConfig(); + map.put("bigint.min", 10L); + map.put("bigint.max", 5L); + Assertions.assertThrows(OptionValidationException.class, () -> validate(map)); + } + + @Test + void floatMinGreaterThanMaxIsRejected() { + Map<String, Object> map = validBaseConfig(); + map.put("float.min", 2.0f); + map.put("float.max", 1.0f); + Assertions.assertThrows(OptionValidationException.class, () -> validate(map)); + } + + @Test + void doubleMinGreaterThanMaxIsRejected() { + Map<String, Object> map = validBaseConfig(); + map.put("double.min", 2.0d); + map.put("double.max", 1.0d); + Assertions.assertThrows(OptionValidationException.class, () -> validate(map)); + } + + @Test + void vectorFloatMinGreaterThanMaxIsRejected() { + Map<String, Object> map = validBaseConfig(); + map.put("vector.float.min", 2.0f); + map.put("vector.float.max", 1.0f); + Assertions.assertThrows(OptionValidationException.class, () -> validate(map)); + } + + @Test + void templateModeWithoutTemplateIsRejected() { + Map<String, Object> map = validBaseConfig(); + map.put("string.fake.mode", "TEMPLATE"); + Assertions.assertThrows(OptionValidationException.class, () -> validate(map)); + } + + @Test + void templateModeWithTemplatePassesValidation() { + Map<String, Object> map = validBaseConfig(); + map.put("string.fake.mode", "TEMPLATE"); + map.put("string.template", Collections.singletonList("hello")); + validate(map); } }
