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 2e00a23e49 [#13616] feat(k8s-job-executor): Register the k8s job
executor, ship it and run built-in jobs on it (#13892)
2e00a23e49 is described below
commit 2e00a23e4993f07c970c1e9e0e698430b7a41210
Author: Jerry Shao <[email protected]>
AuthorDate: Mon Oct 19 17:04:22 2026 +0800
[#13616] feat(k8s-job-executor): Register the k8s job executor, ship it and
run built-in jobs on it (#13892)
### What changes were proposed in this pull request?
The `k8s` job executor added in #13613 can now be used by Gravitino, and
runs the built-in Spark jobs.
- **Registration**: `JobExecutorFactory` registers `k8s` by class name,
as the executor is a plugin that depends on `core`. The name and class
name are public constants of `JobExecutorFactory`, which the plugin
uses, and a plugin test checks the class name against `K8sJobExecutor`.
If the class can't be loaded, the error names the missing
`gravitino-k8s-job-executor-<version>.jar` and the `libs` directory.
- **Distribution**: the shaded jar of the executor is copied into `libs`
of the distribution package. `LICENSE.bin` already lists everything it
bundles (fabric8, zjsonpatch, Jackson, SnakeYAML, SnakeYAML Engine).
- **Smoke test of the shaded jar**: a separate `shadedJarTest` source
set and task, part of `check`, puts the shaded jar on the runtime
classpath of `:server` (with its Jackson 2.15.2), without the module's
classes or unshaded dependencies. It checks that Jackson and SnakeYAML
are relocated, that the jar bundles no class that is also on the server
classpath, and creates the executor through `JobExecutorFactory` to
submit a built-in job to a stub API server. It only needs `shadowJar`,
not a distribution build.
- **Built-in jobs**: the executable of a built-in template is the path
of the jobs jar on the Gravitino server, which a Spark pod can't reach.
The new `spark.builtinJobsJar` is the URI of the jobs jar that the
driver and executors can reach,
`local:///opt/gravitino/jobs/gravitino-jobs.jar` by default: a file in
the Spark image or in a mounted volume, or `https://`, `s3a://` and so
on. A template whose name starts with `builtin-` and whose executable is
a path on the server runs that jar instead. An executable that the
cluster can already reach is kept. The stored templates don't change,
and the `local` executor keeps using the jar on the server.
- The check of whether a resource is reachable from the cluster moves to
`K8sJobResourceUtils.isClusterReachable`, so that the resource check,
the `spark.builtinJobsJar` validation and the substitution share it.
### Why are the changes needed?
Without this, `gravitino.job.executor = k8s` fails to load the executor,
and the built-in jobs can't run on it.
Fix: #13616
### Does this PR introduce _any_ user-facing change?
Yes.
- `gravitino.job.executor = k8s` selects the `k8s` job executor,
configured with the `gravitino.jobExecutor.k8s.` keys added in #13613.
- New key `gravitino.jobExecutor.k8s.spark.builtinJobsJar`, default
`local:///opt/gravitino/jobs/gravitino-jobs.jar`.
- The distribution package ships
`libs/gravitino-k8s-job-executor-<version>.jar`.
The `gravitino-spark` image that holds the jobs jar at the default path,
and the docs, come with #13618.
### How was this patch tested?
- Unit tests for the factory, the new configuration and the
substitution, and the `shadedJarTest` smoke test.
- Manually, with a Gravitino server started from the distribution
package of this branch against a local Kubernetes cluster (OrbStack)
with spark-kubernetes-operator 0.9.0, through the REST API:
- the built-in Spark Pi job: success with its times and driver output,
cancel, a failing job, and a user template whose executable is a path on
the server, which is rejected;
- the jobs jar given as an `http://` link through
`spark.builtinJobsJar`;
- built-in Iceberg jobs on an image without the Iceberg runtime, with
the runtime given through
`gravitino.jobExecutor.k8s.spark.conf.spark.jars` or
`spark.conf.spark.jars.packages` (with `spark.jars.ivy`). Passing
`spark.jars` in the job's `spark_conf` argument doesn't work, which
#13703 addresses.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
Co-authored-by: Qi Yu <[email protected]>
---
build.gradle.kts | 3 +-
.../main/java/org/apache/gravitino/Configs.java | 5 +-
.../apache/gravitino/job/JobExecutorFactory.java | 60 +++++-
.../gravitino/job/TestJobExecutorFactory.java | 43 ++++
docs/development/custom-job-executor.md | 3 +
docs/gravitino-server-config.md | 2 +-
plugins/k8s-job-executor/build.gradle.kts | 52 +++++
.../apache/gravitino/job/k8s/K8sJobExecutor.java | 6 +-
.../gravitino/job/k8s/K8sJobExecutorConfigs.java | 35 +++-
.../gravitino/job/k8s/K8sJobResourceUtils.java | 18 ++
.../job/k8s/spark/SparkApplicationUtils.java | 45 ++++-
.../job/k8s/TestK8sJobExecutorShadedJar.java | 222 +++++++++++++++++++++
.../gravitino/job/k8s/TestK8sJobExecutor.java | 10 +
.../job/k8s/TestK8sJobExecutorConfigs.java | 8 +
.../gravitino/job/k8s/TestK8sJobResourceUtils.java | 20 ++
.../job/k8s/spark/TestSparkApplicationUtils.java | 64 ++++++
16 files changed, 574 insertions(+), 22 deletions(-)
diff --git a/build.gradle.kts b/build.gradle.kts
index c2daee27e3..a1295f06c3 100644
--- a/build.gradle.kts
+++ b/build.gradle.kts
@@ -1187,7 +1187,8 @@ tasks {
":lance:lance-rest-server:copyLibAndConfigs",
":maintenance:optimizer:copyLibAndConfigs",
":plugins:idp-basic:copyLibAndConfigs",
- ":plugins:tms:copyLibAndConfigs"
+ ":plugins:tms:copyLibAndConfigs",
+ ":plugins:k8s-job-executor:copyLibAndConfigs"
)
if (!skipWeb) {
dependencies.add(":web:web:build")
diff --git a/core/src/main/java/org/apache/gravitino/Configs.java
b/core/src/main/java/org/apache/gravitino/Configs.java
index e41901efa9..f85ac413d2 100644
--- a/core/src/main/java/org/apache/gravitino/Configs.java
+++ b/core/src/main/java/org/apache/gravitino/Configs.java
@@ -620,8 +620,9 @@ public class Configs {
public static final ConfigEntry<String> JOB_EXECUTOR =
new ConfigBuilder("gravitino.job.executor")
.doc(
- "The executor to run jobs, by default it is 'local', user can
implement their own "
- + "executor and set it here.")
+ "The executor to run jobs: one of the built-in executors 'local'
(the default) and "
+ + "'k8s', or the name of a custom executor whose class is
set by "
+ + "gravitino.jobExecutor.<name>.class.")
.version(ConfigConstants.VERSION_1_0_0)
.stringConf()
.checkValue(StringUtils::isNotBlank,
ConfigConstants.NOT_BLANK_ERROR_MSG)
diff --git
a/core/src/main/java/org/apache/gravitino/job/JobExecutorFactory.java
b/core/src/main/java/org/apache/gravitino/job/JobExecutorFactory.java
index 821a5d3395..fa96348dcd 100644
--- a/core/src/main/java/org/apache/gravitino/job/JobExecutorFactory.java
+++ b/core/src/main/java/org/apache/gravitino/job/JobExecutorFactory.java
@@ -32,17 +32,36 @@ import org.apache.gravitino.connector.job.JobContext;
import org.apache.gravitino.connector.job.JobExecutor;
import org.apache.gravitino.job.local.LocalJobExecutor;
import org.apache.gravitino.job.local.LocalJobExecutorConfigs;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class JobExecutorFactory {
+ private static final Logger LOG =
LoggerFactory.getLogger(JobExecutorFactory.class);
+
+ /** The name of the built-in k8s job executor. */
+ public static final String K8S_JOB_EXECUTOR_NAME = "k8s";
+
+ /**
+ * The class name of the built-in k8s job executor. The job executor is a
plugin that depends on
+ * core, so core can only refer to its class by name. The plugin uses these
constants and checks
+ * that the name is the one of its class.
+ */
+ public static final String K8S_JOB_EXECUTOR_CLASS_NAME =
+ "org.apache.gravitino.job.k8s.K8sJobExecutor";
+
private static final String JOB_EXECUTOR_CONF_PREFIX =
"gravitino.jobExecutor.";
private static final String JOB_EXECUTOR_CLASS_SUFFIX = ".class";
+ private static final String K8S_JOB_EXECUTOR_JAR =
"gravitino-k8s-job-executor-<version>.jar";
+
private static final Map<String, String> BUILTIN_EXECUTORS =
ImmutableMap.of(
LocalJobExecutorConfigs.LOCAL_JOB_EXECUTOR_NAME,
- LocalJobExecutor.class.getCanonicalName());
+ LocalJobExecutor.class.getCanonicalName(),
+ K8S_JOB_EXECUTOR_NAME,
+ K8S_JOB_EXECUTOR_CLASS_NAME);
private JobExecutorFactory() {
// Private constructor to prevent instantiation
@@ -51,11 +70,21 @@ public class JobExecutorFactory {
public static JobExecutor create(Config config) {
String jobExecutorName = config.get(Configs.JOB_EXECUTOR);
String clzName;
+ String jobExecutorClassKey =
+ JOB_EXECUTOR_CONF_PREFIX + jobExecutorName + JOB_EXECUTOR_CLASS_SUFFIX;
if (BUILTIN_EXECUTORS.containsKey(jobExecutorName)) {
clzName = BUILTIN_EXECUTORS.get(jobExecutorName);
+ String configuredClzName =
StringUtils.trimToNull(config.getRawString(jobExecutorClassKey));
+ if (configuredClzName != null && !configuredClzName.equals(clzName)) {
+ LOG.warn(
+ "Job executor {} is built in, so {} = {} is ignored and {} is
used. Give a custom job"
+ + " executor another name.",
+ jobExecutorName,
+ jobExecutorClassKey,
+ configuredClzName,
+ clzName);
+ }
} else {
- String jobExecutorClassKey =
- JOB_EXECUTOR_CONF_PREFIX + jobExecutorName +
JOB_EXECUTOR_CLASS_SUFFIX;
clzName = config.getRawString(jobExecutorClassKey);
}
@@ -67,8 +96,8 @@ public class JobExecutorFactory {
Map<String, String> configs =
Maps.newHashMap(
config.getConfigsWithPrefix(JOB_EXECUTOR_CONF_PREFIX +
jobExecutorName + "."));
+ Class<?> jobExecutorClass = loadJobExecutorClass(jobExecutorName, clzName);
try {
- Class<?> jobExecutorClass = Class.forName(clzName);
checkJobExecutorClass(jobExecutorClass);
JobExecutor jobExecutor =
(JobExecutor)
jobExecutorClass.getDeclaredConstructor().newInstance();
@@ -140,6 +169,29 @@ public class JobExecutorFactory {
jobExecutorClass.getName());
}
+ private static Class<?> loadJobExecutorClass(String jobExecutorName, String
clzName) {
+ try {
+ return Class.forName(clzName);
+ } catch (ClassNotFoundException | LinkageError e) {
+ // A LinkageError is thrown when the class is there, but a class it
depends on isn't.
+ String hint =
+ K8S_JOB_EXECUTOR_CLASS_NAME.equals(clzName)
+ ? String.format(
+ "It is shipped in %s, make sure the jar is in the libs
directory of the"
+ + " Gravitino server",
+ K8S_JOB_EXECUTOR_JAR)
+ : String.format(
+ "Check %s%s%s, and make sure the jar of the job executor is
on the classpath"
+ + " of the Gravitino server",
+ JOB_EXECUTOR_CONF_PREFIX, jobExecutorName,
JOB_EXECUTOR_CLASS_SUFFIX);
+ throw new RuntimeException(
+ String.format(
+ "Failed to create job executor %s: its class %s can't be loaded.
%s",
+ jobExecutorName, clzName, hint),
+ e);
+ }
+ }
+
private static boolean implementsMethod(
Class<?> jobExecutorClass, String name, Class<?>... parameterTypes) {
// An inherited default method of the interface is declared by the
interface itself.
diff --git
a/core/src/test/java/org/apache/gravitino/job/TestJobExecutorFactory.java
b/core/src/test/java/org/apache/gravitino/job/TestJobExecutorFactory.java
index ff2bfee6c7..293d6c5f89 100644
--- a/core/src/test/java/org/apache/gravitino/job/TestJobExecutorFactory.java
+++ b/core/src/test/java/org/apache/gravitino/job/TestJobExecutorFactory.java
@@ -135,6 +135,49 @@ public class TestJobExecutorFactory {
}
}
+ @Test
+ public void testBuiltinJobExecutorIgnoresConfiguredClass() throws
IOException {
+ // The name of a built-in job executor is reserved, a class configured for
it is ignored.
+ config.loadFromMap(
+ ImmutableMap.of(
+ "gravitino.jobExecutor.local.class",
CustomLocalJobExecutor.class.getName()),
+ key -> true);
+
+ try (JobExecutor executor = JobExecutorFactory.create(config)) {
+ Assertions.assertEquals(LocalJobExecutor.class, executor.getClass());
+ }
+ }
+
+ @Test
+ public void testK8sJobExecutorWithoutItsJar() {
+ // The k8s job executor is registered by class name, and its jar isn't on
the classpath of core.
+ config.set(Configs.JOB_EXECUTOR, JobExecutorFactory.K8S_JOB_EXECUTOR_NAME);
+
+ RuntimeException e =
+ Assertions.assertThrows(RuntimeException.class, () ->
JobExecutorFactory.create(config));
+ Assertions.assertInstanceOf(ClassNotFoundException.class, e.getCause());
+ Assertions.assertTrue(
+
e.getMessage().contains(JobExecutorFactory.K8S_JOB_EXECUTOR_CLASS_NAME),
e.getMessage());
+
Assertions.assertTrue(e.getMessage().contains("gravitino-k8s-job-executor"),
e.getMessage());
+ Assertions.assertTrue(e.getMessage().contains("libs"), e.getMessage());
+ }
+
+ @Test
+ public void testCustomJobExecutorClassNotFound() {
+ config.set(Configs.JOB_EXECUTOR, "missing");
+ config.loadFromMap(
+ ImmutableMap.of("gravitino.jobExecutor.missing.class",
"com.example.MissingJobExecutor"),
+ key -> true);
+
+ RuntimeException e =
+ Assertions.assertThrows(RuntimeException.class, () ->
JobExecutorFactory.create(config));
+ Assertions.assertInstanceOf(ClassNotFoundException.class, e.getCause());
+ Assertions.assertTrue(
+ e.getMessage().contains("com.example.MissingJobExecutor"),
e.getMessage());
+ Assertions.assertTrue(
+ e.getMessage().contains("gravitino.jobExecutor.missing.class"),
e.getMessage());
+ }
+
@Test
public void testRejectJobExecutorBuiltAgainstOldSpi() throws Exception {
// A job executor plugin built before getJobExecutionInfo was added to the
SPI only
diff --git a/docs/development/custom-job-executor.md
b/docs/development/custom-job-executor.md
index 377929b7f3..c89d4f0886 100644
--- a/docs/development/custom-job-executor.md
+++ b/docs/development/custom-job-executor.md
@@ -67,6 +67,9 @@ gravitino.job.executor = airflow
gravitino.jobExecutor.airflow.class = com.example.MyAirflowJobExecutor
```
+The names `local` and `k8s` belong to the built-in job executors, so give your
job executor another
+name: a `gravitino.jobExecutor.local.class` or
`gravitino.jobExecutor.k8s.class` setting is ignored.
+
Configure the job executor with additional properties, like:
```
diff --git a/docs/gravitino-server-config.md b/docs/gravitino-server-config.md
index a76e5fe596..2df4374ed7 100644
--- a/docs/gravitino-server-config.md
+++ b/docs/gravitino-server-config.md
@@ -629,7 +629,7 @@ server, are documented with those services. See
| Configuration Item | Description
| Default Value |
|----------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------|-------------------------------|
-| `gravitino.job.executor` | Executor that runs jobs. Implement
your own and name it here to replace the built-in one.
| `local` |
+| `gravitino.job.executor` | Executor that runs jobs: the
built-in `local` or `k8s`, or the name of a custom executor whose class is set
by `gravitino.jobExecutor.<name>.class`. | `local`
|
| `gravitino.job.stagingDir` | Directory holding staging files for
running jobs. With multiple servers, put it on storage shared by all servers so
that any server can return a job's output. | `/tmp/gravitino/jobs/staging` |
| `gravitino.job.stagingDirKeepTimeInMs` | How long in milliseconds a finished
job's staging files are kept. Use at least 10 minutes outside testing.
| `604800000` (7 days) |
| `gravitino.job.statusPullIntervalInMs` | Interval in milliseconds between
job status polls. Use at least 1 minute outside testing.
| `300000` (5 minutes) |
diff --git a/plugins/k8s-job-executor/build.gradle.kts
b/plugins/k8s-job-executor/build.gradle.kts
index 0005b4832a..ec06621218 100644
--- a/plugins/k8s-job-executor/build.gradle.kts
+++ b/plugins/k8s-job-executor/build.gradle.kts
@@ -24,6 +24,12 @@ plugins {
alias(libs.plugins.shadow)
}
+// Tests of the shaded jar on the classpath of the Gravitino server, without
the classes and the
+// dependencies of this module, to catch relocation mistakes and dependency
conflicts.
+val shadedJarTest: SourceSet by sourceSets.creating
+val shadedJarTestImplementation: Configuration by configurations.getting
+val shadedJarTestRuntimeOnly: Configuration by configurations.getting
+
dependencies {
annotationProcessor(libs.lombok)
@@ -53,6 +59,17 @@ dependencies {
testImplementation(libs.junit.jupiter.params)
testRuntimeOnly(libs.junit.jupiter.engine)
+
+ shadedJarTestImplementation(project(":api"))
+ shadedJarTestImplementation(project(":common"))
+ shadedJarTestImplementation(project(":core"))
+ shadedJarTestImplementation(libs.guava)
+ shadedJarTestImplementation(libs.jackson.databind)
+ shadedJarTestImplementation(libs.junit.jupiter.api)
+
+ shadedJarTestRuntimeOnly(project(":server"))
+ shadedJarTestRuntimeOnly(files(tasks.named<ShadowJar>("shadowJar").flatMap {
it.archiveFile }))
+ shadedJarTestRuntimeOnly(libs.junit.jupiter.engine)
}
tasks.withType(ShadowJar::class.java) {
@@ -83,3 +100,38 @@ tasks.jar {
dependsOn(tasks.named("shadowJar"))
archiveClassifier.set("empty")
}
+
+tasks {
+ val shadedJarTestTask = register<Test>("shadedJarTest") {
+ group = "verification"
+ description = "Tests the shaded jar on the classpath of the Gravitino
server"
+ testClassesDirs = shadedJarTest.output.classesDirs
+ classpath = shadedJarTest.runtimeClasspath
+
+ val shadedJar = named<ShadowJar>("shadowJar").flatMap { it.archiveFile }
+ inputs.file(shadedJar)
+ doFirst {
+ systemProperty("gravitino.k8s.shadedJar",
shadedJar.get().asFile.absolutePath)
+ }
+ }
+
+ check {
+ dependsOn(shadedJarTestTask)
+ }
+
+ val copyLibs by registering(Copy::class) {
+ dependsOn(named("shadowJar"))
+ from(layout.buildDirectory.dir("libs")) {
+ include("gravitino-k8s-job-executor-*.jar")
+ exclude("*-empty.jar", "*-javadoc.jar", "*-sources.jar")
+ }
+ into("$rootDir/distribution/package/libs")
+ duplicatesStrategy = DuplicatesStrategy.EXCLUDE
+ }
+
+ register("copyLibAndConfigs", Copy::class) {
+ group = "gravitino distribution"
+ description = "Copy the k8s job executor jar into distribution package
libs"
+ dependsOn(copyLibs)
+ }
+}
diff --git
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutor.java
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutor.java
index 1433f860bb..5e6583f32d 100644
---
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutor.java
+++
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutor.java
@@ -142,10 +142,12 @@ public class K8sJobExecutor implements JobExecutor {
&& isUnreachable((KubernetesClientException) e),
clock::millis);
LOG.info(
- "Initialized the k8s job executor for cluster {} at {}, default
namespace {}",
+ "Initialized the k8s job executor for cluster {} at {}, default
namespace {}, built-in"
+ + " jobs jar {}",
configs.cluster(),
client.getMasterUrl(),
- configs.namespace());
+ configs.namespace(),
+ configs.sparkBuiltinJobsJar());
}
@Override
diff --git
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutorConfigs.java
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutorConfigs.java
index b4b8eea85d..129a25c202 100644
---
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutorConfigs.java
+++
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutorConfigs.java
@@ -25,6 +25,7 @@ import java.util.TreeMap;
import java.util.regex.Pattern;
import javax.annotation.Nullable;
import org.apache.commons.lang3.StringUtils;
+import org.apache.gravitino.job.JobExecutorFactory;
/**
* The configurations of the {@code k8s} job executor, set with the {@code
@@ -33,7 +34,7 @@ import org.apache.commons.lang3.StringUtils;
public class K8sJobExecutorConfigs {
/** The name of the job executor. */
- public static final String K8S_JOB_EXECUTOR_NAME = "k8s";
+ public static final String K8S_JOB_EXECUTOR_NAME =
JobExecutorFactory.K8S_JOB_EXECUTOR_NAME;
/**
* The name of the Kubernetes cluster the jobs run in, which is part of the
job execution id. It
@@ -156,6 +157,20 @@ public class K8sJobExecutorConfigs {
*/
public static final long DEFAULT_SPARK_START_TIMEOUT_MS = 60 * 60 * 1000L;
+ /**
+ * The URI of the jar of the built-in jobs that the Spark driver and
executors can reach. It
+ * replaces the executable of a built-in job template, which is the path of
the jar on the
+ * Gravitino server.
+ */
+ public static final String SPARK_BUILTIN_JOBS_JAR = "spark.builtinJobsJar";
+
+ /**
+ * The default URI of the jar of the built-in jobs, a file in the Spark
image or in a volume
+ * mounted into the Spark pods.
+ */
+ public static final String DEFAULT_SPARK_BUILTIN_JOBS_JAR =
+ "local:///opt/gravitino/jobs/gravitino-jobs.jar";
+
/**
* The prefix of the default Spark configurations of every Spark job. The
job's own Spark
* configurations take precedence.
@@ -189,6 +204,7 @@ public class K8sJobExecutorConfigs {
private final long sparkTtlAfterStopMs;
private final long sparkDriverStartTimeoutMs;
private final long sparkDriverReadyTimeoutMs;
+ private final String sparkBuiltinJobsJar;
private final Map<String, String> sparkConf;
/**
@@ -290,6 +306,14 @@ public class K8sJobExecutorConfigs {
positiveLongValue(configs, SPARK_DRIVER_START_TIMEOUT_MS,
DEFAULT_SPARK_START_TIMEOUT_MS);
this.sparkDriverReadyTimeoutMs =
positiveLongValue(configs, SPARK_DRIVER_READY_TIMEOUT_MS,
DEFAULT_SPARK_START_TIMEOUT_MS);
+ this.sparkBuiltinJobsJar =
+ stringValue(configs, SPARK_BUILTIN_JOBS_JAR,
DEFAULT_SPARK_BUILTIN_JOBS_JAR);
+ Preconditions.checkArgument(
+ K8sJobResourceUtils.isClusterReachable(sparkBuiltinJobsJar),
+ "%s must be a URI the Spark driver and executors can reach, such as
local:// for a file"
+ + " in the image, https://, s3a:// or hdfs://, not a path on the
Gravitino server: %s",
+ SPARK_BUILTIN_JOBS_JAR,
+ sparkBuiltinJobsJar);
Map<String, String> conf = new TreeMap<>();
configs.forEach(
@@ -486,6 +510,15 @@ public class K8sJobExecutorConfigs {
return sparkDriverReadyTimeoutMs;
}
+ /**
+ * Returns the URI of the jar of the built-in jobs that the Spark driver and
executors can reach.
+ *
+ * @return the URI of the built-in jobs jar
+ */
+ public String sparkBuiltinJobsJar() {
+ return sparkBuiltinJobsJar;
+ }
+
/**
* Returns the default Spark configurations of every Spark job.
*
diff --git
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobResourceUtils.java
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobResourceUtils.java
index b9a3a3b4eb..126fa04777 100644
---
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobResourceUtils.java
+++
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobResourceUtils.java
@@ -22,6 +22,7 @@ import com.google.common.collect.ImmutableMap;
import io.fabric8.kubernetes.api.model.HasMetadata;
import io.fabric8.kubernetes.api.model.ObjectMeta;
import java.util.Map;
+import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.commons.lang3.StringUtils;
@@ -43,6 +44,8 @@ public final class K8sJobResourceUtils {
/** The annotation set when Gravitino requests to cancel the job. */
public static final String ANNOTATION_CANCEL_REQUESTED =
"gravitino.apache.org/cancel-requested";
+ private static final Pattern URI_SCHEME =
Pattern.compile("^([A-Za-z][A-Za-z0-9+.-]*):");
+
// A DNS-1123 label, which a namespace name must be.
private static final Pattern NAMESPACE_PATTERN =
Pattern.compile("[a-z0-9]([-a-z0-9]{0,61}[a-z0-9])?");
@@ -59,6 +62,21 @@ public final class K8sJobResourceUtils {
return namespace != null && NAMESPACE_PATTERN.matcher(namespace).matches();
}
+ /**
+ * Returns whether a resource of a job is a URI that the job cluster can
reach. A resource without
+ * a scheme, with a Windows drive letter or with the {@code file} scheme is
a path on the
+ * Gravitino server, which the pods of the job can't reach.
+ *
+ * @param uri the URI of the resource
+ * @return true if the URI has a scheme other than {@code file}
+ */
+ public static boolean isClusterReachable(String uri) {
+ Matcher scheme = URI_SCHEME.matcher(uri.trim());
+ return scheme.find()
+ && scheme.group(1).length() > 1
+ && !"file".equalsIgnoreCase(scheme.group(1));
+ }
+
/**
* Returns the name of the Kubernetes resource of a job.
*
diff --git
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationUtils.java
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationUtils.java
index 73ea8d6c9c..1a10510520 100644
---
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationUtils.java
+++
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationUtils.java
@@ -29,13 +29,15 @@ import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.TreeMap;
-import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.connector.job.JobContext;
+import org.apache.gravitino.job.JobTemplateProvider;
import org.apache.gravitino.job.SparkJobTemplate;
import org.apache.gravitino.job.k8s.K8sJobExecutorConfigs;
import org.apache.gravitino.job.k8s.K8sJobResourceUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Builds the SparkApplication of a Spark job, a custom resource of <a
@@ -44,6 +46,8 @@ import org.apache.gravitino.job.k8s.K8sJobResourceUtils;
*/
public final class SparkApplicationUtils {
+ private static final Logger LOG =
LoggerFactory.getLogger(SparkApplicationUtils.class);
+
/** The SparkApplication resource definition. */
public static final ResourceDefinitionContext SPARK_APPLICATION =
new ResourceDefinitionContext.Builder()
@@ -83,8 +87,6 @@ public final class SparkApplicationUtils {
private static final String SPARK_FILES = "spark.files";
private static final String SPARK_ARCHIVES = "spark.archives";
- private static final Pattern URI_SCHEME =
Pattern.compile("^([A-Za-z][A-Za-z0-9+.-]*):");
-
private static final Pattern URI_USER_INFO =
Pattern.compile("(?<=://)[^/?#@]*@");
private static final Pattern URI_QUERY_OR_FRAGMENT =
Pattern.compile("[?#].*$");
@@ -92,7 +94,9 @@ public final class SparkApplicationUtils {
private SparkApplicationUtils() {}
/**
- * Builds the SparkApplication of a Spark job.
+ * Builds the SparkApplication of a Spark job. A built-in job, whose
executable is the jobs jar on
+ * the Gravitino server, runs the jar of {@link
K8sJobExecutorConfigs#SPARK_BUILTIN_JOBS_JAR}
+ * instead.
*
* @param context the context of the job run
* @param template the runtime job template, whose resources are the URIs
given in the template
@@ -111,7 +115,8 @@ public final class SparkApplicationUtils {
template.name(), template.environments().keySet()));
}
- checkClusterReachable(template.executable());
+ String executable = getExecutableUri(context, template, configs);
+ checkClusterReachable(executable);
template.jars().forEach(SparkApplicationUtils::checkClusterReachable);
template.files().forEach(SparkApplicationUtils::checkClusterReachable);
template.archives().forEach(SparkApplicationUtils::checkClusterReachable);
@@ -161,7 +166,7 @@ public final class SparkApplicationUtils {
spec.put("mainClass", template.className());
}
// The operator takes the main resource of a Python application from
pyFiles.
- spec.put(isPythonFile(template.executable()) ? "pyFiles" : "jars",
template.executable());
+ spec.put(isPythonFile(executable) ? "pyFiles" : "jars", executable);
spec.put("driverArgs", new ArrayList<>(template.arguments()));
spec.put("sparkConf", sparkConf);
spec.put("runtimeVersions", ImmutableMap.of("sparkVersion",
configs.sparkVersion()));
@@ -197,6 +202,28 @@ public final class SparkApplicationUtils {
return app;
}
+ private static String getExecutableUri(
+ JobContext context, SparkJobTemplate template, K8sJobExecutorConfigs
configs) {
+ // The executable of a built-in job template is the path of the jobs jar
on the Gravitino
+ // server, which is what the local job executor runs. An executable that
is already reachable
+ // from the cluster is kept as it is.
+ if (template.name().startsWith(JobTemplateProvider.BUILTIN_NAME_PREFIX)
+ && !K8sJobResourceUtils.isClusterReachable(template.executable())) {
+ // The job records the executable of the template, so log the jar that
actually runs. It must
+ // be the same version as the jobs jar of the server, which the template
comes from.
+ LOG.info(
+ "Job {} of built-in job template {} runs the jobs jar {} set by {}
instead of {} on the"
+ + " Gravitino server",
+ context.jobId(),
+ template.name(),
+ configs.sparkBuiltinJobsJar(),
+ K8sJobExecutorConfigs.SPARK_BUILTIN_JOBS_JAR,
+ template.executable());
+ return configs.sparkBuiltinJobsJar();
+ }
+ return template.executable();
+ }
+
/**
* Checks that a resource of the job, or each of them if it is a
comma-separated list as Spark
* accepts, is a URI that the cluster can reach rather than a path on the
Gravitino server.
@@ -208,11 +235,7 @@ public final class SparkApplicationUtils {
if (uris.length > 1 && StringUtils.isBlank(uri)) {
continue;
}
- Matcher scheme = URI_SCHEME.matcher(uri.trim());
- // No scheme, a Windows drive letter, or a file URI.
- if (!scheme.find()
- || scheme.group(1).length() == 1
- || "file".equalsIgnoreCase(scheme.group(1))) {
+ if (!K8sJobResourceUtils.isClusterReachable(uri)) {
throw new IllegalArgumentException(
String.format(
"Resource %s of the job is a path on the Gravitino server,
which the Spark job on"
diff --git
a/plugins/k8s-job-executor/src/shadedJarTest/java/org/apache/gravitino/job/k8s/TestK8sJobExecutorShadedJar.java
b/plugins/k8s-job-executor/src/shadedJarTest/java/org/apache/gravitino/job/k8s/TestK8sJobExecutorShadedJar.java
new file mode 100644
index 0000000000..023f930d91
--- /dev/null
+++
b/plugins/k8s-job-executor/src/shadedJarTest/java/org/apache/gravitino/job/k8s/TestK8sJobExecutorShadedJar.java
@@ -0,0 +1,222 @@
+/*
+ * 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.gravitino.job.k8s;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.io.ByteStreams;
+import com.sun.net.httpserver.HttpServer;
+import java.io.File;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.URISyntaxException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Enumeration;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.jar.JarEntry;
+import java.util.jar.JarFile;
+import org.apache.gravitino.Config;
+import org.apache.gravitino.Configs;
+import org.apache.gravitino.connector.job.JobContext;
+import org.apache.gravitino.connector.job.JobExecutor;
+import org.apache.gravitino.job.JobExecutorFactory;
+import org.apache.gravitino.job.SparkJobTemplate;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * Tests the shaded jar of the k8s job executor the way the Gravitino server
loads it: the jar is on
+ * the classpath of the server, next to the server's own Jackson, and neither
the classes nor the
+ * dependencies of the module are. The tests refer to the job executor by name
only.
+ */
+public class TestK8sJobExecutorShadedJar {
+
+ private static final String SHADED_PACKAGE =
"org.apache.gravitino.k8s.shaded.";
+
+ private static final String SPARK_APPLICATIONS_PATH =
+ "/apis/spark.apache.org/v1/namespaces/default/sparkapplications";
+
+ private static File shadedJar;
+
+ @BeforeAll
+ public static void setUp() throws IOException {
+ String path = System.getProperty("gravitino.k8s.shadedJar");
+ Assertions.assertNotNull(path, "Run the tests with the shadedJarTest
Gradle task");
+ shadedJar = new File(path).getCanonicalFile();
+ Assertions.assertTrue(shadedJar.isFile(), shadedJar + " doesn't exist");
+ }
+
+ @Test
+ public void testJacksonIsRelocated() throws Exception {
+ Assertions.assertEquals(
+ shadedJar,
getJarOf(Class.forName(JobExecutorFactory.K8S_JOB_EXECUTOR_CLASS_NAME)));
+ // fabric8 uses the Jackson of the shaded jar, the server keeps its own.
+ Assertions.assertEquals(
+ shadedJar, getJarOf(Class.forName(SHADED_PACKAGE +
ObjectMapper.class.getName())));
+ Assertions.assertNotEquals(shadedJar, getJarOf(ObjectMapper.class));
+
+ try (JarFile jar = new JarFile(shadedJar)) {
+ Set<String> notRelocated = new TreeSet<>();
+ for (JarEntry entry : Collections.list(jar.entries())) {
+ String name = entry.getName();
+ if (name.startsWith("com/fasterxml/")
+ || name.startsWith("org/yaml/")
+ || name.startsWith("org/snakeyaml/")) {
+ notRelocated.add(name);
+ }
+ }
+ Assertions.assertTrue(notRelocated.isEmpty(), "Not relocated: " +
notRelocated);
+ }
+ }
+
+ @Test
+ public void testNoClassOfServerClasspathIsBundled() throws IOException {
+ // A class that is both in the shaded jar and in another jar of the server
is loaded from
+ // whichever comes first on the classpath, so the shaded jar must not
bundle any. The classpath
+ // here is the runtime classpath of :server, the libs directory of the
distribution also gets
+ // the jars of the other plugins (idp-basic, tms and its dependencies),
which this doesn't
+ // cover.
+ Set<String> serverClasses = new TreeSet<>();
+ for (String path :
System.getProperty("java.class.path").split(File.pathSeparator)) {
+ File file = new File(path).getCanonicalFile();
+ if (file.isFile() && file.getName().endsWith(".jar") &&
!file.equals(shadedJar)) {
+ serverClasses.addAll(listClasses(file));
+ }
+ }
+ Assertions.assertFalse(serverClasses.isEmpty());
+
+ Set<String> bundled = new TreeSet<>(listClasses(shadedJar));
+ Assertions.assertFalse(bundled.isEmpty());
+ bundled.retainAll(serverClasses);
+ Assertions.assertTrue(
+ bundled.isEmpty(), "Classes also on the classpath of the server: " +
bundled);
+ }
+
+ @Test
+ public void testCreateJobExecutorAndSubmitJob(@TempDir Path jobDir) throws
Exception {
+ List<String> requests = Collections.synchronizedList(new ArrayList<>());
+ List<String> bodies = Collections.synchronizedList(new ArrayList<>());
+ // A minimal API server, which creates whatever is posted to it.
+ HttpServer apiServer =
+ HttpServer.create(new
InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 0);
+ apiServer.createContext(
+ "/",
+ exchange -> {
+ byte[] body = ByteStreams.toByteArray(exchange.getRequestBody());
+ requests.add(exchange.getRequestMethod() + " " +
exchange.getRequestURI().getPath());
+ boolean created =
+ "POST".equals(exchange.getRequestMethod())
+ &&
SPARK_APPLICATIONS_PATH.equals(exchange.getRequestURI().getPath());
+ if (created) {
+ bodies.add(new String(body, StandardCharsets.UTF_8));
+ exchange.getResponseHeaders().add("Content-Type",
"application/json");
+ exchange.sendResponseHeaders(HttpURLConnection.HTTP_CREATED,
body.length);
+ } else {
+ body = new byte[0];
+ exchange.sendResponseHeaders(HttpURLConnection.HTTP_NOT_FOUND, -1);
+ }
+ try (OutputStream response = exchange.getResponseBody()) {
+ response.write(body);
+ }
+ });
+ apiServer.start();
+
+ try {
+ Config config = new Config(false) {};
+ config.set(Configs.JOB_EXECUTOR,
JobExecutorFactory.K8S_JOB_EXECUTOR_NAME);
+ config.loadFromMap(
+ ImmutableMap.of(
+ "gravitino.jobExecutor.k8s.masterUrl",
+ "http://127.0.0.1:"; + apiServer.getAddress().getPort(),
+ "gravitino.jobExecutor.k8s.namespace",
+ "default",
+ "gravitino.jobExecutor.k8s.spark.image",
+ "gravitino-spark:3.5.9",
+ "gravitino.jobExecutor.k8s.spark.sparkVersion",
+ "3.5.9",
+ "gravitino.jobExecutor.k8s.spark.conf.spark.executor.instances",
+ "2"),
+ key -> true);
+ SparkJobTemplate template =
+ SparkJobTemplate.builder()
+ .withName("builtin-sparkpi")
+ .withExecutable("/gravitino/auxlib/gravitino-jobs.jar")
+
.withClassName("org.apache.gravitino.maintenance.jobs.spark.SparkPiJob")
+ .withConfigs(ImmutableMap.of("spark.master", "local[*]"))
+ .build();
+
+ // The factory finds the built-in k8s job executor by its class name.
+ try (JobExecutor jobExecutor = JobExecutorFactory.create(config)) {
+ Assertions.assertEquals(
+ JobExecutorFactory.K8S_JOB_EXECUTOR_CLASS_NAME,
jobExecutor.getClass().getName());
+
+ String jobExecutionId =
+ jobExecutor.submitJob(new JobContext(1L, "metalake",
jobDir.toFile()), template);
+ Assertions.assertEquals("default/sparkapp/default/gravitino-job-1",
jobExecutionId);
+ }
+ } finally {
+ apiServer.stop(0);
+ }
+
+ Assertions.assertEquals(1, bodies.size(), "Requests: " + requests);
+ // Written by the relocated Jackson of the shaded jar, read by the Jackson
of the server.
+ JsonNode sparkApplication = new ObjectMapper().readTree(bodies.get(0));
+ Assertions.assertEquals("SparkApplication",
sparkApplication.at("/kind").asText());
+ Assertions.assertEquals("gravitino-job-1",
sparkApplication.at("/metadata/name").asText());
+ Assertions.assertEquals(
+ "local:///opt/gravitino/jobs/gravitino-jobs.jar",
+ sparkApplication.at("/spec/jars").asText());
+ JsonNode sparkConf = sparkApplication.at("/spec/sparkConf");
+ Assertions.assertEquals("2",
sparkConf.get("spark.executor.instances").asText());
+ Assertions.assertFalse(sparkConf.has("spark.master"));
+ }
+
+ private static File getJarOf(Class<?> clazz) throws IOException,
URISyntaxException {
+ return new
File(clazz.getProtectionDomain().getCodeSource().getLocation().toURI())
+ .getCanonicalFile();
+ }
+
+ private static Set<String> listClasses(File jarFile) throws IOException {
+ Set<String> classes = new TreeSet<>();
+ try (JarFile jar = new JarFile(jarFile)) {
+ Enumeration<JarEntry> entries = jar.entries();
+ while (entries.hasMoreElements()) {
+ String name = entries.nextElement().getName();
+ // Every jar may have its own module descriptor and multi-release
classes.
+ if (name.endsWith(".class")
+ && !name.startsWith("META-INF/")
+ && !name.endsWith("module-info.class")) {
+ classes.add(name);
+ }
+ }
+ }
+ return classes;
+ }
+}
diff --git
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobExecutor.java
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobExecutor.java
index dd5bd4125c..a5fdc9c92b 100644
---
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobExecutor.java
+++
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobExecutor.java
@@ -57,6 +57,7 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.gravitino.connector.job.JobContext;
import org.apache.gravitino.connector.job.JobExecutionInfo;
import org.apache.gravitino.exceptions.NoSuchJobException;
+import org.apache.gravitino.job.JobExecutorFactory;
import org.apache.gravitino.job.JobHandle;
import org.apache.gravitino.job.ShellJobTemplate;
import org.apache.gravitino.job.SparkJobTemplate;
@@ -131,6 +132,15 @@ public class TestK8sJobExecutor {
server.destroy();
}
+ @Test
+ public void testRegisteredInJobExecutorFactory() {
+ // The factory in core can only refer to the job executor by name, so
check the names here.
+ Assertions.assertEquals(
+ JobExecutorFactory.K8S_JOB_EXECUTOR_CLASS_NAME,
K8sJobExecutor.class.getName());
+ Assertions.assertEquals(
+ JobExecutorFactory.K8S_JOB_EXECUTOR_NAME,
K8sJobExecutorConfigs.K8S_JOB_EXECUTOR_NAME);
+ }
+
@Test
public void testSubmitAndQuery() {
String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
diff --git
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobExecutorConfigs.java
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobExecutorConfigs.java
index bf18938f7c..afd7fd762c 100644
---
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobExecutorConfigs.java
+++
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobExecutorConfigs.java
@@ -55,6 +55,8 @@ public class TestK8sJobExecutorConfigs {
Assertions.assertEquals(604_800_000L, configs.sparkTtlAfterStopMs());
Assertions.assertEquals(3_600_000L, configs.sparkDriverStartTimeoutMs());
Assertions.assertEquals(3_600_000L, configs.sparkDriverReadyTimeoutMs());
+ Assertions.assertEquals(
+ "local:///opt/gravitino/jobs/gravitino-jobs.jar",
configs.sparkBuiltinJobsJar());
Assertions.assertTrue(configs.sparkConf().isEmpty());
}
@@ -66,6 +68,7 @@ public class TestK8sJobExecutorConfigs {
map.put(K8sJobExecutorConfigs.NAME_PREFIX, "gvt-a");
map.put(K8sJobExecutorConfigs.STATUS_CACHE_TTL_MS, "0");
map.put(K8sJobExecutorConfigs.SPARK_TTL_AFTER_STOP_MS, "1000");
+ map.put(K8sJobExecutorConfigs.SPARK_BUILTIN_JOBS_JAR, "
https://repo/gravitino-jobs.jar ");
map.put("spark.conf.spark.executor.instances", "2");
map.put("spark.conf.", "ignored");
@@ -76,6 +79,7 @@ public class TestK8sJobExecutorConfigs {
Assertions.assertEquals("gvt-a", configs.namePrefix());
Assertions.assertEquals(0L, configs.statusCacheTtlMs());
Assertions.assertEquals(1000L, configs.sparkTtlAfterStopMs());
+ Assertions.assertEquals("https://repo/gravitino-jobs.jar";,
configs.sparkBuiltinJobsJar());
Assertions.assertEquals(ImmutableMap.of("spark.executor.instances", "2"),
configs.sparkConf());
}
@@ -185,6 +189,10 @@ public class TestK8sJobExecutorConfigs {
assertInvalid(K8sJobExecutorConfigs.REQUEST_RETRIES, "-1");
assertInvalid(K8sJobExecutorConfigs.REQUEST_RETRIES, "many");
assertInvalid(K8sJobExecutorConfigs.SPARK_DRIVER_START_TIMEOUT_MS, "1h");
+ // The built-in jobs jar must be reachable from the cluster, not a path on
the server.
+ assertInvalid(K8sJobExecutorConfigs.SPARK_BUILTIN_JOBS_JAR,
"/opt/gravitino/jobs.jar");
+ assertInvalid(K8sJobExecutorConfigs.SPARK_BUILTIN_JOBS_JAR,
"file:///opt/gravitino/jobs.jar");
+ assertInvalid(K8sJobExecutorConfigs.SPARK_BUILTIN_JOBS_JAR,
"C:\\gravitino\\jobs.jar");
}
private static void assertInvalid(String... keyValues) {
diff --git
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobResourceUtils.java
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobResourceUtils.java
index 74104fc084..ad4a999938 100644
---
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobResourceUtils.java
+++
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobResourceUtils.java
@@ -96,6 +96,26 @@ public class TestK8sJobResourceUtils {
}
}
+ @Test
+ public void testIsClusterReachable() {
+ for (String uri :
+ new String[] {
+ "local:///opt/app.jar",
+ "https://repo/app.jar";,
+ "s3a://bucket/app.jar",
+ "hdfs://nn/app.jar",
+ " s3a://bucket/app jar "
+ }) {
+ Assertions.assertTrue(K8sJobResourceUtils.isClusterReachable(uri), uri);
+ }
+
+ // Paths on the Gravitino server.
+ for (String uri :
+ new String[] {"/opt/app.jar", "app.jar", "file:///opt/app.jar",
"FILE:/a", "C:\\app.jar"}) {
+ Assertions.assertFalse(K8sJobResourceUtils.isClusterReachable(uri), uri);
+ }
+ }
+
private static ConfigMap resource(String name, long jobId) {
return new ConfigMapBuilder()
.withNewMetadata()
diff --git
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/spark/TestSparkApplicationUtils.java
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/spark/TestSparkApplicationUtils.java
index e9fb7b8a09..fdb6b2e65b 100644
---
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/spark/TestSparkApplicationUtils.java
+++
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/spark/TestSparkApplicationUtils.java
@@ -458,6 +458,61 @@ public class TestSparkApplicationUtils {
Assertions.assertFalse(e.getMessage().contains("abc"), e.getMessage());
}
+ @Test
+ public void testBuiltinJobRunsBuiltinJobsJar() {
+ // The executable of a built-in job template is the jobs jar on the
Gravitino server.
+ for (String uri :
+ new String[] {
+ "/gravitino/auxlib/gravitino-jobs-2.0.0.jar",
"file:///gravitino/auxlib/jobs.jar"
+ }) {
+ SparkJobTemplate template =
+
newTemplateBuilder().withName("builtin-sparkpi").withExecutable(uri).build();
+ Assertions.assertEquals(
+ "local:///opt/gravitino/jobs/gravitino-jobs.jar", getSpec(template,
configs).get("jars"));
+ Assertions.assertEquals(
+ "s3a://bucket/gravitino-jobs.jar",
+ getSpec(
+ template,
+ newConfigs(
+ ImmutableMap.of(
+ K8sJobExecutorConfigs.SPARK_BUILTIN_JOBS_JAR,
+ "s3a://bucket/gravitino-jobs.jar")))
+ .get("jars"));
+ }
+ }
+
+ @Test
+ public void testBuiltinJobKeepsClusterReachableExecutable() {
+ SparkJobTemplate template =
+ newTemplateBuilder()
+ .withName("builtin-sparkpi")
+ .withExecutable("https://repo/gravitino-jobs.jar";)
+ .build();
+ Assertions.assertEquals(
+ "https://repo/gravitino-jobs.jar";, getSpec(template,
configs).get("jars"));
+ }
+
+ @Test
+ public void testBuiltinJobsJarOnlyReplacesExecutableOfBuiltinJobs() {
+ // Not a built-in job: "builtin-" must be the prefix of the name.
+ SparkJobTemplate userJob =
+
newTemplateBuilder().withName("my-builtin-job").withExecutable("/opt/app.jar").build();
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> SparkApplicationUtils.buildSparkApplication(CONTEXT, userJob,
configs));
+
+ // The other resources of a built-in job are still checked.
+ SparkJobTemplate builtinJob =
+ newTemplateBuilder()
+ .withName("builtin-sparkpi")
+ .withExecutable("/gravitino/auxlib/gravitino-jobs-2.0.0.jar")
+ .withJars(ImmutableList.of("/gravitino/auxlib/other.jar"))
+ .build();
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> SparkApplicationUtils.buildSparkApplication(CONTEXT, builtinJob,
configs));
+ }
+
@Test
public void testRejectEnvironments() {
SparkJobTemplate template =
@@ -482,6 +537,15 @@ public class TestSparkApplicationUtils {
return new K8sJobExecutorConfigs(map);
}
+ @SuppressWarnings("unchecked")
+ private static Map<String, Object> getSpec(
+ SparkJobTemplate template, K8sJobExecutorConfigs configs) {
+ return (Map<String, Object>)
+ SparkApplicationUtils.buildSparkApplication(CONTEXT, template, configs)
+ .getAdditionalProperties()
+ .get("spec");
+ }
+
private static SparkJobTemplate.Builder newTemplateBuilder() {
return SparkJobTemplate.builder()
.withName("user-job")
