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-12746-613803559a11a63b0bd6a138122d323a935e5def in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit c37823b9a3db30e7455acf11415a89d5b0a1c7e9 Author: Ryan Yang <[email protected]> AuthorDate: Thu Oct 15 07:13:29 2026 +0000 [Improve][Connector-V2][Vitess] Migrate CDC source option validation to declarative OptionRule (#12746) --- docs/en/connectors/source/Vitess-CDC.md | 5 + docs/zh/connectors/source/Vitess-CDC.md | 2 + .../vitess/config/VitessSchemaMetadataGuards.java | 184 +++++++++++++++ .../cdc/vitess/source/VitessSourceFactory.java | 39 +++- .../cdc/vitess/source/VitessSourceFactoryTest.java | 247 +++++++++++++++++++++ 5 files changed, 467 insertions(+), 10 deletions(-) diff --git a/docs/en/connectors/source/Vitess-CDC.md b/docs/en/connectors/source/Vitess-CDC.md index 1e2202b0ec..301bc7bd83 100644 --- a/docs/en/connectors/source/Vitess-CDC.md +++ b/docs/en/connectors/source/Vitess-CDC.md @@ -82,6 +82,11 @@ separately. ## Notes +- The schema metadata contract is validated when the job is submitted: a `schema` block must declare + `columns` or `metadata_table_id`, every `tables_configs` entry must contain its own `schema` block, + and `table-names` / `table-pattern` are mutually exclusive. +- Checks that need the catalog - a declared table missing from the schema metadata, or a pattern that + matches no table - still run when the source is created. - The first delivery does not read an initial table snapshot. - Dynamic discovery of newly added tables is out of scope. - Schema evolution events are not emitted in this first delivery. diff --git a/docs/zh/connectors/source/Vitess-CDC.md b/docs/zh/connectors/source/Vitess-CDC.md index 7c4e9d5cae..6bb6f5dbe2 100644 --- a/docs/zh/connectors/source/Vitess-CDC.md +++ b/docs/zh/connectors/source/Vitess-CDC.md @@ -80,6 +80,8 @@ JDBC 做验证或示例下游，请额外准备 MySQL JDBC 驱动。 ## 说明 +- schema 元数据契约会在作业提交时校验：`schema` 块必须声明 `columns` 或 `metadata_table_id`，每个 `tables_configs` 元素都必须包含自己的 `schema` 块，`table-names` 与 `table-pattern` 互斥。 +- 需要访问 catalog 的检查（声明的表在 schema 元数据中不存在、正则未匹配到任何表）仍在创建 source 时执行。 - 第一版不支持初始快照读取。 - 第一版不支持运行时动态发现新表。 - 第一版不发送 schema evolution 事件。 diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-vitess/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/vitess/config/VitessSchemaMetadataGuards.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-vitess/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/vitess/config/VitessSchemaMetadataGuards.java new file mode 100644 index 0000000000..4d55c0d530 --- /dev/null +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-vitess/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/vitess/config/VitessSchemaMetadataGuards.java @@ -0,0 +1,184 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.cdc.vitess.config; + +import org.apache.seatunnel.api.configuration.Option; +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.configuration.util.ConditionExtension; +import org.apache.seatunnel.api.configuration.util.OptionValidationException; +import org.apache.seatunnel.api.options.ConnectorCommonOptions; +import org.apache.seatunnel.api.options.table.ColumnOptions; + +import java.util.List; +import java.util.Map; + +/** + * Reusable {@link ConditionExtension} guards for the Vitess CDC schema metadata contract. + * + * <p>Vitess does not provide a SeaTunnel catalog yet, so a source has to describe the tables it + * reads explicitly through {@code schema} or {@code tables_configs}, and the declarative rule + * cannot look inside those structures. Each guard below attaches one of the checks the source + * applies while the job is built to the option it belongs to, so a configuration that cannot + * produce deterministic table schemas is rejected when the job is submitted instead of when it + * starts. The predicate and the wording are shared with the source through {@link + * #declaresSchemaMetadata(Map)}, {@link #missingSchemaBlockMessage(String)} and {@link + * #missingMetadataMessage(String)}, so an operator sees the same message either way. The schema + * guards report that message instead of a plain description because the offending entry has to be + * named, while the selector guard returns {@code false} and lets the rule render its description. + * + * <p>Runtime behavior is unchanged. A configuration accepted here can still fail later when the + * declared tables are resolved against the source metadata, because that requires reading the + * source and is intentionally not part of the rule. + */ +public final class VitessSchemaMetadataGuards { + + private VitessSchemaMetadataGuards() {} + + /** + * Returns whether a schema block declares its columns directly or refers to the metadata table. + */ + public static boolean declaresSchemaMetadata(Map<String, Object> schema) { + return schema != null + && (schema.containsKey(ColumnOptions.COLUMNS.key()) + || schema.containsKey(ColumnOptions.METADATA_TABLE_ID.key())); + } + + /** Message for a schema block that declares neither columns nor a metadata table id. */ + public static String missingMetadataMessage(String optionName) { + return String.format( + "Vitess CDC requires explicit columns or metadata_table_id in '%s' so table schemas stay deterministic.", + optionName); + } + + /** Message for a {@code tables_configs} entry that does not carry a schema block. */ + public static String missingSchemaBlockMessage(String optionName) { + return String.format( + "Vitess CDC requires explicit schema metadata. '%s' must contain a schema block.", + optionName); + } + + /** Rejects a {@code schema} that declares neither columns nor a metadata table id. */ + public static ConditionExtension<Map<String, Object>> rootSchemaGuard( + Option<Map<String, Object>> schemaOption) { + return new RootSchemaGuard(schemaOption.key()); + } + + /** + * Rejects {@code tables_configs} entries that do not carry a schema block declaring columns or + * a metadata table id. An empty list is left to the not-empty constraint of the same option. + */ + public static ConditionExtension<List<Map<String, Object>>> tableConfigsGuard( + Option<List<Map<String, Object>>> tableConfigsOption) { + return new TableConfigsGuard(tableConfigsOption.key()); + } + + /** Rejects a configuration that sets both {@code table-names} and {@code table-pattern}. */ + public static ConditionExtension<List<String>> tableNamesGuard( + Option<String> tablePatternOption) { + return new TableNamesGuard(tablePatternOption); + } + + static final class RootSchemaGuard implements ConditionExtension<Map<String, Object>> { + + private final String guardedKey; + + RootSchemaGuard(String guardedKey) { + this.guardedKey = guardedKey; + } + + @Override + public String description() { + return "must declare '" + + ColumnOptions.COLUMNS.key() + + "' or '" + + ColumnOptions.METADATA_TABLE_ID.key() + + "' so table schemas stay deterministic"; + } + + @Override + public boolean evaluate(ReadonlyConfig config, Map<String, Object> value) { + if (declaresSchemaMetadata(value)) { + return true; + } + throw new OptionValidationException(missingMetadataMessage(guardedKey)); + } + } + + static final class TableConfigsGuard implements ConditionExtension<List<Map<String, Object>>> { + + private final String guardedKey; + + TableConfigsGuard(String guardedKey) { + this.guardedKey = guardedKey; + } + + @Override + public String description() { + return "must contain a '" + + ConnectorCommonOptions.SCHEMA.key() + + "' block declaring '" + + ColumnOptions.COLUMNS.key() + + "' or '" + + ColumnOptions.METADATA_TABLE_ID.key() + + "' for every entry"; + } + + @Override + @SuppressWarnings("unchecked") + public boolean evaluate(ReadonlyConfig config, List<Map<String, Object>> value) { + if (value == null) { + return true; + } + for (int index = 0; index < value.size(); index++) { + Map<String, Object> tableConfig = value.get(index); + Object schemaValue = + tableConfig == null + ? null + : tableConfig.get(ConnectorCommonOptions.SCHEMA.key()); + String optionName = guardedKey + "[" + index + "]"; + if (!(schemaValue instanceof Map)) { + throw new OptionValidationException(missingSchemaBlockMessage(optionName)); + } + if (!declaresSchemaMetadata((Map<String, Object>) schemaValue)) { + throw new OptionValidationException( + missingMetadataMessage(optionName + ".schema")); + } + } + return true; + } + } + + static final class TableNamesGuard implements ConditionExtension<List<String>> { + + private final Option<String> tablePatternOption; + + TableNamesGuard(Option<String> tablePatternOption) { + this.tablePatternOption = tablePatternOption; + } + + @Override + public String description() { + return "must not be used together with '" + tablePatternOption.key() + "'"; + } + + @Override + public boolean evaluate(ReadonlyConfig config, List<String> value) { + return !config.getOptional(tablePatternOption).isPresent(); + } + } +} diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-vitess/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/vitess/source/VitessSourceFactory.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-vitess/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/vitess/source/VitessSourceFactory.java index 3528cfb814..8917a8df8e 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-vitess/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/vitess/source/VitessSourceFactory.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-vitess/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/vitess/source/VitessSourceFactory.java @@ -17,9 +17,9 @@ package org.apache.seatunnel.connectors.seatunnel.cdc.vitess.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.options.table.ColumnOptions; import org.apache.seatunnel.api.source.SeaTunnelSource; import org.apache.seatunnel.api.source.SourceSplit; import org.apache.seatunnel.api.table.catalog.CatalogTable; @@ -30,6 +30,7 @@ import org.apache.seatunnel.api.table.factory.Factory; import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext; import org.apache.seatunnel.connectors.cdc.base.option.SourceOptions; import org.apache.seatunnel.connectors.cdc.base.option.StartupMode; +import org.apache.seatunnel.connectors.seatunnel.cdc.vitess.config.VitessSchemaMetadataGuards; import org.apache.seatunnel.connectors.seatunnel.cdc.vitess.config.VitessSourceConfig; import org.apache.seatunnel.connectors.seatunnel.cdc.vitess.config.VitessSourceOptions; import org.apache.seatunnel.connectors.seatunnel.cdc.vitess.source.split.VitessSourceSplit; @@ -59,8 +60,26 @@ public class VitessSourceFactory implements ChangeStreamTableSourceFactory { return OptionRule.builder() .required(VitessSourceOptions.HOSTNAME, VitessSourceOptions.KEYSPACE) .exclusive(ConnectorCommonOptions.SCHEMA, ConnectorCommonOptions.TABLE_CONFIGS) + .optional( + ConnectorCommonOptions.SCHEMA, + Conditions.extension( + ConnectorCommonOptions.SCHEMA, + VitessSchemaMetadataGuards.rootSchemaGuard( + ConnectorCommonOptions.SCHEMA))) + .optional( + ConnectorCommonOptions.TABLE_CONFIGS, + Conditions.notEmpty(ConnectorCommonOptions.TABLE_CONFIGS), + Conditions.extension( + ConnectorCommonOptions.TABLE_CONFIGS, + VitessSchemaMetadataGuards.tableConfigsGuard( + ConnectorCommonOptions.TABLE_CONFIGS))) .optional( ConnectorCommonOptions.TABLE_NAMES, + Conditions.extension( + ConnectorCommonOptions.TABLE_NAMES, + VitessSchemaMetadataGuards.tableNamesGuard( + ConnectorCommonOptions.TABLE_PATTERN))) + .optional( ConnectorCommonOptions.TABLE_PATTERN, ConnectorCommonOptions.METALAKE_TYPE, VitessSourceOptions.PORT, @@ -125,6 +144,11 @@ public class VitessSourceFactory implements ChangeStreamTableSourceFactory { return filterDeclaredTables(context, discoverTableSchemas(context)); } + /** + * The declarative rule already covers this contract, including an empty {@code tables_configs} + * and a config that sets neither {@code schema} nor {@code tables_configs}. The checks below + * stay as a second line of defence for callers that build the source without running the rule. + */ private void validateSchemaMetadataContract(TableSourceFactoryContext context) { Optional<Map<String, Object>> singleSchema = context.getOptions().getOptional(ConnectorCommonOptions.SCHEMA); @@ -148,14 +172,10 @@ public class VitessSourceFactory implements ChangeStreamTableSourceFactory { "Vitess CDC requires explicit schema metadata through either 'schema' or 'tables_configs'."); } - @SuppressWarnings("unchecked") private void validateRootSchemaDefinition(Map<String, Object> schema, String optionName) { - if (!schema.containsKey(ColumnOptions.COLUMNS.key()) - && !schema.containsKey(ColumnOptions.METADATA_TABLE_ID.key())) { + if (!VitessSchemaMetadataGuards.declaresSchemaMetadata(schema)) { throw new IllegalArgumentException( - String.format( - "Vitess CDC requires explicit columns or metadata_table_id in '%s' so table schemas stay deterministic.", - optionName)); + VitessSchemaMetadataGuards.missingMetadataMessage(optionName)); } } @@ -164,13 +184,12 @@ public class VitessSourceFactory implements ChangeStreamTableSourceFactory { Object schemaValue = config.get(ConnectorCommonOptions.SCHEMA.key()); if (!(schemaValue instanceof Map)) { throw new IllegalArgumentException( - String.format( - "Vitess CDC requires explicit schema metadata. '%s' must contain a schema block.", - optionName)); + VitessSchemaMetadataGuards.missingSchemaBlockMessage(optionName)); } validateRootSchemaDefinition((Map<String, Object>) schemaValue, optionName + ".schema"); } + /** The rule rejects both selectors together, this keeps the runtime path equally explicit. */ private void validateSelectionOptions(TableSourceFactoryContext context) { if (context.getOptions().getOptional(ConnectorCommonOptions.TABLE_NAMES).isPresent() && context.getOptions() diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-vitess/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/vitess/source/VitessSourceFactoryTest.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-vitess/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/vitess/source/VitessSourceFactoryTest.java index b1603127da..6039cfeeaa 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-vitess/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/vitess/source/VitessSourceFactoryTest.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-vitess/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/vitess/source/VitessSourceFactoryTest.java @@ -18,6 +18,11 @@ package org.apache.seatunnel.connectors.seatunnel.cdc.vitess.source; import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.configuration.util.Condition; +import org.apache.seatunnel.api.configuration.util.ConfigValidator; +import org.apache.seatunnel.api.configuration.util.OptionRule; +import org.apache.seatunnel.api.configuration.util.OptionValidationException; +import org.apache.seatunnel.api.options.ConnectorCommonOptions; import org.apache.seatunnel.api.table.catalog.CatalogTable; import org.apache.seatunnel.api.table.catalog.PhysicalColumn; import org.apache.seatunnel.api.table.catalog.PrimaryKey; @@ -43,6 +48,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.stream.Collectors; /** Covers factory-path validation, schema selection, and restore semantics for Vitess CDC. */ class VitessSourceFactoryTest { @@ -248,6 +254,247 @@ class VitessSourceFactoryTest { exception.getMessage().contains("does not belong to keyspace 'test'")); } + /** + * The declarative rule must carry the schema metadata and selector guards the source relies on, + * so that dropping one of them fails this test instead of silently losing the early rejection. + */ + @Test + void testOptionRuleCarriesSchemaMetadataGuards() { + OptionRule optionRule = factory.optionRule(); + + Assertions.assertNotNull(optionRule); + List<String> constraints = + optionRule.getValueConstraints().stream() + .map(Condition::toString) + .collect(Collectors.toList()); + + Assertions.assertTrue( + hasConstraint(constraints, "'schema' must declare"), + "missing root schema guard: " + constraints); + Assertions.assertTrue( + hasConstraint(constraints, "'tables_configs' is not empty"), + "missing not-empty constraint: " + constraints); + Assertions.assertTrue( + hasConstraint(constraints, "'tables_configs' must contain a 'schema' block"), + "missing tables_configs guard: " + constraints); + Assertions.assertTrue( + hasConstraint( + constraints, + "'table-names' must not be used together with 'table-pattern'"), + "missing table-names guard: " + constraints); + Assertions.assertTrue( + optionRule.getOptionalOptions().contains(ConnectorCommonOptions.TABLE_NAMES)); + } + + /** A root schema that lists its columns stays valid. */ + @Test + void testRuleAcceptsRootSchemaWithColumns() { + assertRuleAccepts(ruleOptionsWithRootSchema(rootSchemaWithColumns())); + } + + /** A root schema may point at the metadata table instead of listing its columns. */ + @Test + void testRuleAcceptsRootSchemaWithMetadataTableId() { + assertRuleAccepts(ruleOptionsWithRootSchema(rootSchemaWithMetadataTableId())); + } + + /** A root schema without columns or metadata_table_id must fail at submission time. */ + @Test + void testRuleRejectsRootSchemaWithoutSchemaMetadata() { + Map<String, Object> schema = new HashMap<>(); + schema.put("table", "test.products"); + + OptionValidationException exception = assertRuleRejects(ruleOptionsWithRootSchema(schema)); + + Assertions.assertTrue(exception.getMessage().contains("columns")); + Assertions.assertTrue(exception.getMessage().contains("metadata_table_id")); + } + + /** tables_configs entries with a complete schema block stay valid. */ + @Test + void testRuleAcceptsTableConfigsWithSchemaBlocks() { + Map<String, Object> options = baseOptions(StartupMode.LATEST, null); + options.put("tables_configs", multiTableConfigs("test.products", "test.customers")); + + assertRuleAccepts(options); + } + + /** An empty tables_configs must fail before the source resolves any table. */ + @Test + void testRuleRejectsEmptyTableConfigs() { + Map<String, Object> options = baseOptions(StartupMode.LATEST, null); + options.put("tables_configs", Collections.emptyList()); + + OptionValidationException exception = assertRuleRejects(options); + + Assertions.assertTrue(exception.getMessage().contains("tables_configs")); + Assertions.assertTrue(exception.getMessage().contains("is not empty")); + } + + /** Every tables_configs entry must carry a schema block. */ + @Test + void testRuleRejectsTableConfigsEntryWithoutSchemaBlock() { + Map<String, Object> options = baseOptions(StartupMode.LATEST, null); + options.put("tables_configs", Collections.singletonList(tableConfigWithoutSchema())); + + OptionValidationException exception = assertRuleRejects(options); + + Assertions.assertTrue(exception.getMessage().contains("tables_configs[0]")); + Assertions.assertTrue(exception.getMessage().contains("must contain a schema block")); + } + + /** A nested schema block must declare columns or metadata_table_id as well. */ + @Test + void testRuleRejectsTableConfigsEntryWithoutSchemaMetadata() { + Map<String, Object> schema = new HashMap<>(); + schema.put("table", "test.products"); + Map<String, Object> tableConfig = new HashMap<>(); + tableConfig.put("schema", schema); + + Map<String, Object> options = baseOptions(StartupMode.LATEST, null); + options.put("tables_configs", Collections.singletonList(tableConfig)); + + OptionValidationException exception = assertRuleRejects(options); + + Assertions.assertTrue(exception.getMessage().contains("tables_configs[0].schema")); + Assertions.assertTrue(exception.getMessage().contains("metadata_table_id")); + } + + /** A tables_configs entry may point at the metadata table instead of listing its columns. */ + @Test + void testRuleAcceptsTableConfigsEntryWithMetadataTableId() { + Map<String, Object> schema = new HashMap<>(); + schema.put("table", "test.products"); + schema.put("metadata_table_id", "test.products"); + Map<String, Object> tableConfig = new HashMap<>(); + tableConfig.put("schema", schema); + + Map<String, Object> options = baseOptions(StartupMode.LATEST, null); + options.put("tables_configs", Collections.singletonList(tableConfig)); + + assertRuleAccepts(options); + } + + /** The guard must name the failing entry, not just the first one. */ + @Test + void testRuleRejectsLaterTableConfigsEntryWithoutSchemaMetadata() { + List<Map<String, Object>> tableConfigs = + new ArrayList<>(singleTableConfigs("test.products")); + Map<String, Object> incompleteSchema = new HashMap<>(); + incompleteSchema.put("table", "test.customers"); + Map<String, Object> incompleteTableConfig = new HashMap<>(); + incompleteTableConfig.put("schema", incompleteSchema); + tableConfigs.add(incompleteTableConfig); + + Map<String, Object> options = baseOptions(StartupMode.LATEST, null); + options.put("tables_configs", tableConfigs); + + OptionValidationException exception = assertRuleRejects(options); + + Assertions.assertTrue(exception.getMessage().contains("tables_configs[1].schema")); + Assertions.assertTrue(exception.getMessage().contains("metadata_table_id")); + } + + /** table-names and table-pattern are alternatives, never a pair. */ + @Test + void testRuleRejectsTableNamesTogetherWithTablePattern() { + Map<String, Object> options = ruleOptionsWithRootSchema(rootSchemaWithColumns()); + options.put("table-names", Collections.singletonList("test.products")); + options.put("table-pattern", "test\\..*"); + + OptionValidationException exception = assertRuleRejects(options); + + Assertions.assertTrue( + exception.getMessage().contains("must not be used together with 'table-pattern'")); + } + + /** Either table selector on its own keeps working. */ + @Test + void testRuleAcceptsSingleTableSelector() { + Map<String, Object> tableNamesOnly = ruleOptionsWithRootSchema(rootSchemaWithColumns()); + tableNamesOnly.put("table-names", Collections.singletonList("test.products")); + assertRuleAccepts(tableNamesOnly); + + Map<String, Object> tablePatternOnly = ruleOptionsWithRootSchema(rootSchemaWithColumns()); + tablePatternOnly.put("table-pattern", "test\\..*"); + assertRuleAccepts(tablePatternOnly); + } + + /** The rule and the source path must report the same contract wording. */ + @Test + void testRuleAndFactoryShareSchemaMetadataMessages() { + Map<String, Object> schema = new HashMap<>(); + schema.put("table", "test.products"); + Map<String, Object> rootSchemaOptions = ruleOptionsWithRootSchema(schema); + + Map<String, Object> tableConfig = new HashMap<>(); + tableConfig.put("schema", schema); + Map<String, Object> tableConfigsOptions = baseOptions(StartupMode.LATEST, null); + tableConfigsOptions.put("tables_configs", Collections.singletonList(tableConfig)); + + Assertions.assertTrue( + assertRuleRejects(rootSchemaOptions) + .getMessage() + .contains(runtimeFailureMessage(rootSchemaOptions))); + Assertions.assertTrue( + assertRuleRejects(tableConfigsOptions) + .getMessage() + .contains(runtimeFailureMessage(tableConfigsOptions))); + } + + private boolean hasConstraint(List<String> constraints, String expected) { + return constraints.stream().anyMatch(constraint -> constraint.contains(expected)); + } + + private void assertRuleAccepts(Map<String, Object> options) { + ConfigValidator.of(ReadonlyConfig.fromMap(options)).validate(factory.optionRule()); + } + + private OptionValidationException assertRuleRejects(Map<String, Object> options) { + return Assertions.assertThrows( + OptionValidationException.class, + () -> + ConfigValidator.of(ReadonlyConfig.fromMap(options)) + .validate(factory.optionRule())); + } + + private Map<String, Object> ruleOptionsWithRootSchema(Map<String, Object> schema) { + Map<String, Object> options = baseOptions(StartupMode.LATEST, null); + options.put("schema", schema); + return options; + } + + private Map<String, Object> rootSchemaWithColumns() { + Map<String, Object> schema = new HashMap<>(); + schema.put("table", "test.products"); + schema.put("columns", Collections.singletonList(column("id", "int"))); + return schema; + } + + private Map<String, Object> rootSchemaWithMetadataTableId() { + Map<String, Object> schema = new HashMap<>(); + schema.put("table", "test.products"); + schema.put("metadata_table_id", "test.products"); + return schema; + } + + private Map<String, Object> tableConfigWithoutSchema() { + Map<String, Object> tableConfig = new HashMap<>(); + tableConfig.put("table", "test.products"); + return tableConfig; + } + + private String runtimeFailureMessage(Map<String, Object> options) { + TableSourceFactoryContext context = + new TableSourceFactoryContext( + ReadonlyConfig.fromMap(options), getClass().getClassLoader()); + IllegalArgumentException exception = + Assertions.assertThrows( + IllegalArgumentException.class, + () -> factory.createSource(context).createSource()); + return exception.getMessage(); + } + private VitessSource createVitessSource(TableSource<?, VitessSourceSplit, ?> tableSource) { return (VitessSource) tableSource.createSource(); }
