jerryshao commented on code in PR #13982:
URL: https://github.com/apache/gravitino/pull/13982#discussion_r4332457923
##########
plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationUtils.java:
##########
@@ -216,6 +260,110 @@ public static GenericKubernetesResource
buildSparkApplication(
return app;
}
+ /**
+ * Returns the Spark configurations that a job sets itself, which override
the default Spark
+ * configurations of the job executor.
+ *
+ * <p>They are the configurations of the job template, and for a built-in
job also those of its
+ * {@code spark_conf} parameter, which override them. The driver of a
built-in job applies its
+ * {@code spark_conf} only once it starts, too late for the configurations
that spark-submit
+ * reads, such as {@code spark.jars}, so they are set in the
SparkApplication as well. {@code
+ * spark.master} and {@code spark.submit.deployMode} there are ignored, as
the driver also ignores
+ * them, and so are the {@code spark.kubernetes.*} configurations, which
only the job executor
+ * configurations and job templates may set: whoever can run a built-in job
sets its {@code
+ * spark_conf}. A blank value leaves a configuration unset, as it is what an
optional template
+ * parameter becomes, so it doesn't override the default of the job executor.
+ */
+ private static Map<String, String> getJobSparkConf(
+ JobContext context, SparkJobTemplate template) {
+ Map<String, String> jobSparkConf = new TreeMap<>();
+ template
+ .configs()
+ .forEach(
+ (key, value) -> {
+ if (StringUtils.isNotBlank(value)) {
+ jobSparkConf.put(key, value);
+ }
+ });
+ if (!template.name().startsWith(JobTemplateProvider.BUILTIN_NAME_PREFIX)) {
+ return jobSparkConf;
+ }
+
+ getBuiltinSparkConf(template)
+ .forEach(
+ (key, value) -> {
+ if (StringUtils.isBlank(value)) {
+ return;
+ }
+ if (SUBMIT_DECIDED_CONFIGS.contains(key)) {
+ LOG.warn(
+ "Ignored {}={} in spark_conf of job {} of built-in job
template {}, as the"
+ + " Spark operator decides it",
+ key,
+ value,
+ context.jobId(),
+ template.name());
+ } else if (key.startsWith(SPARK_KUBERNETES_PREFIX)) {
+ // The value isn't logged, as it may name a Secret.
+ LOG.warn(
+ "Ignored {} in spark_conf of job {} of built-in job
template {}, as only the"
+ + " k8s job executor configurations and job templates
may set {}*",
+ key,
+ context.jobId(),
+ template.name(),
+ SPARK_KUBERNETES_PREFIX);
+ } else {
+ jobSparkConf.put(key, value);
Review Comment:
[Important] A non-blank `spark.jars` (or `spark.files` / `spark.archives`)
in `spark_conf` replaces the executor default instead of appending to it, which
breaks a workflow this PR's own docs recommend.
`jobSparkConf.put(key, value)` here lands in `sparkConf` via `putAll` at
:168, overwriting whatever `gravitino.jobExecutor.k8s.spark.conf.spark.jars`
put there from `new TreeMap<>(configs.sparkConf())` at :158. `appendList`
(:447-453) only appends `template.jars()`, never the executor default, so the
default is simply gone.
That collides with the guidance added in this same PR.
`docs/table-maintenance-service/optimizer-cli-reference.md:184-187` tells users
to put a JDBC driver in `spark.jars` in `--spark-conf` under the k8s executor,
while `docs/table-maintenance-service/optimizer-configuration.md:112-117` and
`optimizer-cli-reference.md:256-262` offer the Iceberg runtime either in the
image or via `spark.jars`. A deployment that carries the Iceberg runtime in
`gravitino.jobExecutor.k8s.spark.conf.spark.jars` (which
`testSparkConfOfBuiltinJobKeepsExecutorDefaults` treats as the normal setup)
then loses it the moment a user adds one unrelated jar, and the job dies in the
driver with the "Missing Iceberg Spark session extensions" error -- from a
`spark_conf` that never mentioned Iceberg.
Suggest treating the three resource lists as additive: for `spark.jars`,
`spark.files` and `spark.archives`, merge the executor default and the job
value (comma-joined, as `appendList` already does) rather than replacing, or at
minimum say in the docs that `spark.jars` in `spark_conf` replaces the executor
default so the Iceberg runtime has to be repeated.
Verified by: read SparkApplicationUtils.java:158-219 and :447-453 in full on
29dc9e0, traced the single write path for `spark.jars`; cross-read the three
doc files changed by this PR; confirmed no special case for resource keys
exists in `getJobSparkConf`.
##########
plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationUtils.java:
##########
@@ -136,7 +164,8 @@ public static GenericKubernetesResource
buildSparkApplication(
sparkConf.put(SPARK_DRIVER_ENV_PREFIX + name, value);
sparkConf.put(SPARK_EXECUTOR_ENV_PREFIX + name, value);
});
- sparkConf.putAll(template.configs());
+ Map<String, String> jobSparkConf = getJobSparkConf(context, template);
+ sparkConf.putAll(jobSparkConf);
Review Comment:
[Nit] `spark.executorEnv.*` arriving through `spark_conf` skips the
`checkEnvironmentName` validation that the same keys get from
`template.environments()`.
The loop just above (:159-166) validates every environment variable name
before writing `spark.kubernetes.driverEnv.<name>` and
`spark.executorEnv.<name>`, because Kubernetes rejects some names and the pods
would then fail to start. `jobSparkConf` merged here can carry
`spark.executorEnv.<anything>` straight from a built-in job's `spark_conf`:
`spark.kubernetes.driverEnv.*` is blocked by the `spark.kubernetes.` prefix
rule at :306, but `spark.executorEnv.*` is not, so it reaches `spec.sparkConf`
unvalidated and a name like `spark.executorEnv.BAD NAME` turns a clear
submit-time rejection into an opaque executor-pod creation failure inside the
driver.
Either run `checkEnvironmentName` over the `spark.executorEnv.` suffixes of
`jobSparkConf`, or note in the Javadoc that `spark_conf` can set executor env
vars without that check.
Verified by: read SparkApplicationUtils.java:159-168 and the
`getJobSparkConf` filter at :277-319 on 29dc9e0; the only callers of
`checkEnvironmentName` are in the `template.environments()` loop, and
`ENVIRONMENT_NAME_PATTERN` (:118-120) is applied nowhere else.
##########
maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/command/SubmitUpdateStatsJobCommand.java:
##########
@@ -333,12 +331,10 @@ private static Map<String, String>
resolveTemplateSparkPlaceholders(
Map<String, String> placeholders = new LinkedHashMap<>();
String catalogPrefix = SPARK_SQL_CATALOG_PREFIX + catalogName;
- placeholders.put(
- "spark_master",
- firstNonBlank(
- submitterConfigs.get("spark_master"),
- sparkConfigs.get(SPARK_MASTER_KEY),
- DEFAULT_SPARK_MASTER));
+ // An empty master leaves it to the job executor: spark-submit defaults to
local[*], and the
+ // k8s job executor lets the operator decide it. spark.master in
spark_conf isn't used, as the
+ // job ignores it there too.
+ placeholders.put("spark_master",
firstNonBlank(submitterConfigs.get("spark_master")));
Review Comment:
[Important] Dropping the `spark.master` fallback silently changes where an
existing job runs; the CLI should say so.
Before this commit `spark_master` fell back to
`sparkConfigs.get("spark.master")`. Now it comes only from `submitterConfigs`,
while `spark.master` stays in the `spark_conf` JSON forwarded at :154 -- where
both new filters discard it: `SparkApplicationUtils.getJobSparkConf` ignores it
as a submit-decided config, and `IcebergJobUtils.filterCustomSparkConfigs`
drops it because spark-submit has always set `spark.master` (it defaults to
`local[*]`).
So an existing config that set the master only in `spark_conf`, e.g.
`gravitino.optimizer.jobSubmitterConfig.spark_conf =
{"spark.master":"spark://host:7077"}`, now runs the job on spark-submit's
`local[*]` under the local job executor instead of the standalone cluster. The
only trace is a WARN in the driver log; `validateSparkConfigs` (:292-303) does
not look at `spark.master`, and the CLI prints nothing. The PR body lists this
under "What changes" but not under user-facing changes.
Suggest rejecting or at least warning at submit time, e.g. in
`validateSparkConfigs`: if `sparkConfigs.containsKey("spark.master")`, print
one line telling the user to move it to
`gravitino.optimizer.jobSubmitterConfig.spark_master` / `--spark-master`, since
the value is now ignored. A hard `Preconditions.checkArgument` would be even
clearer and matches how the k8s executor rejects a non-k8s master outright.
Verified by: read SubmitUpdateStatsJobCommand.java:80-160 and :285-345 on
29dc9e0; confirmed `firstNonBlank` returns "" (:379-386) so the placeholder is
empty, not null; confirmed the forwarded `spark_conf` still carries the key at
:154; traced both consuming filters (SparkApplicationUtils.java:298-305,
IcebergJobUtils.java:243-252).
##########
plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationUtils.java:
##########
@@ -18,8 +18,12 @@
*/
package org.apache.gravitino.job.k8s.spark;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
Review Comment:
[Nit] Jackson is now used in this module's main source but is not a declared
dependency of it.
`plugins/k8s-job-executor/build.gradle.kts` declares Jackson only for the
shaded-jar test source set
(`shadedJarTestImplementation(libs.jackson.databind)`, :67). The main source
gets it transitively from `implementation(libs.fabric8.kubernetes.client)`
(:46), with the comment at :45 framing Jackson as fabric8's dependency rather
than this module's. It compiles and relocates correctly today
(`relocate("com.fasterxml.jackson", ...)` at :94), but a fabric8 upgrade that
changes or shades its own Jackson would break this file for a reason unrelated
to Kubernetes.
Suggest adding an explicit `implementation(libs.jackson.databind)` so the
usage is declared where it is used.
Verified by: read plugins/k8s-job-executor/build.gradle.kts:33-70 and :88-96
on 29dc9e0; grepped the module for `jackson` and found the only declaration in
the `shadedJarTest` configuration.
##########
maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergJobUtils.java:
##########
@@ -173,16 +202,67 @@ public static Map<String, String> parseArguments(
* @throws IllegalArgumentException if JSON parsing fails
*/
public static Map<String, String> parseCustomSparkConfigs(String
sparkConfJson) {
- return IcebergSparkConfigUtils.parseFlatJsonMap(sparkConfJson,
"spark-conf");
+ return filterCustomSparkConfigs(
+ IcebergSparkConfigUtils.parseFlatJsonMap(sparkConfJson, "spark-conf"));
+ }
+
+ /**
+ * Filter the custom Spark configurations of a job, which the driver applies
when it creates the
+ * SparkSession, against the configurations spark-submit has already set for
the driver.
+ *
+ * <p>{@code spark.master} and {@code spark.submit.deployMode} are dropped
if spark-submit has set
+ * them, as the job would otherwise run with another master than it was
submitted to. The {@code
+ * spark.kubernetes.*} configurations are dropped, as whoever runs a job
sets them here, while the
+ * driver creates the executor pods from them. A blank value leaves a
configuration unset.
+ *
+ * @param configs the custom Spark configurations
+ * @return the custom Spark configurations to apply
+ */
+ public static Map<String, String> filterCustomSparkConfigs(Map<String,
String> configs) {
+ return filterCustomSparkConfigs(configs, new SparkConf());
Review Comment:
[Nit] `new SparkConf()` makes this filter depend on ambient JVM system
properties, which the existing job tests now rely on.
`new SparkConf()` loads every `spark.*` system property, so whether
`spark.master` from `spark_conf` is kept or dropped depends on the JVM the job
runs in. That is correct under spark-submit, but it makes
`TestIcebergRemoveOrphanFilesJobMain:108`,
`TestIcebergRewriteManifestsJobWithSpark:183` and
`TestIcebergUpdateStatsJobWithSpark:761` -- which pass
`{"spark.master":"local[1]"}` or `local[2]` in `--spark-conf` and expect it
applied -- quietly load-bearing on no `spark.master` system property existing
in the test JVM. I grepped `maintenance/jobs/src/test` and found no
`System.setProperty`, so they pass today, but anything that later sets that
property (a shared forked JVM, a future Spark test fixture) turns them into a
`SparkException: A master URL must be set` with no obvious cause.
Consider taking the submitted conf from the caller (the overload already
exists and `testFilterCustomSparkConfigsKeepsMasterNotSubmitted` uses `new
SparkConf(false)`), or adding a comment on this line recording that the no-arg
constructor reads system properties on purpose.
Verified by: read IcebergJobUtils.java:205-257 on 29dc9e0; grepped
`spark.master` and `System.setProperty` across `maintenance/jobs/src/test`;
read the three tests' `--spark-conf` payloads.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
