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 d7924510b6 [Feature] [Core] Implement config validation for Flink and
Spark (#11434)
d7924510b6 is described below
commit d7924510b68aaac542ec872cce8f576d75a7c19e
Author: lifeng <[email protected]>
AuthorDate: Mon Aug 10 15:49:02 2026 +0800
[Feature] [Core] Implement config validation for Flink and Spark (#11434)
---
seatunnel-core/seatunnel-core-starter/pom.xml | 22 +
.../core/starter/utils/ConfigValidationUtils.java | 448 +++++++++++++++++++++
.../starter/utils/ConfigValidationUtilsTest.java | 214 ++++++++++
.../config/validation_fake_to_inmemory.json | 33 ++
.../config/validation_missing_input_table.json | 26 ++
.../config/validation_multiple_input_tables.json | 35 ++
.../config/validation_sink_assert_failure.json | 28 ++
.../seatunnel-flink-starter-common/pom.xml | 20 +
.../flink/command/FlinkConfValidateCommand.java | 25 +-
.../command/FlinkConfValidateCommandTest.java | 75 ++++
.../test/resources/config/fake_to_inmemory.json | 26 ++
.../resources/config/invalid_restart_strategy.json | 27 ++
.../seatunnel-spark-starter-common/pom.xml | 20 +
.../spark/command/SparkConfValidateCommand.java | 20 +-
.../command/SparkConfValidateCommandTest.java | 71 ++++
.../test/resources/config/fake_to_inmemory.json | 26 ++
.../test/resources/config/missing_input_table.json | 26 ++
17 files changed, 1140 insertions(+), 2 deletions(-)
diff --git a/seatunnel-core/seatunnel-core-starter/pom.xml
b/seatunnel-core/seatunnel-core-starter/pom.xml
index 332a806bc8..73708ae5c4 100644
--- a/seatunnel-core/seatunnel-core-starter/pom.xml
+++ b/seatunnel-core/seatunnel-core-starter/pom.xml
@@ -67,5 +67,27 @@
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-fake</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-e2e-common</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-transforms-v2</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
</dependencies>
</project>
diff --git
a/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigValidationUtils.java
b/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigValidationUtils.java
new file mode 100644
index 0000000000..bc1f2f8b16
--- /dev/null
+++
b/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigValidationUtils.java
@@ -0,0 +1,448 @@
+/*
+ * 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.utils;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+import org.apache.seatunnel.api.common.JobContext;
+import org.apache.seatunnel.api.common.PluginIdentifier;
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.configuration.util.RequiredOption;
+import org.apache.seatunnel.api.options.EnvCommonOptions;
+import org.apache.seatunnel.api.options.EnvOptionRule;
+import org.apache.seatunnel.api.sink.SeaTunnelSink;
+import org.apache.seatunnel.api.sink.SupportMultiTableSink;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
+import org.apache.seatunnel.api.source.SourceSplit;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.api.table.factory.Factory;
+import org.apache.seatunnel.api.table.factory.FactoryUtil;
+import org.apache.seatunnel.api.table.factory.TableSinkFactory;
+import org.apache.seatunnel.api.table.factory.TableSourceFactory;
+import org.apache.seatunnel.api.table.factory.TableTransformFactory;
+import org.apache.seatunnel.api.table.factory.TableTransformFactoryContext;
+import org.apache.seatunnel.api.transform.SeaTunnelTransform;
+import org.apache.seatunnel.common.Constants;
+import org.apache.seatunnel.common.config.CheckResult;
+import org.apache.seatunnel.common.config.Common;
+import org.apache.seatunnel.common.config.TypesafeConfigUtils;
+import org.apache.seatunnel.common.constants.EngineType;
+import org.apache.seatunnel.common.constants.PluginType;
+import org.apache.seatunnel.common.utils.ExceptionUtils;
+import org.apache.seatunnel.common.utils.ReflectionUtils;
+import org.apache.seatunnel.core.starter.exception.ConfigCheckException;
+import org.apache.seatunnel.core.starter.execution.RuntimeEnvironment;
+import
org.apache.seatunnel.plugin.discovery.seatunnel.SeaTunnelFactoryDiscovery;
+import
org.apache.seatunnel.plugin.discovery.seatunnel.SeaTunnelSinkPluginDiscovery;
+import
org.apache.seatunnel.plugin.discovery.seatunnel.SeaTunnelSourcePluginDiscovery;
+
+import scala.Tuple2;
+
+import java.io.Serializable;
+import java.lang.reflect.Method;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static
org.apache.seatunnel.api.options.ConnectorCommonOptions.PLUGIN_INPUT;
+import static
org.apache.seatunnel.api.options.ConnectorCommonOptions.PLUGIN_NAME;
+import static
org.apache.seatunnel.api.options.ConnectorCommonOptions.PLUGIN_OUTPUT;
+import static
org.apache.seatunnel.api.table.factory.FactoryUtil.ensureJobModeMatch;
+
+/** Utility methods for validating SeaTunnel job configuration without
executing a job. */
+@SuppressWarnings({"rawtypes", "unchecked"})
+public final class ConfigValidationUtils {
+
+ private static final BiConsumer<ClassLoader, List<URL>>
ADD_URL_TO_CLASSLOADER =
+ (classLoader, urls) -> {
+ if (classLoader instanceof URLClassLoader) {
+ urls.forEach(url -> ReflectionUtils.invoke(classLoader,
"addURL", url));
+ } else {
+ try {
+ Optional<Method> method =
+ ReflectionUtils.getDeclaredMethod(
+ URLClassLoader.class, "addURL",
URL.class);
+ if (!method.isPresent()) {
+ throw new IllegalStateException(
+ "Unable to find addURL method from
URLClassLoader");
+ }
+ method.get().setAccessible(true);
+ for (URL url : urls) {
+ method.get().invoke(classLoader, url);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Unsupported classloader: " +
classLoader.getClass().getName(), e);
+ }
+ }
+ };
+
+ private ConfigValidationUtils() {}
+
+ public static void validate(Config config) {
+ validate(config, CheckResult.success());
+ }
+
+ public static void validate(Config config, CheckResult checkResult) {
+ if (!checkResult.isSuccess()) {
+ throw new ConfigCheckException(checkResult.getMsg());
+ }
+
+ ClassLoader parentClassLoader =
Thread.currentThread().getContextClassLoader();
+ ClassLoader validationClassLoader = prepareClassLoader(config,
parentClassLoader);
+ Thread.currentThread().setContextClassLoader(validationClassLoader);
+ try {
+ validateEnvironmentConfig(config);
+ JobContext jobContext = new JobContext();
+ jobContext.setJobMode(RuntimeEnvironment.getJobMode(config));
+
jobContext.setEnableCheckpoint(RuntimeEnvironment.getEnableCheckpoint(config));
+
+ List<TableInfo> sourceTables =
+ validateSources(
+ TypesafeConfigUtils.getConfigList(
+ config, Constants.SOURCE,
Collections.emptyList()),
+ jobContext);
+ if (sourceTables.isEmpty()) {
+ throw new ConfigCheckException("At least one source plugin
must be configured.");
+ }
+
+ List<TableInfo> outputTables =
+ validateTransforms(
+ sourceTables,
+ TypesafeConfigUtils.getConfigList(
+ config, Constants.TRANSFORM,
Collections.emptyList()),
+ jobContext);
+
+ validateSinks(
+ outputTables,
+ TypesafeConfigUtils.getConfigList(
+ config, Constants.SINK, Collections.emptyList()),
+ jobContext);
+ } catch (ConfigCheckException e) {
+ throw e;
+ } catch (Exception e) {
+ Throwable rootException = ExceptionUtils.getRootException(e);
+ String message = rootException.getMessage();
+ throw new ConfigCheckException(
+ message == null || message.isEmpty() ? e.getMessage() :
message, e);
+ } finally {
+ Thread.currentThread().setContextClassLoader(parentClassLoader);
+ }
+ }
+
+ private static ClassLoader prepareClassLoader(Config config, ClassLoader
parentClassLoader) {
+ List<URL> additionalJars = new ArrayList<>();
+ if (config.hasPath(Constants.ENV)) {
+ Config envConfig = config.getConfig(Constants.ENV);
+ if (envConfig.hasPath(EnvCommonOptions.JARS.key())) {
+ additionalJars.addAll(
+
Common.getThirdPartyJars(envConfig.getString(EnvCommonOptions.JARS.key()))
+ .stream()
+ .map(Path::toUri)
+ .map(ConfigValidationUtils::toUrl)
+ .collect(Collectors.toList()));
+ }
+ }
+ additionalJars.addAll(
+
Common.getPluginsJarDependenciesWithoutConnectorDependency().stream()
+ .map(Path::toUri)
+ .map(ConfigValidationUtils::toUrl)
+ .collect(Collectors.toList()));
+ additionalJars.addAll(
+ Common.getLibJars().stream()
+ .map(Path::toUri)
+ .map(ConfigValidationUtils::toUrl)
+ .collect(Collectors.toList()));
+ additionalJars =
additionalJars.stream().distinct().collect(Collectors.toList());
+ if (additionalJars.isEmpty()) {
+ return parentClassLoader;
+ }
+
+ try {
+ ADD_URL_TO_CLASSLOADER.accept(parentClassLoader, additionalJars);
+ return parentClassLoader;
+ } catch (RuntimeException e) {
+ return new URLClassLoader(additionalJars.toArray(new URL[0]),
parentClassLoader);
+ }
+ }
+
+ private static URL toUrl(java.net.URI uri) {
+ try {
+ return uri.toURL();
+ } catch (Exception e) {
+ throw new ConfigCheckException("Invalid plugin dependency path: "
+ uri, e);
+ }
+ }
+
+ private static void validateEnvironmentConfig(Config config) {
+ if (!config.hasPath(Constants.ENV)) {
+ return;
+ }
+ ReadonlyConfig envConfig =
ReadonlyConfig.fromConfig(config.getConfig(Constants.ENV));
+ OptionRule optionRule = new EnvOptionRule().optionRule();
+ validateOptionTypes(envConfig, optionRule);
+
org.apache.seatunnel.api.configuration.util.ConfigValidator.of(envConfig)
+ .validate(optionRule);
+ }
+
+ private static void validateOptionTypes(ReadonlyConfig config, OptionRule
rule) {
+ for (Option<?> option : rule.getOptionalOptions()) {
+ config.getOptional(option);
+ }
+ for (RequiredOption requiredOption : rule.getRequiredOptions()) {
+ for (Option<?> option : requiredOption.getOptions()) {
+ config.getOptional(option);
+ }
+ }
+ }
+
+ private static List<TableInfo> validateSources(
+ List<? extends Config> sourceConfigs, JobContext jobContext) {
+ ClassLoader classLoader =
Thread.currentThread().getContextClassLoader();
+ SeaTunnelFactoryDiscovery factoryDiscovery =
+ new SeaTunnelFactoryDiscovery(
+
org.apache.seatunnel.api.table.factory.TableSourceFactory.class,
+ ADD_URL_TO_CLASSLOADER);
+ SeaTunnelSourcePluginDiscovery sourcePluginDiscovery =
+ new SeaTunnelSourcePluginDiscovery(ADD_URL_TO_CLASSLOADER);
+ Function<PluginIdentifier, SeaTunnelSource> fallbackCreateSource =
+ sourcePluginDiscovery::createPluginInstance;
+
+ List<TableInfo> sourceTables = new ArrayList<>();
+ for (Config sourceConfig : sourceConfigs) {
+ PluginIdentifier pluginIdentifier =
+ getPluginIdentifier(sourceConfig, PluginType.SOURCE);
+ TableSourceFactory sourceFactory =
+ (TableSourceFactory)
+ factoryDiscovery
+
.createOptionalPluginInstance(pluginIdentifier)
+ .orElse(null);
+ if (sourceFactory != null) {
+
org.apache.seatunnel.api.configuration.util.ConfigValidator.validateUnknownKeys(
+ ReadonlyConfig.fromConfig(sourceConfig),
+ sourceFactory.optionRule(),
+ pluginIdentifier.getPluginName());
+ }
+ ClassLoader sourceClassLoader =
getFactoryClassLoader(sourceFactory, classLoader);
+ Tuple2<SeaTunnelSource<Object, SourceSplit, Serializable>,
List<CatalogTable>> source =
+ FactoryUtil.createAndPrepareSource(
+ ReadonlyConfig.fromConfig(sourceConfig),
+ sourceClassLoader,
+ pluginIdentifier.getPluginName(),
+ fallbackCreateSource,
+ sourceFactory,
+ null);
+
+ source._1().setJobContext(jobContext);
+ ensureJobModeMatch(jobContext, source._1());
+ sourceTables.add(
+ new TableInfo(
+ source._2(),
+
ReadonlyConfig.fromConfig(sourceConfig).get(PLUGIN_OUTPUT)));
+ }
+ return sourceTables;
+ }
+
+ private static List<TableInfo> validateTransforms(
+ List<TableInfo> upstreamTables,
+ List<? extends Config> transformConfigs,
+ JobContext jobContext) {
+ if (transformConfigs.isEmpty()) {
+ return upstreamTables;
+ }
+
+ ClassLoader classLoader =
Thread.currentThread().getContextClassLoader();
+ SeaTunnelFactoryDiscovery factoryDiscovery =
+ new SeaTunnelFactoryDiscovery(TableTransformFactory.class,
ADD_URL_TO_CLASSLOADER);
+ TableInfo defaultInput = upstreamTables.get(0);
+ Map<String, TableInfo> outputTables =
+ upstreamTables.stream()
+ .collect(
+ Collectors.toMap(
+ TableInfo::getTableName,
+ Function.identity(),
+ (left, right) -> right,
+ LinkedHashMap::new));
+
+ for (Config transformConfig : transformConfigs) {
+ PluginIdentifier pluginIdentifier =
+ getPluginIdentifier(transformConfig, PluginType.TRANSFORM);
+
+ TableInfo inputTable =
+ resolveInputTable(
+ transformConfig,
+ new ArrayList<>(outputTables.values()),
+ "Multiple input tables are not supported
in the current version")
+ .orElse(defaultInput);
+
+ TableTransformFactory factory =
+ (TableTransformFactory)
factoryDiscovery.createPluginInstance(pluginIdentifier);
+
org.apache.seatunnel.api.configuration.util.ConfigValidator.validateUnknownKeys(
+ ReadonlyConfig.fromConfig(transformConfig),
+ factory.optionRule(),
+ pluginIdentifier.getPluginName());
+ ClassLoader transformClassLoader = getFactoryClassLoader(factory,
classLoader);
+ TableTransformFactoryContext context =
+ new TableTransformFactoryContext(
+ inputTable.getCatalogTables(),
+ ReadonlyConfig.fromConfig(transformConfig),
+ transformClassLoader);
+
org.apache.seatunnel.api.configuration.util.ConfigValidator.of(context.getOptions())
+ .validate(factory.optionRule());
+ SeaTunnelTransform<?> transform =
factory.createTransform(context).createTransform();
+ transform.setJobContext(jobContext);
+
+ String pluginOutputIdentifier =
+
ReadonlyConfig.fromConfig(transformConfig).get(PLUGIN_OUTPUT);
+ outputTables.put(
+ pluginOutputIdentifier,
+ new TableInfo(transform.getProducedCatalogTables(),
pluginOutputIdentifier));
+ }
+ return new ArrayList<>(outputTables.values());
+ }
+
+ private static void validateSinks(
+ List<TableInfo> upstreamTables,
+ List<? extends Config> sinkConfigs,
+ JobContext jobContext) {
+ if (sinkConfigs.isEmpty()) {
+ throw new ConfigCheckException("At least one sink plugin must be
configured.");
+ }
+
+ ClassLoader classLoader =
Thread.currentThread().getContextClassLoader();
+ SeaTunnelFactoryDiscovery factoryDiscovery =
+ new SeaTunnelFactoryDiscovery(TableSinkFactory.class,
ADD_URL_TO_CLASSLOADER);
+ SeaTunnelSinkPluginDiscovery sinkPluginDiscovery =
+ new SeaTunnelSinkPluginDiscovery(ADD_URL_TO_CLASSLOADER);
+ Function<PluginIdentifier, SeaTunnelSink> fallbackCreateSink =
+ sinkPluginDiscovery::createPluginInstance;
+
+ TableInfo defaultInput = upstreamTables.get(upstreamTables.size() - 1);
+ for (Config sinkConfig : sinkConfigs) {
+ PluginIdentifier pluginIdentifier =
getPluginIdentifier(sinkConfig, PluginType.SINK);
+ TableSinkFactory sinkFactory =
+ (TableSinkFactory)
+ factoryDiscovery
+
.createOptionalPluginInstance(pluginIdentifier)
+ .orElse(null);
+ if (sinkFactory != null) {
+
org.apache.seatunnel.api.configuration.util.ConfigValidator.validateUnknownKeys(
+ ReadonlyConfig.fromConfig(sinkConfig),
+ sinkFactory.optionRule(),
+ pluginIdentifier.getPluginName());
+ }
+ ClassLoader sinkClassLoader = getFactoryClassLoader(sinkFactory,
classLoader);
+ TableInfo inputTable =
+ resolveInputTable(
+ sinkConfig,
+ upstreamTables,
+ "Multiple input tables are not supported
in the current version")
+ .orElse(defaultInput);
+
+ Map<TablePath, SeaTunnelSink> sinks = new LinkedHashMap<>();
+ for (CatalogTable catalogTable : inputTable.getCatalogTables()) {
+ SeaTunnelSink sink =
+ FactoryUtil.createAndPrepareSink(
+ catalogTable,
+ ReadonlyConfig.fromConfig(sinkConfig),
+ sinkClassLoader,
+ pluginIdentifier.getPluginName(),
+ fallbackCreateSink,
+ sinkFactory);
+ sink.setJobContext(jobContext);
+ sinks.put(catalogTable.getTableId().toTablePath(), sink);
+ }
+
+ if (!sinks.isEmpty()
+ &&
sinks.values().stream().allMatch(SupportMultiTableSink.class::isInstance)) {
+ FactoryUtil.createMultiTableSink(
+ sinks, ReadonlyConfig.fromConfig(sinkConfig),
sinkClassLoader);
+ }
+ }
+ }
+
+ static ClassLoader getFactoryClassLoader(Factory factory, ClassLoader
fallbackClassLoader) {
+ return factory == null ? fallbackClassLoader :
factory.getClass().getClassLoader();
+ }
+
+ private static Optional<TableInfo> resolveInputTable(
+ Config pluginConfig, List<TableInfo> upstreamTables, String
multiInputMessage) {
+ ReadonlyConfig readonlyConfig =
ReadonlyConfig.fromConfig(pluginConfig);
+ if (!readonlyConfig.getOptional(PLUGIN_INPUT).isPresent()) {
+ return Optional.empty();
+ }
+
+ List<String> pluginInputIdentifiers = readonlyConfig.get(PLUGIN_INPUT);
+ if (pluginInputIdentifiers.isEmpty()) {
+ throw new ConfigCheckException("plugin_input must not be empty
when configured");
+ }
+ if (pluginInputIdentifiers.size() > 1) {
+ throw new ConfigCheckException(multiInputMessage);
+ }
+
+ String pluginInputIdentifier = pluginInputIdentifiers.get(0);
+ return Optional.of(
+ upstreamTables.stream()
+ .filter(info ->
pluginInputIdentifier.equals(info.getTableName()))
+ .findFirst()
+ .orElseThrow(
+ () ->
+ new ConfigCheckException(
+ String.format(
+ "table %s not found",
+
pluginInputIdentifier))));
+ }
+
+ private static PluginIdentifier getPluginIdentifier(
+ Config pluginConfig, PluginType pluginType) {
+ return PluginIdentifier.of(
+ EngineType.SEATUNNEL.getEngine(),
+ pluginType.getType(),
+ pluginConfig.getString(PLUGIN_NAME.key()));
+ }
+
+ private static final class TableInfo {
+ private final List<CatalogTable> catalogTables;
+ private final String tableName;
+
+ private TableInfo(List<CatalogTable> catalogTables, String tableName) {
+ this.catalogTables = catalogTables;
+ this.tableName = tableName;
+ }
+
+ public List<CatalogTable> getCatalogTables() {
+ return catalogTables;
+ }
+
+ public String getTableName() {
+ return tableName;
+ }
+ }
+}
diff --git
a/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigValidationUtilsTest.java
b/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigValidationUtilsTest.java
new file mode 100644
index 0000000000..1ef1b340fc
--- /dev/null
+++
b/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigValidationUtilsTest.java
@@ -0,0 +1,214 @@
+/*
+ * 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.utils;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+import org.apache.seatunnel.shade.com.typesafe.config.ConfigValueFactory;
+
+import org.apache.seatunnel.api.table.factory.Factory;
+import org.apache.seatunnel.common.Constants;
+import org.apache.seatunnel.core.starter.exception.ConfigCheckException;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Proxy;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.List;
+import java.util.stream.Collectors;
+
+class ConfigValidationUtilsTest {
+
+ @Test
+ void shouldUseFactoryClassLoader() throws Exception {
+ ClassLoader parentClassLoader =
ConfigValidationUtilsTest.class.getClassLoader();
+ try (URLClassLoader factoryClassLoader =
+ new URLClassLoader(new URL[0], parentClassLoader)) {
+ Factory factory =
+ (Factory)
+ Proxy.newProxyInstance(
+ factoryClassLoader,
+ new Class<?>[] {Factory.class},
+ (proxy, method, args) -> null);
+
+ Assertions.assertSame(
+ factoryClassLoader,
+ ConfigValidationUtils.getFactoryClassLoader(factory,
parentClassLoader));
+ }
+ }
+
+ @Test
+ void shouldUseFallbackClassLoaderWhenFactoryIsMissing() {
+ ClassLoader fallbackClassLoader =
ConfigValidationUtilsTest.class.getClassLoader();
+
+ Assertions.assertSame(
+ fallbackClassLoader,
+ ConfigValidationUtils.getFactoryClassLoader(null,
fallbackClassLoader));
+ }
+
+ @Test
+ void shouldRestoreContextClassLoaderAfterValidation() throws Exception {
+ Config config =
+
ConfigBuilder.of(getResourcePath("/config/validation_fake_to_inmemory.json"));
+ ClassLoader originalClassLoader =
Thread.currentThread().getContextClassLoader();
+ try (URLClassLoader validationClassLoader =
+ new URLClassLoader(new URL[0], originalClassLoader)) {
+
Thread.currentThread().setContextClassLoader(validationClassLoader);
+
+ Assertions.assertDoesNotThrow(() ->
ConfigValidationUtils.validate(config));
+ Assertions.assertSame(
+ validationClassLoader,
Thread.currentThread().getContextClassLoader());
+ } finally {
+ Thread.currentThread().setContextClassLoader(originalClassLoader);
+ }
+ }
+
+ @Test
+ void shouldFailWhenConnectorOptionIsUnknown() throws URISyntaxException {
+ Config config =
+ withPluginOption(
+ ConfigBuilder.of(
+
getResourcePath("/config/validation_fake_to_inmemory.json")),
+ Constants.SOURCE,
+ 0,
+ "unknown_option",
+ true);
+
+ ConfigCheckException exception =
+ Assertions.assertThrows(
+ ConfigCheckException.class, () ->
ConfigValidationUtils.validate(config));
+
+
Assertions.assertTrue(exception.getMessage().contains("unknown_option"));
+ }
+
+ @Test
+ void shouldFailWhenPluginInputIsEmpty() throws URISyntaxException {
+ Config config =
+ withPluginOption(
+ ConfigBuilder.of(
+
getResourcePath("/config/validation_fake_to_inmemory.json")),
+ Constants.SINK,
+ 0,
+ "plugin_input",
+ java.util.Collections.emptyList());
+
+ ConfigCheckException exception =
+ Assertions.assertThrows(
+ ConfigCheckException.class, () ->
ConfigValidationUtils.validate(config));
+
+ Assertions.assertEquals(
+ "plugin_input must not be empty when configured",
exception.getMessage());
+ }
+
+ @Test
+ void shouldValidateHappyPathConfig() throws URISyntaxException {
+ Config config =
+
ConfigBuilder.of(getResourcePath("/config/validation_fake_to_inmemory.json"));
+ Assertions.assertDoesNotThrow(() ->
ConfigValidationUtils.validate(config));
+ }
+
+ @Test
+ void shouldFailWhenSinkInputTableIsMissing() throws URISyntaxException {
+ Config config =
+
ConfigBuilder.of(getResourcePath("/config/validation_missing_input_table.json"));
+
+ ConfigCheckException exception =
+ Assertions.assertThrows(
+ ConfigCheckException.class, () ->
ConfigValidationUtils.validate(config));
+
+ Assertions.assertEquals("table missing_table not found",
exception.getMessage());
+ }
+
+ @Test
+ void shouldFailWhenSinkHasMultipleInputTables() throws URISyntaxException {
+ Config config =
+
ConfigBuilder.of(getResourcePath("/config/validation_multiple_input_tables.json"));
+
+ ConfigCheckException exception =
+ Assertions.assertThrows(
+ ConfigCheckException.class, () ->
ConfigValidationUtils.validate(config));
+
+ Assertions.assertEquals(
+ "Multiple input tables are not supported in the current
version",
+ exception.getMessage());
+ }
+
+ @Test
+ void shouldFailWhenSinkFactoryRejectsConfig() throws URISyntaxException {
+ Config config =
+
ConfigBuilder.of(getResourcePath("/config/validation_sink_assert_failure.json"));
+
+ ConfigCheckException exception =
+ Assertions.assertThrows(
+ ConfigCheckException.class, () ->
ConfigValidationUtils.validate(config));
+
+ Assertions.assertTrue(exception.getMessage().contains("assert key and
value not match"));
+ }
+
+ @Test
+ void shouldFailWhenSourceSectionIsMissing() throws URISyntaxException {
+ Config config =
+
ConfigBuilder.of(getResourcePath("/config/validation_fake_to_inmemory.json"))
+ .withoutPath(Constants.SOURCE);
+
+ ConfigCheckException exception =
+ Assertions.assertThrows(
+ ConfigCheckException.class, () ->
ConfigValidationUtils.validate(config));
+
+ Assertions.assertEquals(
+ "At least one source plugin must be configured.",
exception.getMessage());
+ }
+
+ @Test
+ void shouldFailWhenSinkSectionIsMissing() throws URISyntaxException {
+ Config config =
+
ConfigBuilder.of(getResourcePath("/config/validation_fake_to_inmemory.json"))
+ .withoutPath(Constants.SINK);
+
+ ConfigCheckException exception =
+ Assertions.assertThrows(
+ ConfigCheckException.class, () ->
ConfigValidationUtils.validate(config));
+
+ Assertions.assertEquals(
+ "At least one sink plugin must be configured.",
exception.getMessage());
+ }
+
+ private static Path getResourcePath(String resourcePath) throws
URISyntaxException {
+ URL resource =
ConfigValidationUtilsTest.class.getResource(resourcePath);
+ Assertions.assertNotNull(resource);
+ return Paths.get(resource.toURI());
+ }
+
+ private static Config withPluginOption(
+ Config config, String section, int index, String key, Object
value) {
+ List<Object> pluginConfigs =
+ config.getConfigList(section).stream()
+ .map(
+ pluginConfig ->
+ pluginConfig
+ .withValue(
+ key,
ConfigValueFactory.fromAnyRef(value))
+ .root())
+ .collect(Collectors.toList());
+ return config.withValue(section,
ConfigValueFactory.fromIterable(pluginConfigs));
+ }
+}
diff --git
a/seatunnel-core/seatunnel-core-starter/src/test/resources/config/validation_fake_to_inmemory.json
b/seatunnel-core/seatunnel-core-starter/src/test/resources/config/validation_fake_to_inmemory.json
new file mode 100644
index 0000000000..415174c623
--- /dev/null
+++
b/seatunnel-core/seatunnel-core-starter/src/test/resources/config/validation_fake_to_inmemory.json
@@ -0,0 +1,33 @@
+{
+ "env": {
+ "parallelism": 1,
+ "job.mode": "BATCH"
+ },
+ "source": [
+ {
+ "plugin_name": "FakeSource",
+ "plugin_output": "fake_table",
+ "row.num": 10,
+ "schema": {
+ "fields": {
+ "name": "string",
+ "age": "int"
+ }
+ }
+ }
+ ],
+ "transform": [
+ {
+ "plugin_name": "Filter",
+ "plugin_input": "fake_table",
+ "plugin_output": "filtered_table",
+ "include_fields": ["name"]
+ }
+ ],
+ "sink": [
+ {
+ "plugin_name": "InMemory",
+ "plugin_input": "filtered_table"
+ }
+ ]
+}
diff --git
a/seatunnel-core/seatunnel-core-starter/src/test/resources/config/validation_missing_input_table.json
b/seatunnel-core/seatunnel-core-starter/src/test/resources/config/validation_missing_input_table.json
new file mode 100644
index 0000000000..c6d1043e02
--- /dev/null
+++
b/seatunnel-core/seatunnel-core-starter/src/test/resources/config/validation_missing_input_table.json
@@ -0,0 +1,26 @@
+{
+ "env": {
+ "parallelism": 1,
+ "job.mode": "BATCH"
+ },
+ "source": [
+ {
+ "plugin_name": "FakeSource",
+ "plugin_output": "fake_table",
+ "row.num": 10,
+ "schema": {
+ "fields": {
+ "name": "string",
+ "age": "int"
+ }
+ }
+ }
+ ],
+ "transform": [],
+ "sink": [
+ {
+ "plugin_name": "InMemory",
+ "plugin_input": "missing_table"
+ }
+ ]
+}
diff --git
a/seatunnel-core/seatunnel-core-starter/src/test/resources/config/validation_multiple_input_tables.json
b/seatunnel-core/seatunnel-core-starter/src/test/resources/config/validation_multiple_input_tables.json
new file mode 100644
index 0000000000..3e3bddc305
--- /dev/null
+++
b/seatunnel-core/seatunnel-core-starter/src/test/resources/config/validation_multiple_input_tables.json
@@ -0,0 +1,35 @@
+{
+ "env": {
+ "parallelism": 1,
+ "job.mode": "BATCH"
+ },
+ "source": [
+ {
+ "plugin_name": "FakeSource",
+ "plugin_output": "first_table",
+ "row.num": 10,
+ "schema": {
+ "fields": {
+ "name": "string"
+ }
+ }
+ },
+ {
+ "plugin_name": "FakeSource",
+ "plugin_output": "second_table",
+ "row.num": 10,
+ "schema": {
+ "fields": {
+ "name": "string"
+ }
+ }
+ }
+ ],
+ "transform": [],
+ "sink": [
+ {
+ "plugin_name": "InMemory",
+ "plugin_input": ["first_table", "second_table"]
+ }
+ ]
+}
diff --git
a/seatunnel-core/seatunnel-core-starter/src/test/resources/config/validation_sink_assert_failure.json
b/seatunnel-core/seatunnel-core-starter/src/test/resources/config/validation_sink_assert_failure.json
new file mode 100644
index 0000000000..7b9843f3e3
--- /dev/null
+++
b/seatunnel-core/seatunnel-core-starter/src/test/resources/config/validation_sink_assert_failure.json
@@ -0,0 +1,28 @@
+{
+ "env": {
+ "parallelism": 1,
+ "job.mode": "BATCH"
+ },
+ "source": [
+ {
+ "plugin_name": "FakeSource",
+ "plugin_output": "fake_table",
+ "row.num": 10,
+ "schema": {
+ "fields": {
+ "name": "string",
+ "age": "int"
+ }
+ }
+ }
+ ],
+ "transform": [],
+ "sink": [
+ {
+ "plugin_name": "InMemory",
+ "plugin_input": "fake_table",
+ "assert_options_key": "left",
+ "assert_options_value": "right"
+ }
+ ]
+}
diff --git
a/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/pom.xml
b/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/pom.xml
index ee8481127f..cd68112bed 100644
---
a/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/pom.xml
+++
b/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/pom.xml
@@ -74,6 +74,26 @@
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-console</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-fake</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.junit.platform</groupId>
+ <artifactId>junit-platform-launcher</artifactId>
+ <scope>test</scope>
+ </dependency>
+
</dependencies>
</project>
diff --git
a/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/main/java/org/apache/seatunnel/core/starter/flink/command/FlinkConfValidateCommand.java
b/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/main/java/org/apache/seatunnel/core/starter/flink/command/FlinkConfValidateCommand.java
index 78921e6980..a6de3fce74 100644
---
a/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/main/java/org/apache/seatunnel/core/starter/flink/command/FlinkConfValidateCommand.java
+++
b/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/main/java/org/apache/seatunnel/core/starter/flink/command/FlinkConfValidateCommand.java
@@ -17,15 +17,26 @@
package org.apache.seatunnel.core.starter.flink.command;
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+import org.apache.seatunnel.shade.com.typesafe.config.ConfigUtil;
+import org.apache.seatunnel.shade.com.typesafe.config.ConfigValueFactory;
+
+import org.apache.seatunnel.api.metalake.MetalakeConfigUtils;
+import org.apache.seatunnel.common.Constants;
import org.apache.seatunnel.core.starter.command.Command;
import org.apache.seatunnel.core.starter.exception.ConfigCheckException;
import org.apache.seatunnel.core.starter.flink.args.FlinkCommandArgs;
+import org.apache.seatunnel.core.starter.flink.utils.EnvironmentUtil;
+import org.apache.seatunnel.core.starter.utils.ConfigBuilder;
+import org.apache.seatunnel.core.starter.utils.ConfigValidationUtils;
import org.apache.seatunnel.core.starter.utils.FileUtils;
import lombok.extern.slf4j.Slf4j;
import java.nio.file.Path;
+import static
org.apache.seatunnel.core.starter.utils.FileUtils.checkConfigExist;
+
/** Use to validate the configuration of the SeaTunnel API. */
@Slf4j
public class FlinkConfValidateCommand implements Command<FlinkCommandArgs> {
@@ -39,6 +50,18 @@ public class FlinkConfValidateCommand implements
Command<FlinkCommandArgs> {
@Override
public void execute() throws ConfigCheckException {
Path configPath = FileUtils.getConfigPath(flinkCommandArgs);
- // TODO: validate the config by new api
+ checkConfigExist(configPath);
+ Config config =
+ MetalakeConfigUtils.getMetalakeConfig(
+ ConfigBuilder.of(configPath,
flinkCommandArgs.getVariables()));
+ if (!flinkCommandArgs.getJobName().equals(Constants.LOGO)) {
+ config =
+ config.withValue(
+ ConfigUtil.joinPath("env", "job.name"),
+
ConfigValueFactory.fromAnyRef(flinkCommandArgs.getJobName()));
+ }
+ ConfigValidationUtils.validate(
+ config,
EnvironmentUtil.checkRestartStrategy(config.getConfig("env")));
+ log.info("Config validation succeeded: {}", configPath);
}
}
diff --git
a/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/test/java/org/apache/seatunnel/core/starter/flink/command/FlinkConfValidateCommandTest.java
b/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/test/java/org/apache/seatunnel/core/starter/flink/command/FlinkConfValidateCommandTest.java
new file mode 100644
index 0000000000..9d596e2b07
--- /dev/null
+++
b/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/test/java/org/apache/seatunnel/core/starter/flink/command/FlinkConfValidateCommandTest.java
@@ -0,0 +1,75 @@
+/*
+ * 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.flink.command;
+
+import org.apache.seatunnel.common.config.DeployMode;
+import org.apache.seatunnel.core.starter.command.Command;
+import org.apache.seatunnel.core.starter.exception.ConfigCheckException;
+import org.apache.seatunnel.core.starter.flink.args.FlinkCommandArgs;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Collections;
+
+class FlinkConfValidateCommandTest {
+
+ @Test
+ void shouldValidateFlinkConfig() throws URISyntaxException {
+ FlinkCommandArgs commandArgs =
buildArgs("/config/fake_to_inmemory.json");
+ Command<?> command = commandArgs.buildCommand();
+
+ Assertions.assertInstanceOf(FlinkConfValidateCommand.class, command);
+ Assertions.assertDoesNotThrow(command::execute);
+ }
+
+ @Test
+ void shouldRejectInvalidRestartStrategy() throws URISyntaxException {
+ FlinkCommandArgs commandArgs =
buildArgs("/config/invalid_restart_strategy.json");
+ Command<?> command = commandArgs.buildCommand();
+
+ Assertions.assertInstanceOf(FlinkConfValidateCommand.class, command);
+ ConfigCheckException exception =
+ Assertions.assertThrows(ConfigCheckException.class,
command::execute);
+
+ Assertions.assertTrue(
+ exception
+ .getMessage()
+ .contains(
+ "fixed-delay restart strategy must set
[execution.restart.attempts],[execution.restart.delayBetweenAttempts]"));
+ }
+
+ private static FlinkCommandArgs buildArgs(String resourcePath) throws
URISyntaxException {
+ FlinkCommandArgs commandArgs = new FlinkCommandArgs();
+ commandArgs.setDeployMode(DeployMode.RUN);
+ commandArgs.setCheckConfig(true);
+ commandArgs.setConfigFile(getResourcePath(resourcePath).toString());
+ commandArgs.setVariables(Collections.emptyList());
+ return commandArgs;
+ }
+
+ private static Path getResourcePath(String resourcePath) throws
URISyntaxException {
+ URL resource =
FlinkConfValidateCommandTest.class.getResource(resourcePath);
+ Assertions.assertNotNull(resource);
+ return Paths.get(resource.toURI());
+ }
+}
diff --git
a/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/test/resources/config/fake_to_inmemory.json
b/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/test/resources/config/fake_to_inmemory.json
new file mode 100644
index 0000000000..66fed43e3c
--- /dev/null
+++
b/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/test/resources/config/fake_to_inmemory.json
@@ -0,0 +1,26 @@
+{
+ "env": {
+ "parallelism": 1,
+ "job.mode": "BATCH"
+ },
+ "source": [
+ {
+ "plugin_name": "FakeSource",
+ "plugin_output": "fake_table",
+ "row.num": 10,
+ "schema": {
+ "fields": {
+ "name": "string",
+ "age": "int"
+ }
+ }
+ }
+ ],
+ "transform": [],
+ "sink": [
+ {
+ "plugin_name": "Console",
+ "plugin_input": "fake_table"
+ }
+ ]
+}
diff --git
a/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/test/resources/config/invalid_restart_strategy.json
b/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/test/resources/config/invalid_restart_strategy.json
new file mode 100644
index 0000000000..872c497e90
--- /dev/null
+++
b/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/test/resources/config/invalid_restart_strategy.json
@@ -0,0 +1,27 @@
+{
+ "env": {
+ "parallelism": 1,
+ "job.mode": "BATCH",
+ "execution.restart.strategy": "fixed-delay"
+ },
+ "source": [
+ {
+ "plugin_name": "FakeSource",
+ "plugin_output": "fake_table",
+ "row.num": 10,
+ "schema": {
+ "fields": {
+ "name": "string",
+ "age": "int"
+ }
+ }
+ }
+ ],
+ "transform": [],
+ "sink": [
+ {
+ "plugin_name": "Console",
+ "plugin_input": "fake_table"
+ }
+ ]
+}
diff --git
a/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/pom.xml
b/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/pom.xml
index ec4df4bfa9..35bc36b12e 100644
---
a/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/pom.xml
+++
b/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/pom.xml
@@ -64,6 +64,26 @@
<scope>${spark.scope}</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-console</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-fake</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.junit.platform</groupId>
+ <artifactId>junit-platform-launcher</artifactId>
+ <scope>test</scope>
+ </dependency>
+
</dependencies>
</project>
diff --git
a/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/main/java/org/apache/seatunnel/core/starter/spark/command/SparkConfValidateCommand.java
b/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/main/java/org/apache/seatunnel/core/starter/spark/command/SparkConfValidateCommand.java
index c198dff3d0..eb948b6a8e 100644
---
a/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/main/java/org/apache/seatunnel/core/starter/spark/command/SparkConfValidateCommand.java
+++
b/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/main/java/org/apache/seatunnel/core/starter/spark/command/SparkConfValidateCommand.java
@@ -17,9 +17,17 @@
package org.apache.seatunnel.core.starter.spark.command;
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+import org.apache.seatunnel.shade.com.typesafe.config.ConfigUtil;
+import org.apache.seatunnel.shade.com.typesafe.config.ConfigValueFactory;
+
+import org.apache.seatunnel.api.metalake.MetalakeConfigUtils;
+import org.apache.seatunnel.common.Constants;
import org.apache.seatunnel.core.starter.command.Command;
import org.apache.seatunnel.core.starter.exception.ConfigCheckException;
import org.apache.seatunnel.core.starter.spark.args.SparkCommandArgs;
+import org.apache.seatunnel.core.starter.utils.ConfigBuilder;
+import org.apache.seatunnel.core.starter.utils.ConfigValidationUtils;
import org.apache.spark.SparkFiles;
@@ -44,6 +52,16 @@ public class SparkConfValidateCommand implements
Command<SparkCommandArgs> {
Path configPath =
SparkTaskExecuteCommand.resolveConfigPath(sparkCommandArgs,
SparkFiles::get);
checkConfigExist(configPath);
- // TODO: validate the config by new api
+ Config config =
+ MetalakeConfigUtils.getMetalakeConfig(
+ ConfigBuilder.of(configPath,
sparkCommandArgs.getVariables()));
+ if (!sparkCommandArgs.getJobName().equals(Constants.LOGO)) {
+ config =
+ config.withValue(
+ ConfigUtil.joinPath("env", "job.name"),
+
ConfigValueFactory.fromAnyRef(sparkCommandArgs.getJobName()));
+ }
+ ConfigValidationUtils.validate(config);
+ log.info("Config validation succeeded: {}", configPath);
}
}
diff --git
a/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/test/java/org/apache/seatunnel/core/starter/spark/command/SparkConfValidateCommandTest.java
b/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/test/java/org/apache/seatunnel/core/starter/spark/command/SparkConfValidateCommandTest.java
new file mode 100644
index 0000000000..290b4568a2
--- /dev/null
+++
b/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/test/java/org/apache/seatunnel/core/starter/spark/command/SparkConfValidateCommandTest.java
@@ -0,0 +1,71 @@
+/*
+ * 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.spark.command;
+
+import org.apache.seatunnel.common.config.DeployMode;
+import org.apache.seatunnel.core.starter.command.Command;
+import org.apache.seatunnel.core.starter.exception.ConfigCheckException;
+import org.apache.seatunnel.core.starter.spark.args.SparkCommandArgs;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Collections;
+
+class SparkConfValidateCommandTest {
+
+ @Test
+ void shouldValidateSparkConfig() throws URISyntaxException {
+ SparkCommandArgs commandArgs =
buildArgs("/config/fake_to_inmemory.json");
+ Command<?> command = commandArgs.buildCommand();
+
+ Assertions.assertInstanceOf(SparkConfValidateCommand.class, command);
+ Assertions.assertDoesNotThrow(command::execute);
+ }
+
+ @Test
+ void shouldRejectMissingSinkInputTable() throws URISyntaxException {
+ SparkCommandArgs commandArgs =
buildArgs("/config/missing_input_table.json");
+ Command<?> command = commandArgs.buildCommand();
+
+ Assertions.assertInstanceOf(SparkConfValidateCommand.class, command);
+ ConfigCheckException exception =
+ Assertions.assertThrows(ConfigCheckException.class,
command::execute);
+
+ Assertions.assertEquals("table missing_table not found",
exception.getMessage());
+ }
+
+ private static SparkCommandArgs buildArgs(String resourcePath) throws
URISyntaxException {
+ SparkCommandArgs commandArgs = new SparkCommandArgs();
+ commandArgs.setDeployMode(DeployMode.CLIENT);
+ commandArgs.setCheckConfig(true);
+ commandArgs.setConfigFile(getResourcePath(resourcePath).toString());
+ commandArgs.setVariables(Collections.emptyList());
+ return commandArgs;
+ }
+
+ private static Path getResourcePath(String resourcePath) throws
URISyntaxException {
+ URL resource =
SparkConfValidateCommandTest.class.getResource(resourcePath);
+ Assertions.assertNotNull(resource);
+ return Paths.get(resource.toURI());
+ }
+}
diff --git
a/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/test/resources/config/fake_to_inmemory.json
b/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/test/resources/config/fake_to_inmemory.json
new file mode 100644
index 0000000000..66fed43e3c
--- /dev/null
+++
b/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/test/resources/config/fake_to_inmemory.json
@@ -0,0 +1,26 @@
+{
+ "env": {
+ "parallelism": 1,
+ "job.mode": "BATCH"
+ },
+ "source": [
+ {
+ "plugin_name": "FakeSource",
+ "plugin_output": "fake_table",
+ "row.num": 10,
+ "schema": {
+ "fields": {
+ "name": "string",
+ "age": "int"
+ }
+ }
+ }
+ ],
+ "transform": [],
+ "sink": [
+ {
+ "plugin_name": "Console",
+ "plugin_input": "fake_table"
+ }
+ ]
+}
diff --git
a/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/test/resources/config/missing_input_table.json
b/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/test/resources/config/missing_input_table.json
new file mode 100644
index 0000000000..f21f963ddc
--- /dev/null
+++
b/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/test/resources/config/missing_input_table.json
@@ -0,0 +1,26 @@
+{
+ "env": {
+ "parallelism": 1,
+ "job.mode": "BATCH"
+ },
+ "source": [
+ {
+ "plugin_name": "FakeSource",
+ "plugin_output": "fake_table",
+ "row.num": 10,
+ "schema": {
+ "fields": {
+ "name": "string",
+ "age": "int"
+ }
+ }
+ }
+ ],
+ "transform": [],
+ "sink": [
+ {
+ "plugin_name": "Console",
+ "plugin_input": "missing_table"
+ }
+ ]
+}
