This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 c37823b9a3 [Improve][Connector-V2][Vitess] Migrate CDC source option
validation to declarative OptionRule (#12746)
c37823b9a3 is described below
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();
}
