jerryshao commented on code in PR #13134:
URL: https://github.com/apache/gravitino/pull/13134#discussion_r4004967675
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -947,17 +945,91 @@ static String replacePlaceholder(String inputString,
Map<String, String> replace
String key = matcher.group(1);
String replacement = replacements.get(key);
if (replacement != null) {
- matcher.appendReplacement(result, replacement);
+ matcher.appendReplacement(result,
Matcher.quoteReplacement(replacement));
} else {
// If no replacement is found, keep the placeholder as is
- matcher.appendReplacement(result, matcher.group(0));
+ matcher.appendReplacement(result,
Matcher.quoteReplacement(matcher.group(0)));
}
}
matcher.appendTail(result);
return result.toString();
}
+ /**
+ * Drop blank / unresolved optional template arguments after placeholder
substitution.
+ *
+ * <p>Built-in templates always list optional flags as {@code --flag} +
{@code {{placeholder}}}.
+ * When the job conf omits that key or supplies an empty value, leaving the
flag in the command
+ * produces dangling arguments such as {@code --updater-options
--spark-conf}. This method
+ * removes:
+ *
+ * <ul>
+ * <li>blank tokens
+ * <li>tokens that are still an entire unresolved {@code {{placeholder}}}
+ * <li>{@code --flag} pairs whose following value is blank or an
unresolved placeholder
+ * </ul>
+ *
+ * @param arguments arguments after {@link #replacePlaceholder(String, Map)}
+ * @return compacted argument list suitable for process execution
+ */
+ @VisibleForTesting
+ static List<String> omitEmptyArguments(List<String> arguments) {
+ if (arguments == null || arguments.isEmpty()) {
+ return arguments;
+ }
+
+ List<String> result = new ArrayList<>(arguments.size());
+ for (int i = 0; i < arguments.size(); i++) {
+ String arg = arguments.get(i);
+ if (isUnresolvedOptionalValue(arg)) {
Review Comment:
**correctness (high)**: `omitEmptyArguments` applies
`isUnresolvedOptionalValue` to *every* argument position uniformly, not just
the value half of a `--flag`/value pair — so it silently drops a
bare/positional argument that never had a `--flag` prefix.
This is reachable today, not just theoretical: `SparkPiJob.jobTemplate()`
builds its arguments as `Collections.singletonList("{{slices}}")` (a single
bare positional argument), and the existing test `TestJobManager.java:675`
builds a `ShellJobTemplate` the same way with
`withArguments(Lists.newArrayList("{{greeting}}"))`. If the caller omits that
jobConf key, the argument used to arrive as a visibly-broken literal
`"{{slices}}"`/`"{{greeting}}"` (loud, diagnosable) — now it's deleted
entirely, silently shrinking `args.length` for any executable expecting a fixed
argument count.
---
**correctness**: Separately, the flag/value pairing heuristic a few lines
below (`arg.startsWith("--") && ... isUnresolvedOptionalValue(next)`) assumes
any `--flag` token immediately followed by a blank/unresolved token is that
flag's own droppable value — with no way to tell that apart from an unrelated,
independent optional-flag placeholder that just happens to sit next in the
list. `omitEmptyArguments(List.of("--verbose", "{{unset_flag}}"))` returns `[]`
instead of `["--verbose"]`. It doesn't trigger in the three shipped Iceberg
templates only because every fixed flag there happens to be followed by its own
value slot first — but it's a real hazard for any other
`SparkJobTemplate`/`ShellJobTemplate` built the same way.
---
**conventions**: Also, this method explicitly guards `arguments == null` and
can return `null`, so null is a real in-contract input/output, but the
parameter isn't annotated `@Nullable` per CLAUDE.md's "Use `@Nullable`
annotations" rule (this is fresh logic — `JobManager.java` has zero
pre-existing `@Nullable` usages to inherit from).
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -846,15 +848,11 @@ public static JobTemplate createRuntimeJobTemplate(
replacePlaceholder(content.executable(), jobConf), stagingDir,
TIMEOUT_IN_MS);
List<String> args =
- content.arguments().stream()
- .map(arg -> replacePlaceholder(arg, jobConf))
- .collect(Collectors.toList());
- Map<String, String> environments =
- content.environments().entrySet().stream()
- .collect(
- Collectors.toMap(
- entry -> replacePlaceholder(entry.getKey(), jobConf),
- entry -> replacePlaceholder(entry.getValue(), jobConf)));
+ omitEmptyArguments(
+ content.arguments().stream()
+ .map(arg -> replacePlaceholder(arg, jobConf))
+ .collect(Collectors.toList()));
+ Map<String, String> environments =
omitUnresolvedTemplateMap(content.environments(), jobConf);
Review Comment:
**correctness**: `arguments`/`environments` get the new blank/placeholder
filtering right here, but `SparkJobTemplate.configs` (built a few lines below
at line ~910-915, the `--conf` map passed to `spark-submit`) is still built
with plain `Collectors.toMap(...replacePlaceholder...)` and never routed
through `omitUnresolvedTemplateMap`/`omitEmptyArguments`.
All three Iceberg jobs call `.withConfigs(buildSparkConfigs())`, and
`IcebergSparkConfigUtils.buildTemplateSparkConfigs()` fills optional keys like
`spark.executor.instances` with `{{spark_executor_instances}}`. If a caller
omits that optional jobConf key, the literal string
`"{{spark_executor_instances}}"` stays in the resolved `configs` map and is
emitted as `--conf spark.executor.instances={{spark_executor_instances}}` to
`spark-submit` — exactly the "dangling unresolved placeholder" class of bug
this PR claims to fix for arguments/environments, left open for Spark configs.
##########
maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergJobUtils.java:
##########
@@ -18,21 +18,69 @@
*/
package org.apache.gravitino.maintenance.jobs.iceberg;
-import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.HashMap;
import java.util.Map;
+import
org.apache.gravitino.maintenance.optimizer.common.util.IcebergSparkConfigUtils;
/**
* Shared utility methods for Iceberg maintenance jobs.
*
- * <p>Provides SQL escaping, argument parsing, and Spark configuration
utilities used by both {@link
- * IcebergRewriteDataFilesJob} and {@link IcebergExpireSnapshotsJob}.
+ * <p>Provides SQL escaping, argument parsing, Spark configuration utilities,
and classpath checks
+ * used by built-in Iceberg Spark jobs.
*/
public final class IcebergJobUtils {
+ private static final String ICEBERG_SPARK_CATALOG =
"org.apache.iceberg.spark.SparkCatalog";
+ private static final String OPTION_SPARK_CONF = "spark-conf";
+
private IcebergJobUtils() {}
+ /**
+ * Ensures the Iceberg Spark runtime is on the current classpath.
+ *
+ * <p>Built-in templates configure {@code IcebergSparkSessionExtensions} and
{@code SparkCatalog},
+ * but Spark only warns when those classes are missing and continues without
Iceberg support. Call
+ * this after {@code SparkSession} creation (so {@code spark.jars} from
{@code spark_conf} is
+ * visible) and fail the job when the runtime is absent.
+ *
+ * @throws IllegalStateException when required Iceberg Spark classes cannot
be loaded
+ */
+ public static void requireIcebergSparkRuntime() {
+ requireClass(
+ IcebergSparkConfigUtils.ICEBERG_SPARK_EXTENSIONS, "Iceberg Spark
session extensions");
+ requireClass(ICEBERG_SPARK_CATALOG, "Iceberg Spark catalog");
+ }
+
+ /** Visible for unit tests that assert the missing-class error message. */
+ static void requireClassForTest(String className, String description) {
Review Comment:
**conventions**: The new package-private `requireClassForTest` and private
`requireClass` are inserted directly after the constructor, between it and the
class's pre-existing public methods (`escapeSqlString`, `escapeSqlIdentifier`,
`parseArguments`, `parseCustomSparkConfigs`), breaking the visibility ordering
the class had before this diff.
CLAUDE.md's Class Member Ordering rule states: "Methods (Group by
visibility, putting `private` methods at the end)." This insertion point
produces public -> package-private -> private -> public -> public -> public ->
public, sandwiching private/package-private methods in the middle instead of at
the end. This ordering was clean before this PR.
##########
maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergJobUtils.java:
##########
@@ -18,21 +18,69 @@
*/
package org.apache.gravitino.maintenance.jobs.iceberg;
-import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.HashMap;
import java.util.Map;
+import
org.apache.gravitino.maintenance.optimizer.common.util.IcebergSparkConfigUtils;
/**
* Shared utility methods for Iceberg maintenance jobs.
*
- * <p>Provides SQL escaping, argument parsing, and Spark configuration
utilities used by both {@link
- * IcebergRewriteDataFilesJob} and {@link IcebergExpireSnapshotsJob}.
+ * <p>Provides SQL escaping, argument parsing, Spark configuration utilities,
and classpath checks
+ * used by built-in Iceberg Spark jobs.
*/
public final class IcebergJobUtils {
+ private static final String ICEBERG_SPARK_CATALOG =
"org.apache.iceberg.spark.SparkCatalog";
+ private static final String OPTION_SPARK_CONF = "spark-conf";
+
private IcebergJobUtils() {}
+ /**
+ * Ensures the Iceberg Spark runtime is on the current classpath.
+ *
+ * <p>Built-in templates configure {@code IcebergSparkSessionExtensions} and
{@code SparkCatalog},
+ * but Spark only warns when those classes are missing and continues without
Iceberg support. Call
+ * this after {@code SparkSession} creation (so {@code spark.jars} from
{@code spark_conf} is
+ * visible) and fail the job when the runtime is absent.
+ *
+ * @throws IllegalStateException when required Iceberg Spark classes cannot
be loaded
+ */
+ public static void requireIcebergSparkRuntime() {
+ requireClass(
+ IcebergSparkConfigUtils.ICEBERG_SPARK_EXTENSIONS, "Iceberg Spark
session extensions");
+ requireClass(ICEBERG_SPARK_CATALOG, "Iceberg Spark catalog");
+ }
+
+ /** Visible for unit tests that assert the missing-class error message. */
+ static void requireClassForTest(String className, String description) {
+ requireClass(className, description);
+ }
+
+ private static void requireClass(String className, String description) {
+ ClassLoader contextLoader = Thread.currentThread().getContextClassLoader();
+ ClassLoader fallbackLoader = IcebergJobUtils.class.getClassLoader();
+ try {
+ Class.forName(className, true, contextLoader != null ? contextLoader :
fallbackLoader);
Review Comment:
**efficiency**: `requireClass` calls `Class.forName(className, true,
loader)` with `initialize=true`, forcing the target class's static initializer
to run just to check it's on the classpath. Every job launch pays the cost of
running `<clinit>` for `IcebergSparkSessionExtensions`/`SparkCatalog` just to
confirm they're loadable — work Spark will redundantly repeat moments later
when it actually uses them.
Worse: if `iceberg-spark-runtime` is correctly on the classpath but
`SparkCatalog`'s static init fails for an unrelated reason (e.g. a transitive
dependency version conflict), the resulting `LinkageError` subtype is caught by
this same generic handler and the job exits with "Missing Iceberg Spark catalog
... Match the artifact to your Spark/Scala/Iceberg versions" even though the
runtime jar is present and the real cause is a different classpath conflict —
misdirecting operators. `Class.forName(className, false, loader)` would avoid
both the wasted init cost and this misdiagnosis.
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -947,17 +945,91 @@ static String replacePlaceholder(String inputString,
Map<String, String> replace
String key = matcher.group(1);
String replacement = replacements.get(key);
if (replacement != null) {
- matcher.appendReplacement(result, replacement);
+ matcher.appendReplacement(result,
Matcher.quoteReplacement(replacement));
} else {
// If no replacement is found, keep the placeholder as is
- matcher.appendReplacement(result, matcher.group(0));
+ matcher.appendReplacement(result,
Matcher.quoteReplacement(matcher.group(0)));
}
}
matcher.appendTail(result);
return result.toString();
}
+ /**
+ * Drop blank / unresolved optional template arguments after placeholder
substitution.
+ *
+ * <p>Built-in templates always list optional flags as {@code --flag} +
{@code {{placeholder}}}.
+ * When the job conf omits that key or supplies an empty value, leaving the
flag in the command
+ * produces dangling arguments such as {@code --updater-options
--spark-conf}. This method
+ * removes:
+ *
+ * <ul>
+ * <li>blank tokens
+ * <li>tokens that are still an entire unresolved {@code {{placeholder}}}
+ * <li>{@code --flag} pairs whose following value is blank or an
unresolved placeholder
+ * </ul>
+ *
+ * @param arguments arguments after {@link #replacePlaceholder(String, Map)}
+ * @return compacted argument list suitable for process execution
+ */
+ @VisibleForTesting
+ static List<String> omitEmptyArguments(List<String> arguments) {
+ if (arguments == null || arguments.isEmpty()) {
+ return arguments;
+ }
+
+ List<String> result = new ArrayList<>(arguments.size());
+ for (int i = 0; i < arguments.size(); i++) {
+ String arg = arguments.get(i);
+ if (isUnresolvedOptionalValue(arg)) {
+ continue;
+ }
+
+ if (arg.startsWith("--") && i + 1 < arguments.size()) {
+ String next = arguments.get(i + 1);
+ if (!next.startsWith("--") && isUnresolvedOptionalValue(next)) {
+ i++;
+ continue;
+ }
+ }
+
+ result.add(arg);
+ }
+ return result;
+ }
+
+ /**
+ * Resolves optional template maps such as {@code environments}. Entries
whose keys or values are
+ * blank or still an unresolved {@code {{placeholder}}} after substitution
are dropped so
+ * unauthenticated / optional credentials do not become literal placeholder
strings.
+ *
+ * <p>{@code arguments} use {@link #omitEmptyArguments(List)}; {@code
customFields} still keep
+ * unresolved placeholders as literal text.
+ *
+ * @param source template map before substitution
+ * @param jobConf replacement values
+ * @return resolved map without blank or unresolved optional entries
+ */
+ private static Map<String, String> omitUnresolvedTemplateMap(
+ Map<String, String> source, Map<String, String> jobConf) {
+ Map<String, String> resolved = new LinkedHashMap<>();
+ for (Map.Entry<String, String> entry : source.entrySet()) {
+ String key = replacePlaceholder(entry.getKey(), jobConf);
+ String value = replacePlaceholder(entry.getValue(), jobConf);
+ if (isUnresolvedOptionalValue(key) || isUnresolvedOptionalValue(value)) {
+ continue;
+ }
+ resolved.put(key, value);
Review Comment:
**correctness (high)**: `omitUnresolvedTemplateMap` replaced the old
`Collectors.toMap(...)`-based resolution (which threw `IllegalStateException`
on a duplicate-key collision after placeholder substitution) with this
hand-rolled loop doing `resolved.put(key, value)` — silently keeping only the
last value on a collision instead of failing loudly.
A template's `environments = {"{{A}}": "v1", "{{B}}": "v2"}` submitted with
`jobConf = {A: "SAME", B: "SAME"}` previously failed fast at job-creation time
with a clear "Duplicate key" error. Now it silently produces `{"SAME": "v2"}`,
dropping `v1` with no indication to the caller — a regression from fail-fast
validation to silent data loss.
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -947,17 +945,91 @@ static String replacePlaceholder(String inputString,
Map<String, String> replace
String key = matcher.group(1);
String replacement = replacements.get(key);
if (replacement != null) {
- matcher.appendReplacement(result, replacement);
+ matcher.appendReplacement(result,
Matcher.quoteReplacement(replacement));
} else {
// If no replacement is found, keep the placeholder as is
- matcher.appendReplacement(result, matcher.group(0));
+ matcher.appendReplacement(result,
Matcher.quoteReplacement(matcher.group(0)));
}
}
matcher.appendTail(result);
return result.toString();
}
+ /**
+ * Drop blank / unresolved optional template arguments after placeholder
substitution.
+ *
+ * <p>Built-in templates always list optional flags as {@code --flag} +
{@code {{placeholder}}}.
+ * When the job conf omits that key or supplies an empty value, leaving the
flag in the command
+ * produces dangling arguments such as {@code --updater-options
--spark-conf}. This method
+ * removes:
+ *
+ * <ul>
+ * <li>blank tokens
+ * <li>tokens that are still an entire unresolved {@code {{placeholder}}}
+ * <li>{@code --flag} pairs whose following value is blank or an
unresolved placeholder
+ * </ul>
+ *
+ * @param arguments arguments after {@link #replacePlaceholder(String, Map)}
+ * @return compacted argument list suitable for process execution
+ */
+ @VisibleForTesting
+ static List<String> omitEmptyArguments(List<String> arguments) {
+ if (arguments == null || arguments.isEmpty()) {
+ return arguments;
+ }
+
+ List<String> result = new ArrayList<>(arguments.size());
+ for (int i = 0; i < arguments.size(); i++) {
+ String arg = arguments.get(i);
+ if (isUnresolvedOptionalValue(arg)) {
+ continue;
+ }
+
+ if (arg.startsWith("--") && i + 1 < arguments.size()) {
+ String next = arguments.get(i + 1);
+ if (!next.startsWith("--") && isUnresolvedOptionalValue(next)) {
+ i++;
+ continue;
+ }
+ }
+
+ result.add(arg);
+ }
+ return result;
+ }
+
+ /**
+ * Resolves optional template maps such as {@code environments}. Entries
whose keys or values are
+ * blank or still an unresolved {@code {{placeholder}}} after substitution
are dropped so
+ * unauthenticated / optional credentials do not become literal placeholder
strings.
+ *
+ * <p>{@code arguments} use {@link #omitEmptyArguments(List)}; {@code
customFields} still keep
+ * unresolved placeholders as literal text.
+ *
+ * @param source template map before substitution
+ * @param jobConf replacement values
+ * @return resolved map without blank or unresolved optional entries
+ */
+ private static Map<String, String> omitUnresolvedTemplateMap(
+ Map<String, String> source, Map<String, String> jobConf) {
+ Map<String, String> resolved = new LinkedHashMap<>();
+ for (Map.Entry<String, String> entry : source.entrySet()) {
+ String key = replacePlaceholder(entry.getKey(), jobConf);
+ String value = replacePlaceholder(entry.getValue(), jobConf);
+ if (isUnresolvedOptionalValue(key) || isUnresolvedOptionalValue(value)) {
+ continue;
+ }
+ resolved.put(key, value);
+ }
+ return resolved;
+ }
+
+ @VisibleForTesting
+ static boolean isUnresolvedOptionalValue(String value) {
+ return StringUtils.isBlank(value) ||
PLACEHOLDER_PATTERN.matcher(value).matches();
Review Comment:
**correctness**: `isUnresolvedOptionalValue` only detects a token that IS
ENTIRELY one unresolved placeholder
(`PLACEHOLDER_PATTERN.matcher(value).matches()`, a full-string match). A
composite value with a literal prefix/suffix around an unresolved placeholder —
or multiple concatenated placeholders where only one resolves — is never
classified as unresolved and passes through with literal `{{...}}` text still
embedded.
An argument or environment value built as `"prefix-{{missing_key}}"` or
`"{{a}}.{{b}}"` (only one of `a`/`b` supplied) resolves via
`replacePlaceholder` to e.g. `"prefix-{{missing_key}}"`, which is not blank and
does not fully match the placeholder pattern, so it's kept verbatim and handed
to the job process — reproducing the exact dangling-literal-placeholder defect
(a fake-looking credential, a broken SQL/JSON fragment) this PR's own
description says it eliminates, just for composite strings instead of
whole-value placeholders.
##########
maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergJobUtils.java:
##########
@@ -18,21 +18,69 @@
*/
package org.apache.gravitino.maintenance.jobs.iceberg;
-import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.HashMap;
import java.util.Map;
+import
org.apache.gravitino.maintenance.optimizer.common.util.IcebergSparkConfigUtils;
/**
* Shared utility methods for Iceberg maintenance jobs.
*
- * <p>Provides SQL escaping, argument parsing, and Spark configuration
utilities used by both {@link
- * IcebergRewriteDataFilesJob} and {@link IcebergExpireSnapshotsJob}.
+ * <p>Provides SQL escaping, argument parsing, Spark configuration utilities,
and classpath checks
+ * used by built-in Iceberg Spark jobs.
*/
public final class IcebergJobUtils {
+ private static final String ICEBERG_SPARK_CATALOG =
"org.apache.iceberg.spark.SparkCatalog";
+ private static final String OPTION_SPARK_CONF = "spark-conf";
Review Comment:
**correctness**: This PR's stated goal is fixing wrong/hardcoded CLI flag
names in JSON parse errors, and `IcebergUpdateStatsAndMetricsJob` does this
correctly end-to-end via shared `OPTION_UPDATER_OPTIONS`/`OPTION_SPARK_CONF`
constants used both for `argMap.get(...)` and the error message. But
`OPTION_SPARK_CONF` here is private to this file, and
`IcebergExpireSnapshotsJob.java:120` /
`IcebergRewriteDataFilesJob.java:130-131` still call
`argMap.get("spark-conf")`/`argMap.get("options")` with hardcoded string
literals, disconnected from this constant and from
`IcebergRewriteDataFilesJob`'s own `OPTION_OPTIONS` (which only feeds its error
message).
In those two classes, the string actually used to read the CLI flag and the
constant used to name that flag in the error message remain two independent
sources of truth. If either flag is renamed later, it's easy to update one and
miss the other — reproducing the exact "error message names a flag that doesn't
match the real CLI parsing" bug this PR claims to have fixed, just in these two
classes instead of all three.
##########
maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergExpireSnapshotsJob.java:
##########
@@ -148,6 +148,13 @@ public static void main(String[] args) {
}
SparkSession spark = sparkBuilder.getOrCreate();
+ try {
Review Comment:
**reuse**: The fail-fast block (`try {
IcebergJobUtils.requireIcebergSparkRuntime(); } catch (IllegalStateException e)
{ System.err.println(...); spark.stop(); System.exit(1); }`) is copy-pasted
verbatim into all three job `main()` methods (also
`IcebergRewriteDataFilesJob.java:162` and
`IcebergUpdateStatsAndMetricsJob.java:116`) instead of being one shared helper.
A future change to the fail-fast contract (log format, exit code, a
metric/hook before exit, or wrapping `spark.stop()` in its own try/catch to
avoid masking the original error) requires editing three files in lockstep; a
maintainer who updates two and misses the third silently reintroduces
inconsistent behavior. A one-line static helper (e.g.
`IcebergJobUtils.requireIcebergSparkRuntimeOrExit(SparkSession spark)`) would
collapse this to one call site per job.
--
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]
