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 f01ee9f6b7 [Fix][Core] Fix Spark cluster config lookup (#11366)
f01ee9f6b7 is described below
commit f01ee9f6b7eb324be198dfc741a43349f09d0627
Author: Jast <[email protected]>
AuthorDate: Tue Jul 21 22:51:52 2026 +0800
[Fix][Core] Fix Spark cluster config lookup (#11366)
---
.../spark/command/SparkConfValidateCommand.java | 9 +-
.../spark/command/SparkTaskExecuteCommand.java | 51 +++++++-
.../spark/command/SparkTaskExecuteCommandTest.java | 134 +++++++++++++++++++++
3 files changed, 191 insertions(+), 3 deletions(-)
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 9da9679703..c198dff3d0 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
@@ -20,12 +20,15 @@ package org.apache.seatunnel.core.starter.spark.command;
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.FileUtils;
+
+import org.apache.spark.SparkFiles;
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 SparkConfValidateCommand implements Command<SparkCommandArgs> {
@@ -38,7 +41,9 @@ public class SparkConfValidateCommand implements
Command<SparkCommandArgs> {
@Override
public void execute() throws ConfigCheckException {
- Path configPath = FileUtils.getConfigPath(sparkCommandArgs);
+ Path configPath =
+ SparkTaskExecuteCommand.resolveConfigPath(sparkCommandArgs,
SparkFiles::get);
+ checkConfigExist(configPath);
// TODO: validate the config by new api
}
}
diff --git
a/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/main/java/org/apache/seatunnel/core/starter/spark/command/SparkTaskExecuteCommand.java
b/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/main/java/org/apache/seatunnel/core/starter/spark/command/SparkTaskExecuteCommand.java
index 67fef2ac4c..33ba4db28f 100644
---
a/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/main/java/org/apache/seatunnel/core/starter/spark/command/SparkTaskExecuteCommand.java
+++
b/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/main/java/org/apache/seatunnel/core/starter/spark/command/SparkTaskExecuteCommand.java
@@ -23,6 +23,7 @@ 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.common.config.DeployMode;
import org.apache.seatunnel.core.starter.command.Command;
import org.apache.seatunnel.core.starter.exception.CommandExecuteException;
import org.apache.seatunnel.core.starter.spark.args.SparkCommandArgs;
@@ -30,9 +31,14 @@ import
org.apache.seatunnel.core.starter.spark.execution.SparkExecution;
import org.apache.seatunnel.core.starter.utils.ConfigBuilder;
import org.apache.seatunnel.core.starter.utils.FileUtils;
+import org.apache.spark.SparkFiles;
+
import lombok.extern.slf4j.Slf4j;
+import java.nio.file.Files;
import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.function.Function;
import static
org.apache.seatunnel.core.starter.utils.FileUtils.checkConfigExist;
@@ -47,7 +53,7 @@ public class SparkTaskExecuteCommand implements
Command<SparkCommandArgs> {
@Override
public void execute() throws CommandExecuteException {
- Path configFile = FileUtils.getConfigPath(sparkCommandArgs);
+ Path configFile = resolveConfigPath(sparkCommandArgs, SparkFiles::get);
checkConfigExist(configFile);
Config config =
MetalakeConfigUtils.getMetalakeConfig(
@@ -65,4 +71,47 @@ public class SparkTaskExecuteCommand implements
Command<SparkCommandArgs> {
throw new CommandExecuteException("Run SeaTunnel on spark failed",
e);
}
}
+
+ static Path resolveConfigPath(
+ SparkCommandArgs sparkCommandArgs, Function<String, String>
sparkFilesResolver) {
+ Path configFile = FileUtils.getConfigPath(sparkCommandArgs);
+ if (!DeployMode.CLUSTER.equals(sparkCommandArgs.getDeployMode())) {
+ log.info("Resolved SeaTunnel config file: {}",
configFile.toAbsolutePath());
+ return configFile;
+ }
+
+ Path fileName = configFile.getFileName();
+ if (fileName == null) {
+ log.info("Resolved SeaTunnel config file: {}",
configFile.toAbsolutePath());
+ return configFile;
+ }
+
+ String sparkFilePath;
+ try {
+ sparkFilePath = sparkFilesResolver.apply(fileName.toString());
+ } catch (RuntimeException e) {
+ log.debug("Failed to resolve SeaTunnel config file from
SparkFiles", e);
+ log.info("Resolved SeaTunnel config file: {}",
configFile.toAbsolutePath());
+ return configFile;
+ }
+ if (sparkFilePath == null) {
+ log.info("Resolved SeaTunnel config file: {}",
configFile.toAbsolutePath());
+ return configFile;
+ }
+
+ Path shippedConfigFile = Paths.get(sparkFilePath);
+ if (Files.exists(shippedConfigFile)) {
+ log.info("Resolved SeaTunnel config file: {}",
shippedConfigFile.toAbsolutePath());
+ return shippedConfigFile;
+ }
+ if (Files.exists(configFile)) {
+ log.info("Resolved SeaTunnel config file: {}",
configFile.toAbsolutePath());
+ return configFile;
+ }
+ log.warn(
+ "SeaTunnel config file was not found in working directory {}
or SparkFiles path {}",
+ configFile.toAbsolutePath(),
+ shippedConfigFile.toAbsolutePath());
+ return configFile;
+ }
}
diff --git
a/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/test/java/org/apache/seatunnel/core/starter/spark/command/SparkTaskExecuteCommandTest.java
b/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/test/java/org/apache/seatunnel/core/starter/spark/command/SparkTaskExecuteCommandTest.java
new file mode 100644
index 0000000000..74e3e686ad
--- /dev/null
+++
b/seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/test/java/org/apache/seatunnel/core/starter/spark/command/SparkTaskExecuteCommandTest.java
@@ -0,0 +1,134 @@
+/*
+ * 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.spark.args.SparkCommandArgs;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+class SparkTaskExecuteCommandTest {
+
+ @TempDir private Path tempDir;
+
+ @Test
+ void resolveClusterConfigFromSparkFilesWhenLocalBasenameMissing() throws
IOException {
+ Path shippedConfig =
Files.createFile(tempDir.resolve("v2.batch.config"));
+ SparkCommandArgs args = new SparkCommandArgs();
+ args.setDeployMode(DeployMode.CLUSTER);
+ args.setConfigFile(configFilePath("v2.batch.config"));
+
+ Path resolvedConfig =
+ SparkTaskExecuteCommand.resolveConfigPath(
+ args, fileName ->
tempDir.resolve(fileName).toString());
+
+ Assertions.assertEquals(shippedConfig, resolvedConfig);
+ }
+
+ @Test
+ void keepClientConfigPathWithoutCallingSparkFiles() throws IOException {
+ Path clientConfig = Files.createFile(tempDir.resolve("client.conf"));
+ SparkCommandArgs args = new SparkCommandArgs();
+ args.setDeployMode(DeployMode.CLIENT);
+ args.setConfigFile(clientConfig.toString());
+ AtomicBoolean sparkFilesCalled = new AtomicBoolean(false);
+
+ Path resolvedConfig =
+ SparkTaskExecuteCommand.resolveConfigPath(
+ args,
+ fileName -> {
+ sparkFilesCalled.set(true);
+ return tempDir.resolve(fileName).toString();
+ });
+
+ Assertions.assertEquals(clientConfig, resolvedConfig);
+ Assertions.assertFalse(sparkFilesCalled.get());
+ }
+
+ @Test
+ void keepClusterConfigPathWhenSparkFilesReturnsNull() {
+ SparkCommandArgs args = new SparkCommandArgs();
+ args.setDeployMode(DeployMode.CLUSTER);
+ args.setConfigFile(configFilePath("missing.conf"));
+
+ Path resolvedConfig = SparkTaskExecuteCommand.resolveConfigPath(args,
fileName -> null);
+
+ Assertions.assertEquals(Paths.get("missing.conf"), resolvedConfig);
+ }
+
+ @Test
+ void keepClusterConfigPathWhenSparkFilesThrowsException() {
+ SparkCommandArgs args = new SparkCommandArgs();
+ args.setDeployMode(DeployMode.CLUSTER);
+ args.setConfigFile(configFilePath("missing.conf"));
+
+ Path resolvedConfig =
+ SparkTaskExecuteCommand.resolveConfigPath(
+ args,
+ fileName -> {
+ throw new NullPointerException("SparkEnv is not
initialized");
+ });
+
+ Assertions.assertEquals(Paths.get("missing.conf"), resolvedConfig);
+ }
+
+ @Test
+ void keepClusterConfigPathWhenSparkFilesPathDoesNotExist() {
+ SparkCommandArgs args = new SparkCommandArgs();
+ args.setDeployMode(DeployMode.CLUSTER);
+ args.setConfigFile(configFilePath("missing.conf"));
+
+ Path resolvedConfig =
+ SparkTaskExecuteCommand.resolveConfigPath(
+ args, fileName ->
tempDir.resolve(fileName).toString());
+
+ Assertions.assertEquals(Paths.get("missing.conf"), resolvedConfig);
+ }
+
+ @Test
+ void preferSparkFilesConfigWhenLocalBasenameExists() throws IOException {
+ Path localConfig = Paths.get("spark-local-basename-exists.conf");
+ Path shippedConfig =
Files.createFile(tempDir.resolve(localConfig.getFileName()));
+ Files.createFile(localConfig);
+ try {
+ SparkCommandArgs args = new SparkCommandArgs();
+ args.setDeployMode(DeployMode.CLUSTER);
+
args.setConfigFile(configFilePath(localConfig.getFileName().toString()));
+
+ Path resolvedConfig =
+ SparkTaskExecuteCommand.resolveConfigPath(
+ args, fileName ->
tempDir.resolve(fileName).toString());
+
+ Assertions.assertEquals(shippedConfig, resolvedConfig);
+ } finally {
+ Files.deleteIfExists(localConfig);
+ }
+ }
+
+ private String configFilePath(String fileName) {
+ return Paths.get("opt", "seatunnel", "config", fileName).toString();
+ }
+}
