This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 0f2d22cd6b [#13131] fix(core): Reject Spark jobs at submission when
Spark is not available in the local job executor (#13132)
0f2d22cd6b is described below
commit 0f2d22cd6b2fe34e8a0813b9571b2de506a639fe
Author: Jerry Shao <[email protected]>
AuthorDate: Mon Sep 14 17:58:32 2026 +0800
[#13131] fix(core): Reject Spark jobs at submission when Spark is not
available in the local job executor (#13132)
### What changes were proposed in this pull request?
Reject a Spark job at submission when the local job executor cannot
launch it, instead of accepting it and failing it asynchronously.
- `SparkProcessBuilder`: extract the `sparkHome`/`SPARK_HOME` and
`spark-submit` checks into `resolveSparkSubmit(configs)`, and replace
the mistaken `org.apache.arrow.util` imports with Guava.
- `LocalJobExecutor`:
- `submitJob()` validates Spark templates before queueing and throws
`IllegalArgumentException` if Spark is not available.
- `initialize()` logs a warning when Spark is not available. Server
startup is not blocked, since Spark is optional for the local executor.
- `JobManager.runJob()`:
- Rethrows `IllegalArgumentException` from the executor as is, so the
REST API returns 400 with the original reason instead of a generic 500.
- Removes the staging directory of a job whose submission fails.
- Docs: update `manage-jobs-in-gravitino.md` and document the 400
response of `runJob` in `docs/open-api/jobs.yaml`.
### Why are the changes needed?
With `sparkHome` and `SPARK_HOME` unset, a Spark job run request
succeeds and the job stays `QUEUED` until the next status poll, then
flips to `FAILED`. The reason appears only in the server log, so an
operator using the API or UI cannot tell which setting is missing.
Fix: #13131
### Does this PR introduce _any_ user-facing change?
Yes. Running a Spark job with the local job executor when Spark is not
available now returns 400 with the reason, and no job is created.
Previously the request succeeded and the job failed later without a
visible reason. No API or configuration keys are added or removed.
### How was this patch tested?
- Unit tests:
- `TestSparkProcessBuilder#testResolveSparkSubmit`
-
`TestLocalJobExecutor#testSubmitSparkJobRejectedWhenSparkSubmitIsNotAvailable`
- `TestJobManager#testRunJobPropagatesJobExecutorRejection`
- Integration test
`JobIT#testRunSparkJobRejectedWhenSparkIsNotAvailable` covers the client
→ REST → executor path. It verifies the rejection reason reaches the
client, no job or staging directory is left behind, and shell jobs are
unaffected.
- `./gradlew :docs:build` passes for the OpenAPI change.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5 <[email protected]>
---
.../gravitino/client/integration/test/JobIT.java | 53 ++++++++++++-
.../java/org/apache/gravitino/job/JobManager.java | 20 +++++
.../gravitino/job/local/LocalJobExecutor.java | 17 +++++
.../gravitino/job/local/SparkProcessBuilder.java | 38 ++++++++--
.../org/apache/gravitino/job/TestJobManager.java | 32 ++++++++
.../gravitino/job/local/TestLocalJobExecutor.java | 36 +++++++++
.../job/local/TestSparkProcessBuilder.java | 88 ++++++++++++++++++++++
docs/manage-jobs-in-gravitino.md | 5 +-
docs/open-api/jobs.yaml | 20 +++++
9 files changed, 298 insertions(+), 11 deletions(-)
diff --git
a/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/JobIT.java
b/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/JobIT.java
index 61bb843233..a093fca5a8 100644
---
a/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/JobIT.java
+++
b/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/JobIT.java
@@ -39,6 +39,7 @@ import org.apache.gravitino.job.JobHandle;
import org.apache.gravitino.job.JobTemplate;
import org.apache.gravitino.job.JobTemplateChange;
import org.apache.gravitino.job.ShellJobTemplate;
+import org.apache.gravitino.job.SparkJobTemplate;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
@@ -52,6 +53,7 @@ public class JobIT extends BaseIT {
private static final String METALAKE_NAME =
GravitinoITUtils.genRandomName("job_it_metalake");
private File testStagingDir;
+ private File testSparkHome;
private String testEntryScriptPath;
private String testLibScriptPath;
private ShellJobTemplate.Builder builder;
@@ -61,6 +63,9 @@ public class JobIT extends BaseIT {
@Override
public void startIntegrationTest() throws Exception {
testStagingDir = Files.createTempDirectory("test_staging_dir").toFile();
+ // A Spark home without bin/spark-submit, so Spark jobs cannot be
launched. The configuration
+ // takes precedence over the SPARK_HOME environment variable, keeping the
test deterministic.
+ testSparkHome = Files.createTempDirectory("test_spark_home").toFile();
testEntryScriptPath = generateTestEntryScript();
testLibScriptPath = generateTestLibScript();
@@ -78,7 +83,9 @@ public class JobIT extends BaseIT {
"gravitino.job.stagingDir",
testStagingDir.getAbsolutePath(),
"gravitino.job.statusPullIntervalInMs",
- "3000");
+ "3000",
+ "gravitino.jobExecutor.local.sparkHome",
+ testSparkHome.getAbsolutePath());
registerCustomConfigs(configs);
super.startIntegrationTest();
}
@@ -86,6 +93,7 @@ public class JobIT extends BaseIT {
@AfterAll
public void tearDown() throws Exception {
FileUtils.deleteDirectory(testStagingDir);
+ FileUtils.deleteDirectory(testSparkHome);
}
@BeforeEach
@@ -340,6 +348,49 @@ public class JobIT extends BaseIT {
});
}
+ @Test
+ public void testRunSparkJobRejectedWhenSparkIsNotAvailable() {
+ SparkJobTemplate template =
+ SparkJobTemplate.builder()
+ .withName("test_run_spark_without_spark_submit")
+ .withComment("Test spark job template")
+ .withExecutable(testEntryScriptPath)
+ .withClassName("org.apache.gravitino.test.SparkJob")
+ .build();
+ Assertions.assertDoesNotThrow(() ->
metalake.registerJobTemplate(template));
+
+ // The run request is rejected with the reason instead of being queued and
failing later.
+ IllegalArgumentException e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> metalake.runJob(template.name(), Collections.emptyMap()));
+ Assertions.assertTrue(
+ e.getMessage()
+ .contains(
+ "spark-submit is not found or not executable: "
+ + testSparkHome.getAbsolutePath()
+ + "/bin/spark-submit"),
+ e.getMessage());
+
+ // No job is created, and the staging directory of the rejected job is
removed.
+ Assertions.assertTrue(metalake.listJobs(template.name()).isEmpty());
+ String[] jobStagingDirs =
+ new File(testStagingDir, METALAKE_NAME + File.separator +
template.name()).list();
+ Assertions.assertTrue(jobStagingDirs == null || jobStagingDirs.length ==
0);
+
+ // Shell jobs are not affected by the missing Spark installation.
+ JobTemplate shellTemplate =
builder.withName("test_run_shell_without_spark_submit").build();
+ Assertions.assertDoesNotThrow(() ->
metalake.registerJobTemplate(shellTemplate));
+ JobHandle jobHandle =
+ metalake.runJob(
+ shellTemplate.name(),
+ ImmutableMap.of("arg1", "value1", "arg2", "success", "env_var",
"value2"));
+ Assertions.assertEquals(JobHandle.Status.QUEUED, jobHandle.jobStatus());
+ Awaitility.await()
+ .atMost(3, TimeUnit.MINUTES)
+ .until(() -> metalake.getJob(jobHandle.jobId()).jobStatus() ==
JobHandle.Status.SUCCEEDED);
+ }
+
@Test
public void testRunAndGetJob() {
JobTemplate template = builder.withName("test_run_get").build();
diff --git a/core/src/main/java/org/apache/gravitino/job/JobManager.java
b/core/src/main/java/org/apache/gravitino/job/JobManager.java
index e8365cda86..44e287a6a4 100644
--- a/core/src/main/java/org/apache/gravitino/job/JobManager.java
+++ b/core/src/main/java/org/apache/gravitino/job/JobManager.java
@@ -473,7 +473,13 @@ public class JobManager implements JobOperationDispatcher {
String jobExecutionId;
try {
jobExecutionId = jobExecutor.submitJob(jobTemplate);
+ } catch (IllegalArgumentException e) {
+ // The job executor rejects the job because it cannot be launched, for
example, a required
+ // configuration is missing. Rethrow it as is so the caller gets the
original reason.
+ deleteStagingDirOfUnsubmittedJob(jobStagingDir, jobId);
+ throw e;
} catch (Exception e) {
+ deleteStagingDirOfUnsubmittedJob(jobStagingDir, jobId);
throw new RuntimeException(
String.format("Failed to submit job template %s for execution",
jobTemplate), e);
}
@@ -1098,6 +1104,20 @@ public class JobManager implements
JobOperationDispatcher {
.build();
}
+ private void deleteStagingDirOfUnsubmittedJob(File jobStagingDir, long
jobId) {
+ // The job is not tracked by any job entity, so the periodic cleanup will
never remove its
+ // staging directory. A cleanup failure must not mask the original
submission failure.
+ try {
+ FileUtils.deleteDirectory(jobStagingDir);
+ } catch (IOException e) {
+ LOG.warn(
+ "Failed to delete staging directory {} of job {} whose submission
failed",
+ jobStagingDir,
+ jobId,
+ e);
+ }
+ }
+
private <T> T updatedValue(T currentValue, Optional<T> newValue) {
return newValue.orElse(currentValue);
}
diff --git
a/core/src/main/java/org/apache/gravitino/job/local/LocalJobExecutor.java
b/core/src/main/java/org/apache/gravitino/job/local/LocalJobExecutor.java
index f3b6e5d631..295a0a8483 100644
--- a/core/src/main/java/org/apache/gravitino/job/local/LocalJobExecutor.java
+++ b/core/src/main/java/org/apache/gravitino/job/local/LocalJobExecutor.java
@@ -44,6 +44,7 @@ import org.apache.gravitino.connector.job.JobExecutor;
import org.apache.gravitino.exceptions.NoSuchJobException;
import org.apache.gravitino.job.JobHandle;
import org.apache.gravitino.job.JobTemplate;
+import org.apache.gravitino.job.SparkJobTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -150,10 +151,26 @@ public class LocalJobExecutor implements JobExecutor {
TimeUnit.MILLISECONDS);
this.runningProcesses = Maps.newConcurrentMap();
+
+ // Spark is optional for the local job executor, so a missing Spark
installation must not fail
+ // the server startup. Warn early instead; Spark jobs will be rejected at
submission.
+ try {
+ SparkProcessBuilder.resolveSparkSubmit(configs);
+ } catch (IllegalArgumentException e) {
+ LOG.warn(
+ "Spark jobs cannot be run by the local job executor and will be
rejected: {}",
+ e.getMessage());
+ }
}
@Override
public String submitJob(JobTemplate jobTemplate) {
+ // Validate the job can be launched before queueing it, so that a
misconfiguration is reported
+ // to the caller directly instead of only failing the job asynchronously
in the worker thread.
+ if (jobTemplate instanceof SparkJobTemplate) {
+ SparkProcessBuilder.resolveSparkSubmit(configs);
+ }
+
String newJobId = LOCAL_JOB_PREFIX + UUID.randomUUID();
Pair<String, JobTemplate> jobPair = Pair.of(newJobId, jobTemplate);
diff --git
a/core/src/main/java/org/apache/gravitino/job/local/SparkProcessBuilder.java
b/core/src/main/java/org/apache/gravitino/job/local/SparkProcessBuilder.java
index 918c48f564..06d9cef8e5 100644
--- a/core/src/main/java/org/apache/gravitino/job/local/SparkProcessBuilder.java
+++ b/core/src/main/java/org/apache/gravitino/job/local/SparkProcessBuilder.java
@@ -21,14 +21,15 @@ package org.apache.gravitino.job.local;
import static
org.apache.gravitino.job.local.LocalJobExecutorConfigs.SPARK_HOME;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Joiner;
+import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import java.io.File;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-import org.apache.arrow.util.Preconditions;
-import org.apache.arrow.util.VisibleForTesting;
+import javax.annotation.Nullable;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.job.SparkJobTemplate;
import org.slf4j.Logger;
@@ -51,18 +52,39 @@ public class SparkProcessBuilder extends
LocalProcessBuilder {
protected SparkProcessBuilder(SparkJobTemplate sparkJobTemplate, Map<String,
String> configs) {
super(sparkJobTemplate, configs);
- String sparkHome =
-
Optional.ofNullable(configs.get(SPARK_HOME)).orElse(System.getenv(ENV_SPARK_HOME));
+ this.sparkSubmit = resolveSparkSubmit(configs);
+ }
+
+ /**
+ * Resolves the spark-submit executable from the local job executor
configurations, falling back
+ * to the {@code SPARK_HOME} environment variable.
+ *
+ * @param configs The local job executor configurations.
+ * @return The absolute path of the spark-submit executable.
+ * @throws IllegalArgumentException If neither the Spark home configuration
nor the {@code
+ * SPARK_HOME} environment variable is set, or spark-submit is not found
or not executable.
+ */
+ static String resolveSparkSubmit(Map<String, String> configs) {
+ return resolveSparkSubmit(configs, System.getenv(ENV_SPARK_HOME));
+ }
+
+ @VisibleForTesting
+ static String resolveSparkSubmit(Map<String, String> configs, @Nullable
String envSparkHome) {
+ String sparkHome =
Optional.ofNullable(configs.get(SPARK_HOME)).orElse(envSparkHome);
Preconditions.checkArgument(
StringUtils.isNotBlank(sparkHome),
"gravitino.jobExecutor.local.sparkHome or SPARK_HOME environment
variable must"
+ " be set for Spark jobs");
- this.sparkSubmit = sparkHome + "/bin/spark-submit";
- File sparkSubmitFile = new File(sparkSubmit);
+ // Resolve to an absolute path: the Spark process runs in the job staging
directory, so a
+ // relative path validated against the server working directory would not
be found there.
+ File sparkSubmitFile = new File(sparkHome,
"bin/spark-submit").getAbsoluteFile();
+ // canExecute() alone is also true for a searchable directory, so require
a regular file.
Preconditions.checkArgument(
- sparkSubmitFile.canExecute(),
- "spark-submit is not found or not executable: " + sparkSubmit);
+ sparkSubmitFile.isFile() && sparkSubmitFile.canExecute(),
+ "spark-submit is not found or not executable: %s",
+ sparkSubmitFile);
+ return sparkSubmitFile.getPath();
}
@VisibleForTesting
diff --git a/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
b/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
index 011281f3f0..1ca2f43e6a 100644
--- a/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
+++ b/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
@@ -661,6 +661,38 @@ public class TestJobManager {
() -> jobManager.runJob(metalake, "shell_job",
Collections.emptyMap()));
}
+ @Test
+ public void testRunJobPropagatesJobExecutorRejection() throws IOException {
+ mockedMetalake
+ .when(() -> MetalakeManager.checkMetalake(metalakeIdent, entityStore))
+ .thenAnswer(a -> null);
+
+ JobTemplateEntity shellJobTemplate =
+ newShellJobTemplateEntity("shell_job", "A shell job template");
+ when(jobManager.getJobTemplate(metalake,
shellJobTemplate.name())).thenReturn(shellJobTemplate);
+
+ IllegalArgumentException rejection =
+ new IllegalArgumentException(
+ "gravitino.jobExecutor.local.sparkHome or SPARK_HOME environment
variable must"
+ + " be set for Spark jobs");
+ doThrow(rejection).when(jobExecutor).submitJob(any());
+
+ // The rejection must reach the caller as is, so the REST layer reports
the original reason
+ // with a 400 instead of wrapping it into a generic 500 error.
+ IllegalArgumentException e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> jobManager.runJob(metalake, "shell_job",
Collections.emptyMap()));
+ Assertions.assertSame(rejection, e);
+
+ // No job entity is registered and the staging directory of the rejected
job is removed.
+ verify(entityStore, never()).put(any(JobEntity.class), anyBoolean());
+ File templateStagingDir =
+ new File(testStagingDir, metalake + File.separator +
shellJobTemplate.name());
+ String[] jobStagingDirs = templateStagingDir.list();
+ Assertions.assertTrue(jobStagingDirs == null || jobStagingDirs.length ==
0);
+ }
+
@Test
public void testRunJobPopulatesResolvedRuntimeJobTemplate() throws
IOException {
mockedMetalake
diff --git
a/core/src/test/java/org/apache/gravitino/job/local/TestLocalJobExecutor.java
b/core/src/test/java/org/apache/gravitino/job/local/TestLocalJobExecutor.java
index 1c7cbfae10..45b49e01e4 100644
---
a/core/src/test/java/org/apache/gravitino/job/local/TestLocalJobExecutor.java
+++
b/core/src/test/java/org/apache/gravitino/job/local/TestLocalJobExecutor.java
@@ -33,6 +33,7 @@ import org.apache.gravitino.connector.job.JobExecutor;
import org.apache.gravitino.job.JobHandle;
import org.apache.gravitino.job.JobManager;
import org.apache.gravitino.job.JobTemplate;
+import org.apache.gravitino.job.SparkJobTemplate;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.JobTemplateEntity;
import org.apache.gravitino.utils.NamespaceUtil;
@@ -159,6 +160,41 @@ public class TestLocalJobExecutor {
Assertions.assertEquals(JobHandle.Status.FAILED,
jobExecutor.getJobStatus(jobId));
}
+ @Test
+ public void testSubmitSparkJobRejectedWhenSparkSubmitIsNotAvailable() throws
IOException {
+ File sparkHome = new File(workingDir, "spark");
+ LocalJobExecutor exec = new LocalJobExecutor();
+ exec.initialize(
+ ImmutableMap.of(LocalJobExecutorConfigs.SPARK_HOME,
sparkHome.getAbsolutePath()));
+
+ try {
+ SparkJobTemplate template =
+ SparkJobTemplate.builder()
+ .withName("spark-job")
+ .withExecutable(new File(workingDir,
"spark-demo.jar").getAbsolutePath())
+ .withClassName("com.example.MainClass")
+ .build();
+
+ // spark-submit does not exist, the job is rejected at submission
instead of being queued.
+ IllegalArgumentException e =
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
exec.submitJob(template));
+ Assertions.assertTrue(e.getMessage().contains("spark-submit is not found
or not executable"));
+
+ // Once spark-submit is available, the same job is accepted.
+ File sparkSubmit = new File(sparkHome, "bin/spark-submit");
+ FileUtils.writeStringToFile(sparkSubmit, "#!/bin/sh\nexit 0\n", "UTF-8");
+ Assertions.assertTrue(sparkSubmit.setExecutable(true));
+
+ String jobId = exec.submitJob(template);
+ Assertions.assertNotNull(jobId);
+ Awaitility.await()
+ .atMost(1, TimeUnit.MINUTES)
+ .until(() -> exec.getJobStatus(jobId) == JobHandle.Status.SUCCEEDED);
+ } finally {
+ exec.close();
+ }
+ }
+
@Test
public void testCancelJob() throws InterruptedException {
Map<String, String> jobConf =
diff --git
a/core/src/test/java/org/apache/gravitino/job/local/TestSparkProcessBuilder.java
b/core/src/test/java/org/apache/gravitino/job/local/TestSparkProcessBuilder.java
index 12c2a69325..e3d8f4295d 100644
---
a/core/src/test/java/org/apache/gravitino/job/local/TestSparkProcessBuilder.java
+++
b/core/src/test/java/org/apache/gravitino/job/local/TestSparkProcessBuilder.java
@@ -22,8 +22,14 @@ package org.apache.gravitino.job.local;
import com.google.common.base.Joiner;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
+import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.job.SparkJobTemplate;
import org.junit.jupiter.api.Assertions;
@@ -144,6 +150,88 @@ public class TestSparkProcessBuilder {
Assertions.assertTrue(command4.contains("arg2"));
}
+ @Test
+ public void testResolveSparkSubmit() throws IOException {
+ File validSparkHome =
Files.createTempDirectory("gravitino-test-spark-home").toFile();
+ File invalidSparkHome =
Files.createTempDirectory("gravitino-test-no-spark-home").toFile();
+ try {
+ File sparkSubmit = new File(validSparkHome, "bin/spark-submit");
+ FileUtils.writeStringToFile(sparkSubmit, "#!/bin/sh\n", "UTF-8");
+ Assertions.assertTrue(sparkSubmit.setExecutable(true));
+ String validPath = validSparkHome.getAbsolutePath();
+ String invalidPath = invalidSparkHome.getAbsolutePath();
+
+ // Neither the configuration nor the environment variable is set.
+ IllegalArgumentException e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
SparkProcessBuilder.resolveSparkSubmit(Collections.emptyMap(), null));
+ Assertions.assertEquals(
+ "gravitino.jobExecutor.local.sparkHome or SPARK_HOME environment
variable must"
+ + " be set for Spark jobs",
+ e.getMessage());
+
+ // Falls back to the environment variable when the configuration is not
set.
+ Assertions.assertEquals(
+ sparkSubmit.getAbsolutePath(),
+ SparkProcessBuilder.resolveSparkSubmit(Collections.emptyMap(),
validPath));
+
+ // The configuration takes precedence over the environment variable.
+ Assertions.assertEquals(
+ sparkSubmit.getAbsolutePath(),
+ SparkProcessBuilder.resolveSparkSubmit(
+ ImmutableMap.of(LocalJobExecutorConfigs.SPARK_HOME, validPath),
invalidPath));
+ e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ SparkProcessBuilder.resolveSparkSubmit(
+ ImmutableMap.of(LocalJobExecutorConfigs.SPARK_HOME,
invalidPath), validPath));
+ Assertions.assertEquals(
+ "spark-submit is not found or not executable: " + invalidPath +
"/bin/spark-submit",
+ e.getMessage());
+
+ // A directory at bin/spark-submit is not a valid executable, even if it
is searchable.
+ File sparkSubmitDir = new File(invalidSparkHome, "bin/spark-submit");
+ Assertions.assertTrue(sparkSubmitDir.mkdirs());
+ Assertions.assertTrue(sparkSubmitDir.setExecutable(true));
+ e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
SparkProcessBuilder.resolveSparkSubmit(Collections.emptyMap(), invalidPath));
+ Assertions.assertEquals(
+ "spark-submit is not found or not executable: " + invalidPath +
"/bin/spark-submit",
+ e.getMessage());
+ } finally {
+ FileUtils.deleteDirectory(validSparkHome);
+ FileUtils.deleteDirectory(invalidSparkHome);
+ }
+ }
+
+ @Test
+ public void testResolveSparkSubmitWithRelativeSparkHome() throws IOException
{
+ // Created under the current working directory, so it is referenced by a
relative path.
+ File relativeSparkHome =
+ Files.createTempDirectory(Paths.get(""),
"gravitino-test-relative-spark-home").toFile();
+ try {
+ Assertions.assertFalse(relativeSparkHome.isAbsolute());
+ File sparkSubmit = new File(relativeSparkHome, "bin/spark-submit");
+ FileUtils.writeStringToFile(sparkSubmit, "#!/bin/sh\n", "UTF-8");
+ Assertions.assertTrue(sparkSubmit.setExecutable(true));
+
+ // The Spark process runs in the job staging directory, so the resolved
path must be
+ // absolute rather than relative to the server working directory.
+ String resolved =
+ SparkProcessBuilder.resolveSparkSubmit(
+ ImmutableMap.of(LocalJobExecutorConfigs.SPARK_HOME,
relativeSparkHome.getPath()),
+ null);
+ Assertions.assertTrue(new File(resolved).isAbsolute());
+ Assertions.assertEquals(sparkSubmit.getAbsolutePath(), resolved);
+ } finally {
+ FileUtils.deleteDirectory(relativeSparkHome);
+ }
+ }
+
@Test
public void testSparkEnvironmentInfoOutputWithoutValue() {
SparkJobTemplate template =
diff --git a/docs/manage-jobs-in-gravitino.md b/docs/manage-jobs-in-gravitino.md
index 67fddc8901..c3e93de708 100644
--- a/docs/manage-jobs-in-gravitino.md
+++ b/docs/manage-jobs-in-gravitino.md
@@ -69,8 +69,9 @@ curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \
### Register a Spark Template
A Spark template submits an application. Running one with the local executor
needs either
-`gravitino.jobExecutor.local.sparkHome` or `SPARK_HOME` set before the server
starts, or the job
-fails to launch.
+`gravitino.jobExecutor.local.sparkHome` or `SPARK_HOME` set before the server
starts, pointing to a
+Spark installation with an executable `bin/spark-submit`. Otherwise, the run
request is rejected with
+an error that names the missing setting, and no job is created.
```json
{
diff --git a/docs/open-api/jobs.yaml b/docs/open-api/jobs.yaml
index 7729301263..40a20b5060 100644
--- a/docs/open-api/jobs.yaml
+++ b/docs/open-api/jobs.yaml
@@ -255,6 +255,15 @@ paths:
examples:
JobTemplateAlreadyExistsException:
$ref: "#/components/examples/NoSuchJobTemplateException"
+ "400":
+ description: Bad Request - The job cannot be launched by the job
executor, for example, a required job executor configuration is missing
+ content:
+ application/vnd.gravitino.v1+json:
+ schema:
+ $ref: "./openapi.yaml#/components/schemas/ErrorModel"
+ examples:
+ JobRunIllegalArgumentException:
+ $ref: "#/components/examples/JobRunIllegalArgumentException"
"5xx":
$ref: "./openapi.yaml#/components/responses/ServerErrorResponse"
@@ -1140,3 +1149,14 @@ components:
"..."
]
}
+
+ JobRunIllegalArgumentException:
+ value: {
+ "code": 1001,
+ "type": "IllegalArgumentException",
+ "message": "Failed to operate job(s) operation [RUN] under object
[my_test_metalake], reason [gravitino.jobExecutor.local.sparkHome or SPARK_HOME
environment variable must be set for Spark jobs]",
+ "stack": [
+ "java.lang.IllegalArgumentException:
gravitino.jobExecutor.local.sparkHome or SPARK_HOME environment variable must
be set for Spark jobs",
+ "..."
+ ]
+ }
