jerryshao commented on code in PR #13134:
URL: https://github.com/apache/gravitino/pull/13134#discussion_r4015334111
##########
maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteDataFilesJob.java:
##########
@@ -407,19 +426,19 @@ static Map<String, String> parseOptionsJson(String
optionsJson) {
*/
private static List<String> buildArguments() {
return Arrays.asList(
- "--catalog",
+ "--" + IcebergJobUtils.OPTION_CATALOG,
"{{catalog_name}}",
- "--table",
+ "--" + IcebergJobUtils.OPTION_TABLE,
"{{table_identifier}}",
- "--strategy",
+ "--" + OPTION_STRATEGY,
"{{strategy}}",
- "--sort-order",
+ "--" + OPTION_SORT_ORDER,
"{{sort_order}}",
- "--where",
+ "--" + OPTION_WHERE,
Review Comment:
**correctness (high) — regression introduced by this round's fix**: The
optional `--where` flag (`OPTION_WHERE = "where"`) pairs with placeholder
`{{where_clause}}` here — the names don't match after normalization, so
`JobManager.isMatchingUnresolvedFlagValue` never recognizes this as a droppable
optional pair. The *old* logic (`isUnresolvedOptionalValue`) only checked
whether the whole token was a placeholder, with no name-matching requirement,
so it correctly dropped `{{where_clause}}` when unresolved — the new
name-matching requirement broke this specific pair.
Submit `builtin-iceberg-rewrite-data-files` omitting `where_clause` (a
completely normal, non-adversarial choice not to filter — the class's own
Javadoc documents `jobConf.put("where_clause", ...)` as the expected key).
`omitEmptyArguments` doesn't drop the pair (name mismatch), and
`rejectEmbeddedUnresolvedPlaceholder` doesn't fire either since
`{{where_clause}}` is a whole-token match, not an embedded one. The literal
string `"{{where_clause}}"` is passed to the job process, parsed as
`whereClause`, and embedded verbatim into the generated SQL as `where =>
'{{where_clause}}'` — producing a SQL failure instead of running unfiltered. No
test exercises this: `TestIcebergRewriteDataFilesJob` only asserts the raw
(unsubstituted) `buildArguments()` list, and none of the new `TestJobTemplate`
tests use `where_clause`.
---
**correctness**: The required `--catalog`/`{{catalog_name}}` and
`--table`/`{{table_identifier}}` pairs a few lines above have the same name
mismatch. This one is pre-existing (not introduced by this diff) but directly
relevant to a PR titled "Harden built-in Iceberg jobs": if a caller omits
`catalog_name`, the literal string `"{{catalog_name}}"` is non-null, so the
job's own `catalogName == null` guard doesn't trigger, and Spark/Iceberg later
fails with an opaque catalog-resolution error instead of the intended clear
message.
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -841,35 +844,41 @@ public static JobTemplate createRuntimeJobTemplate(
String comment = jobTemplateEntity.comment();
JobTemplateEntity.TemplateContent content =
jobTemplateEntity.templateContent();
- String executable =
- fetchFileFromUri(
- replacePlaceholder(content.executable(), jobConf), stagingDir,
TIMEOUT_IN_MS);
+ String executableUri = replacePlaceholder(content.executable(), jobConf);
+ rejectEmbeddedUnresolvedPlaceholder(executableUri, "executable");
+ String executable = fetchFileFromUri(executableUri, stagingDir,
TIMEOUT_IN_MS);
List<String> args =
Review Comment:
**correctness**: This PR's javadoc/tests advertise that an explicit empty
string in `jobConf` is now kept rather than dropped. But the downstream CLI
parser used by all three built-in Iceberg jobs
(`IcebergJobUtils.parseArguments`, untouched by this PR) unconditionally drops
any `--flag`/empty-value pair via `if (value != null &&
!value.trim().isEmpty())`. The two layers are out of sync, so this advertised
behavior has no observable effect for any built-in Iceberg job: a caller
setting `jobConf.put("updater_options", "")` to explicitly clear/override the
default gets `"--updater-options" ""` kept here, but
`argMap.get("updater-options")` is still `null` once the job process parses it
— identical to the key never having been supplied.
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -947,17 +960,143 @@ 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 unresolved optional {@code --flag}/value pairs after placeholder
substitution.
+ *
+ * <p>Built-in templates list optional flags as {@code --flag} + {@code
{{placeholder}}} with
+ * matching names (for example {@code --updater-options} / {@code
{{updater_options}}}). When the
+ * job conf omits that key, leaving the unresolved placeholder produces
dangling arguments such as
+ * {@code --updater-options {{updater_options}}}. This method removes a pair
only when the
+ * following token is an entire unresolved placeholder whose name matches
the flag (after
+ * normalizing {@code -} / {@code _} and case).
+ *
+ * <p>A boolean flag followed by an unrelated placeholder (for example
{@code --verbose} / {@code
+ * {{unset_flag}}}) is not treated as a pair. Bare / positional unresolved
placeholders (for
+ * example {@code {{slices}}} in SparkPi) are kept so a missing required
argument stays visible.
+ * An explicit empty string in {@code jobConf} is also kept.
+ *
+ * @param arguments arguments after {@link #replacePlaceholder(String,
Map)}; must not be null
+ * @return compacted argument list suitable for process execution; never null
+ */
+ @VisibleForTesting
+ static List<String> omitEmptyArguments(List<String> arguments) {
+ Preconditions.checkNotNull(arguments, "arguments");
+ if (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 (arg.startsWith("--") && i + 1 < arguments.size()) {
+ String next = arguments.get(i + 1);
+ if (!next.startsWith("--") && isMatchingUnresolvedFlagValue(arg,
next)) {
+ i++;
+ continue;
+ }
+ }
+
+ result.add(arg);
+ }
+ return result;
+ }
+
+ /**
+ * Whether {@code value} is an unresolved {@code {{placeholder}}} whose name
matches {@code
+ * flagArg} (for example {@code --updater-options} matches {@code
{{updater_options}}}).
+ */
+ @VisibleForTesting
+ static boolean isMatchingUnresolvedFlagValue(String flagArg, String value) {
+ if (!flagArg.startsWith("--") || !isUnresolvedPlaceholder(value)) {
+ return false;
+ }
+ Matcher matcher = PLACEHOLDER_PATTERN.matcher(value);
+ if (!matcher.matches()) {
+ return false;
+ }
+ String normalizedFlag = flagArg.substring(2).replace('-',
'_').toLowerCase(Locale.ROOT);
+ String normalizedPlaceholder = matcher.group(1).replace('-',
'_').toLowerCase(Locale.ROOT);
+ return normalizedFlag.equals(normalizedPlaceholder);
+ }
+
+ /**
+ * Resolves optional template maps such as {@code environments} and Spark
{@code configs}. Entries
+ * whose keys or values are still an unresolved {@code {{placeholder}}}
after substitution are
+ * dropped so optional credentials / Spark confs do not become literal
placeholder strings.
+ *
+ * <p>Explicit empty strings from {@code jobConf} are kept. {@code
arguments} use {@link
+ * #omitEmptyArguments(List)}; {@code customFields} still keep unresolved
placeholders as literal
+ * text when the entire token is one placeholder.
+ *
+ * <p>If two entries resolve to the same key after placeholder substitution,
this method fails
+ * fast with {@link IllegalStateException}, matching {@link
Collectors#toMap}.
+ *
+ * <p>Composite values that still embed an unresolved placeholder (for
example {@code
+ * prefix-{{missing}}} or {@code {{a}}.{{b}}} with only one key supplied)
fail fast with {@link
+ * IllegalArgumentException} instead of being passed through as literal text.
+ *
+ * @param source template map before substitution
+ * @param jobConf replacement values
+ * @param context label used in error messages (for example {@code
environment} or {@code
+ * configs})
+ * @return resolved map without unresolved optional entries
+ */
+ private static Map<String, String> omitUnresolvedTemplateMap(
Review Comment:
**conventions**: This new `private` `omitUnresolvedTemplateMap` is inserted
between two package-private `@VisibleForTesting` methods
(`isMatchingUnresolvedFlagValue` above it,
`isUnresolvedPlaceholder`/`rejectEmbeddedUnresolvedPlaceholder` below it),
breaking the visibility ordering CLAUDE.md requires — a new violation
introduced by this round's fixes (the previously-reported ordering issue was in
`IcebergJobUtils.java` and is correctly fixed there).
CLAUDE.md's Class Member Ordering rule: "Methods (Group by visibility,
putting `private` methods at the end)."
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -947,17 +960,143 @@ 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 unresolved optional {@code --flag}/value pairs after placeholder
substitution.
+ *
+ * <p>Built-in templates list optional flags as {@code --flag} + {@code
{{placeholder}}} with
+ * matching names (for example {@code --updater-options} / {@code
{{updater_options}}}). When the
+ * job conf omits that key, leaving the unresolved placeholder produces
dangling arguments such as
+ * {@code --updater-options {{updater_options}}}. This method removes a pair
only when the
+ * following token is an entire unresolved placeholder whose name matches
the flag (after
+ * normalizing {@code -} / {@code _} and case).
+ *
+ * <p>A boolean flag followed by an unrelated placeholder (for example
{@code --verbose} / {@code
+ * {{unset_flag}}}) is not treated as a pair. Bare / positional unresolved
placeholders (for
+ * example {@code {{slices}}} in SparkPi) are kept so a missing required
argument stays visible.
+ * An explicit empty string in {@code jobConf} is also kept.
+ *
+ * @param arguments arguments after {@link #replacePlaceholder(String,
Map)}; must not be null
+ * @return compacted argument list suitable for process execution; never null
+ */
+ @VisibleForTesting
+ static List<String> omitEmptyArguments(List<String> arguments) {
+ Preconditions.checkNotNull(arguments, "arguments");
+ if (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 (arg.startsWith("--") && i + 1 < arguments.size()) {
+ String next = arguments.get(i + 1);
+ if (!next.startsWith("--") && isMatchingUnresolvedFlagValue(arg,
next)) {
+ i++;
+ continue;
+ }
+ }
+
+ result.add(arg);
+ }
+ return result;
+ }
+
+ /**
+ * Whether {@code value} is an unresolved {@code {{placeholder}}} whose name
matches {@code
+ * flagArg} (for example {@code --updater-options} matches {@code
{{updater_options}}}).
+ */
+ @VisibleForTesting
+ static boolean isMatchingUnresolvedFlagValue(String flagArg, String value) {
+ if (!flagArg.startsWith("--") || !isUnresolvedPlaceholder(value)) {
+ return false;
+ }
+ Matcher matcher = PLACEHOLDER_PATTERN.matcher(value);
+ if (!matcher.matches()) {
+ return false;
+ }
+ String normalizedFlag = flagArg.substring(2).replace('-',
'_').toLowerCase(Locale.ROOT);
+ String normalizedPlaceholder = matcher.group(1).replace('-',
'_').toLowerCase(Locale.ROOT);
+ return normalizedFlag.equals(normalizedPlaceholder);
+ }
+
+ /**
+ * Resolves optional template maps such as {@code environments} and Spark
{@code configs}. Entries
+ * whose keys or values are still an unresolved {@code {{placeholder}}}
after substitution are
+ * dropped so optional credentials / Spark confs do not become literal
placeholder strings.
+ *
+ * <p>Explicit empty strings from {@code jobConf} are kept. {@code
arguments} use {@link
+ * #omitEmptyArguments(List)}; {@code customFields} still keep unresolved
placeholders as literal
+ * text when the entire token is one placeholder.
+ *
+ * <p>If two entries resolve to the same key after placeholder substitution,
this method fails
+ * fast with {@link IllegalStateException}, matching {@link
Collectors#toMap}.
+ *
+ * <p>Composite values that still embed an unresolved placeholder (for
example {@code
+ * prefix-{{missing}}} or {@code {{a}}.{{b}}} with only one key supplied)
fail fast with {@link
+ * IllegalArgumentException} instead of being passed through as literal text.
+ *
+ * @param source template map before substitution
+ * @param jobConf replacement values
+ * @param context label used in error messages (for example {@code
environment} or {@code
+ * configs})
+ * @return resolved map without unresolved optional entries
+ */
+ private static Map<String, String> omitUnresolvedTemplateMap(
+ Map<String, String> source, Map<String, String> jobConf, String context)
{
+ 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 (isUnresolvedPlaceholder(key) || isUnresolvedPlaceholder(value)) {
+ continue;
+ }
+ rejectEmbeddedUnresolvedPlaceholder(key, context + " key");
+ rejectEmbeddedUnresolvedPlaceholder(value, context + " value");
+ if (resolved.containsKey(key)) {
+ throw new IllegalStateException(
+ String.format(
+ "Duplicate key %s (attempted merging values %s and %s)",
Review Comment:
**correctness**: This new `IllegalStateException` (thrown on a duplicate key
after placeholder substitution) is not special-cased by `JobExceptionHandler`
on the server side (which only handles
`IllegalArgumentException`/`NotFoundException`/`NotInUseException`/`ForbiddenException`),
so it falls through to the generic handler: HTTP 500 instead of 400, and it
also pollutes `ServerHealth.recordFailure` as if it were an internal server
fault.
A user submitting a job via `POST .../jobs/runs` with a `jobConf` where two
environment/config keys collapse to the same string after substitution (exactly
this PR's own new test scenario) gets an unhelpful 500 for what is purely a
client input error.
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -947,17 +960,143 @@ 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 unresolved optional {@code --flag}/value pairs after placeholder
substitution.
+ *
+ * <p>Built-in templates list optional flags as {@code --flag} + {@code
{{placeholder}}} with
+ * matching names (for example {@code --updater-options} / {@code
{{updater_options}}}). When the
+ * job conf omits that key, leaving the unresolved placeholder produces
dangling arguments such as
+ * {@code --updater-options {{updater_options}}}. This method removes a pair
only when the
+ * following token is an entire unresolved placeholder whose name matches
the flag (after
+ * normalizing {@code -} / {@code _} and case).
+ *
+ * <p>A boolean flag followed by an unrelated placeholder (for example
{@code --verbose} / {@code
+ * {{unset_flag}}}) is not treated as a pair. Bare / positional unresolved
placeholders (for
+ * example {@code {{slices}}} in SparkPi) are kept so a missing required
argument stays visible.
+ * An explicit empty string in {@code jobConf} is also kept.
+ *
+ * @param arguments arguments after {@link #replacePlaceholder(String,
Map)}; must not be null
+ * @return compacted argument list suitable for process execution; never null
+ */
+ @VisibleForTesting
+ static List<String> omitEmptyArguments(List<String> arguments) {
+ Preconditions.checkNotNull(arguments, "arguments");
+ if (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 (arg.startsWith("--") && i + 1 < arguments.size()) {
+ String next = arguments.get(i + 1);
+ if (!next.startsWith("--") && isMatchingUnresolvedFlagValue(arg,
next)) {
+ i++;
+ continue;
+ }
+ }
+
+ result.add(arg);
+ }
+ return result;
+ }
+
+ /**
+ * Whether {@code value} is an unresolved {@code {{placeholder}}} whose name
matches {@code
+ * flagArg} (for example {@code --updater-options} matches {@code
{{updater_options}}}).
+ */
+ @VisibleForTesting
+ static boolean isMatchingUnresolvedFlagValue(String flagArg, String value) {
+ if (!flagArg.startsWith("--") || !isUnresolvedPlaceholder(value)) {
+ return false;
+ }
+ Matcher matcher = PLACEHOLDER_PATTERN.matcher(value);
+ if (!matcher.matches()) {
+ return false;
+ }
+ String normalizedFlag = flagArg.substring(2).replace('-',
'_').toLowerCase(Locale.ROOT);
+ String normalizedPlaceholder = matcher.group(1).replace('-',
'_').toLowerCase(Locale.ROOT);
+ return normalizedFlag.equals(normalizedPlaceholder);
+ }
+
+ /**
+ * Resolves optional template maps such as {@code environments} and Spark
{@code configs}. Entries
+ * whose keys or values are still an unresolved {@code {{placeholder}}}
after substitution are
+ * dropped so optional credentials / Spark confs do not become literal
placeholder strings.
+ *
+ * <p>Explicit empty strings from {@code jobConf} are kept. {@code
arguments} use {@link
+ * #omitEmptyArguments(List)}; {@code customFields} still keep unresolved
placeholders as literal
+ * text when the entire token is one placeholder.
+ *
+ * <p>If two entries resolve to the same key after placeholder substitution,
this method fails
+ * fast with {@link IllegalStateException}, matching {@link
Collectors#toMap}.
+ *
+ * <p>Composite values that still embed an unresolved placeholder (for
example {@code
+ * prefix-{{missing}}} or {@code {{a}}.{{b}}} with only one key supplied)
fail fast with {@link
+ * IllegalArgumentException} instead of being passed through as literal text.
+ *
+ * @param source template map before substitution
+ * @param jobConf replacement values
+ * @param context label used in error messages (for example {@code
environment} or {@code
+ * configs})
+ * @return resolved map without unresolved optional entries
+ */
+ private static Map<String, String> omitUnresolvedTemplateMap(
+ Map<String, String> source, Map<String, String> jobConf, String context)
{
+ 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 (isUnresolvedPlaceholder(key) || isUnresolvedPlaceholder(value)) {
+ continue;
+ }
+ rejectEmbeddedUnresolvedPlaceholder(key, context + " key");
+ rejectEmbeddedUnresolvedPlaceholder(value, context + " value");
+ if (resolved.containsKey(key)) {
+ throw new IllegalStateException(
+ String.format(
+ "Duplicate key %s (attempted merging values %s and %s)",
+ key, resolved.get(key), value));
+ }
+ resolved.put(key, value);
+ }
+ return resolved;
+ }
+
+ /**
+ * Whether {@code value} is exactly one unresolved {@code {{placeholder}}}
token.
+ *
+ * <p>Blank / empty strings are not unresolved placeholders: a present
{@code jobConf} key with
+ * value {@code ""} means the caller explicitly supplied an empty value.
+ */
+ @VisibleForTesting
+ static boolean isUnresolvedPlaceholder(String value) {
+ return value != null && PLACEHOLDER_PATTERN.matcher(value).matches();
+ }
+
+ @VisibleForTesting
+ static void rejectEmbeddedUnresolvedPlaceholder(String value, String
context) {
Review Comment:
**conventions**: The earlier-round "missing `@Nullable`" finding wasn't
actually fixed, just relocated — `rejectEmbeddedUnresolvedPlaceholder(String
value, String context)` and `isUnresolvedPlaceholder(String value)` both safely
accept and handle `null` (guarded internally, exercised with `null` in tests)
but carry no `@Nullable` annotation, and the file has no such import at all.
---
**efficiency (minor)**: `rejectEmbeddedUnresolvedPlaceholder` calls
`isUnresolvedPlaceholder` (a full `.matches()` pass) then separately does
`PLACEHOLDER_PATTERN.matcher(value).find()` — two full regex passes and two
`Matcher` allocations per value, invoked up to ~9 times per job template
resolution. `isMatchingUnresolvedFlagValue` and `omitUnresolvedTemplateMap`
have the same doubled/tripled-pass pattern. Reusing one `Matcher` (match once,
branch, then read the result) would remove the redundant passes — low severity
since this isn't a hot loop, but an easy win.
##########
core/src/test/java/org/apache/gravitino/job/TestJobTemplate.java:
##########
@@ -458,4 +458,415 @@ public void
testCreateSparkRuntimeJobTemplateWithReplacements() throws IOExcepti
.collect(Collectors.toList());
Assertions.assertTrue(archiveNames.contains(archive1.getName()));
}
+
+ @Test
+ public void testOmitEmptyArguments() {
+ // Explicit "" after --updater-options is kept; name-matched --options
{{options}} is dropped;
+ // bare {{stream_results}} / {{slices}} are kept so missing positionals
stay visible.
+ Assertions.assertEquals(
+ Lists.newArrayList(
+ "--catalog",
+ "iceberg",
+ "--table",
+ "db.t",
+ "--updater-options",
+ "",
+ "--spark-conf",
+ "{\"k\":\"v\"}",
+ "{{stream_results}}",
+ " "),
+ JobManager.omitEmptyArguments(
+ Lists.newArrayList(
+ "--catalog",
+ "iceberg",
+ "--table",
+ "db.t",
+ "--updater-options",
+ "",
+ "--spark-conf",
+ "{\"k\":\"v\"}",
+ "--options",
+ "{{options}}",
+ "{{stream_results}}",
+ " ")));
+
+ Assertions.assertEquals(
+ Lists.newArrayList("{{slices}}"),
+ JobManager.omitEmptyArguments(Lists.newArrayList("{{slices}}")));
+
+ // Boolean flag + unrelated placeholder must not be dropped as a pair.
+ Assertions.assertEquals(
+ Lists.newArrayList("--verbose", "{{unset_flag}}"),
+ JobManager.omitEmptyArguments(Lists.newArrayList("--verbose",
"{{unset_flag}}")));
+
+ // Hyphen / underscore names still match for built-in optional flags.
+ Assertions.assertEquals(
+ Lists.newArrayList(),
+ JobManager.omitEmptyArguments(
+ Lists.newArrayList("--updater-options", "{{updater_options}}")));
+
+ Assertions.assertEquals(
+ Lists.newArrayList("--catalog", "iceberg", "--stream-results"),
+ JobManager.omitEmptyArguments(
+ Lists.newArrayList("--catalog", "iceberg", "--stream-results")));
+
+ Assertions.assertThrows(NullPointerException.class, () ->
JobManager.omitEmptyArguments(null));
+ Assertions.assertEquals(
+ Lists.newArrayList(),
JobManager.omitEmptyArguments(Lists.newArrayList()));
+ }
+
+ @Test
+ public void testCreateSparkRuntimeJobTemplateOmitsUnresolvedEnvironments()
throws IOException {
+ File executable = Files.createTempFile(tempDir.toPath(), "testSparkJob",
".jar").toFile();
+ SparkJobTemplate sparkJobTemplate =
+ SparkJobTemplate.builder()
+ .withName("testSparkJobAuthEnv")
+ .withExecutable(executable.toURI().toString())
+ .withClassName("org.apache.gravitino.TestSparkJob")
+ .withEnvironments(
+ ImmutableMap.of(
+ "GRAVITINO_AUTH_TYPE",
+ "{{gravitino_auth_type}}",
+ "GRAVITINO_AUTH_USERNAME",
+ "{{gravitino_auth_username}}",
+ "KEEP_ME",
+ "literal"))
+ .build();
+
+ JobTemplateEntity entity =
+ JobTemplateEntity.builder()
+ .withId(2L)
+ .withName(sparkJobTemplate.name())
+ .withNamespace(NamespaceUtil.ofJobTemplate("test"))
+ .withTemplateContent(
+
JobTemplateEntity.TemplateContent.fromJobTemplate(sparkJobTemplate))
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .build();
+
+ JobTemplate omitted =
+ JobManager.createRuntimeJobTemplate(entity, ImmutableMap.of(),
tempStagingDir);
+ Assertions.assertEquals(ImmutableMap.of("KEEP_ME", "literal"),
omitted.environments());
+
+ JobTemplate withEmptyAuthType =
+ JobManager.createRuntimeJobTemplate(
+ entity, ImmutableMap.of("gravitino_auth_type", ""),
tempStagingDir);
+ Assertions.assertEquals(
+ ImmutableMap.of("GRAVITINO_AUTH_TYPE", "", "KEEP_ME", "literal"),
+ withEmptyAuthType.environments());
+
+ JobTemplate resolved =
+ JobManager.createRuntimeJobTemplate(
+ entity,
+ ImmutableMap.of(
+ "gravitino_auth_type", "basic",
+ "gravitino_auth_username", "admin"),
+ tempStagingDir);
+ Assertions.assertEquals(
+ ImmutableMap.of(
+ "GRAVITINO_AUTH_TYPE",
+ "basic",
+ "GRAVITINO_AUTH_USERNAME",
+ "admin",
+ "KEEP_ME",
+ "literal"),
+ resolved.environments());
+ }
+
+ @Test
+ public void testCreateSparkRuntimeJobTemplateOmitsUnresolvedConfigs() throws
IOException {
+ File executable = Files.createTempFile(tempDir.toPath(), "testSparkJob",
".jar").toFile();
+ SparkJobTemplate sparkJobTemplate =
+ SparkJobTemplate.builder()
+ .withName("testSparkJobConfigs")
+ .withExecutable(executable.toURI().toString())
+ .withClassName("org.apache.gravitino.TestSparkJob")
+ .withConfigs(
+ ImmutableMap.of(
+ "spark.executor.instances",
+ "{{spark_executor_instances}}",
+ "spark.executor.memory",
+ "{{spark_executor_memory}}",
+ "spark.master",
+ "local"))
+ .build();
+
+ JobTemplateEntity entity =
+ JobTemplateEntity.builder()
+ .withId(6L)
+ .withName(sparkJobTemplate.name())
+ .withNamespace(NamespaceUtil.ofJobTemplate("test"))
+ .withTemplateContent(
+
JobTemplateEntity.TemplateContent.fromJobTemplate(sparkJobTemplate))
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .build();
+
+ JobTemplate omitted =
+ JobManager.createRuntimeJobTemplate(entity, ImmutableMap.of(),
tempStagingDir);
+ Assertions.assertEquals(
+ ImmutableMap.of("spark.master", "local"), ((SparkJobTemplate)
omitted).configs());
+
+ JobTemplate withEmptyInstances =
+ JobManager.createRuntimeJobTemplate(
+ entity, ImmutableMap.of("spark_executor_instances", ""),
tempStagingDir);
+ Assertions.assertEquals(
+ ImmutableMap.of("spark.executor.instances", "", "spark.master",
"local"),
+ ((SparkJobTemplate) withEmptyInstances).configs());
+
+ JobTemplate resolved =
+ JobManager.createRuntimeJobTemplate(
+ entity,
+ ImmutableMap.of(
+ "spark_executor_instances", "2",
+ "spark_executor_memory", "1g"),
+ tempStagingDir);
+ Assertions.assertEquals(
+ ImmutableMap.of(
+ "spark.executor.instances",
+ "2",
+ "spark.executor.memory",
+ "1g",
+ "spark.master",
+ "local"),
+ ((SparkJobTemplate) resolved).configs());
+ }
+
+ @Test
+ public void testCreateRuntimeJobTemplateFailsOnEmbeddedUnresolvedConfig()
throws IOException {
+ File executable = Files.createTempFile(tempDir.toPath(), "testSparkJob",
".jar").toFile();
+ SparkJobTemplate sparkJobTemplate =
+ SparkJobTemplate.builder()
+ .withName("testSparkJobEmbeddedConfig")
+ .withExecutable(executable.toURI().toString())
+ .withClassName("org.apache.gravitino.TestSparkJob")
+ .withConfigs(ImmutableMap.of("spark.sql.extensions",
"prefix-{{missing_ext}}"))
+ .build();
+
+ JobTemplateEntity entity =
+ JobTemplateEntity.builder()
+ .withId(7L)
+ .withName(sparkJobTemplate.name())
+ .withNamespace(NamespaceUtil.ofJobTemplate("test"))
+ .withTemplateContent(
+
JobTemplateEntity.TemplateContent.fromJobTemplate(sparkJobTemplate))
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .build();
+
+ IllegalArgumentException thrown =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> JobManager.createRuntimeJobTemplate(entity,
ImmutableMap.of(), tempStagingDir));
+
Assertions.assertTrue(thrown.getMessage().contains("prefix-{{missing_ext}}"));
+ }
+
+ @Test
+ public void testCreateRuntimeJobTemplateFailsOnDuplicateEnvironmentKeys()
throws IOException {
+ File executable = Files.createTempFile(tempDir.toPath(), "testSparkJob",
".jar").toFile();
+ SparkJobTemplate sparkJobTemplate =
+ SparkJobTemplate.builder()
+ .withName("testSparkJobDupEnv")
+ .withExecutable(executable.toURI().toString())
+ .withClassName("org.apache.gravitino.TestSparkJob")
+ .withEnvironments(ImmutableMap.of("{{A}}", "v1", "{{B}}", "v2"))
+ .build();
+
+ JobTemplateEntity entity =
+ JobTemplateEntity.builder()
+ .withId(3L)
+ .withName(sparkJobTemplate.name())
+ .withNamespace(NamespaceUtil.ofJobTemplate("test"))
+ .withTemplateContent(
+
JobTemplateEntity.TemplateContent.fromJobTemplate(sparkJobTemplate))
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .build();
+
+ IllegalStateException thrown =
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () ->
+ JobManager.createRuntimeJobTemplate(
+ entity, ImmutableMap.of("A", "SAME", "B", "SAME"),
tempStagingDir));
+ Assertions.assertTrue(thrown.getMessage().contains("Duplicate key SAME"));
+ }
+
+ @Test
+ public void testRejectEmbeddedUnresolvedPlaceholder() {
Review Comment:
**test-coverage**: Of the ~9 `rejectEmbeddedUnresolvedPlaceholder` call
sites added by the fix commits (`executable`, `argument`, `customFields`
key/value, `script`, `className`, `jar`, `file`, `archive`, plus the map-based
`environment`/`configs` sites), only `configs`, `environment`, and one
composite `argument` case are exercised end-to-end through
`createRuntimeJobTemplate`.
`executable`/`script`/`className`/`jar`/`file`/`archive`/`customFields` have no
failure-path test — `testRejectEmbeddedUnresolvedPlaceholder` here only
unit-tests the helper directly with a generic `"test"` context, not these
specific call sites.
CLAUDE.md: "Write unit tests for ALL new logic. NO tests = NO merge."
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -886,35 +895,39 @@ public static JobTemplate createRuntimeJobTemplate(
// For Spark job template
if (content.jobType() == JobTemplate.JobType.SPARK) {
String className = replacePlaceholder(content.className(), jobConf);
+ rejectEmbeddedUnresolvedPlaceholder(className, "className");
List<String> jars =
content.jars().stream()
.map(
- jar ->
- fetchFileFromUri(replacePlaceholder(jar, jobConf),
stagingDir, TIMEOUT_IN_MS))
+ jar -> {
+ String resolved = replacePlaceholder(jar, jobConf);
+ rejectEmbeddedUnresolvedPlaceholder(resolved, "jar");
+ return fetchFileFromUri(resolved, stagingDir,
TIMEOUT_IN_MS);
+ })
.collect(Collectors.toList());
List<String> files =
content.files().stream()
.map(
- file ->
- fetchFileFromUri(
- replacePlaceholder(file, jobConf), stagingDir,
TIMEOUT_IN_MS))
+ file -> {
+ String resolved = replacePlaceholder(file, jobConf);
+ rejectEmbeddedUnresolvedPlaceholder(resolved, "file");
+ return fetchFileFromUri(resolved, stagingDir,
TIMEOUT_IN_MS);
+ })
.collect(Collectors.toList());
List<String> archives =
content.archives().stream()
.map(
- archive ->
- fetchFileFromUri(
- replacePlaceholder(archive, jobConf), stagingDir,
TIMEOUT_IN_MS))
+ archive -> {
+ String resolved = replacePlaceholder(archive, jobConf);
+ rejectEmbeddedUnresolvedPlaceholder(resolved, "archive");
+ return fetchFileFromUri(resolved, stagingDir,
TIMEOUT_IN_MS);
+ })
.collect(Collectors.toList());
Map<String, String> configs =
- content.configs().entrySet().stream()
- .collect(
- Collectors.toMap(
- entry -> replacePlaceholder(entry.getKey(), jobConf),
- entry -> replacePlaceholder(entry.getValue(), jobConf)));
+ omitUnresolvedTemplateMap(content.configs(), jobConf, "configs");
Review Comment:
**correctness (high)**:
`IcebergSparkConfigUtils.buildTemplateSparkConfigs()`, shared by all three
built-in Iceberg jobs, produces Spark config map **keys** that are themselves
composite strings (`SPARK_SQL_CATALOG_PREFIX + "{{catalog_name}}"`, plus
`.type`/`.uri`/`.warehouse` variants) — exactly the pattern
`rejectEmbeddedUnresolvedPlaceholder` (called via `omitUnresolvedTemplateMap`
here) is designed to reject.
If a caller omits `catalog_name` from `jobConf` (which, per the companion
comment on `IcebergRewriteDataFilesJob.java`, doesn't get caught earlier
either), this line resolves that key to a string with an embedded-but-not-whole
unresolved placeholder, and throws `IllegalArgumentException` with a
`JobManager`-level message ("Unresolved placeholder remains embedded in configs
key after substitution: spark.sql.catalog.{{catalog_name}}") instead of the
job's own clear "Error: --catalog and --table are required arguments" message.
This affects all three Iceberg jobs identically, and no test runs any of the
three jobs' actual `jobTemplate()` output through `createRuntimeJobTemplate` to
catch it — the new embedded-placeholder tests only use synthetic
`SparkJobTemplate` builders with placeholders in *values*, never in composite
*keys* like the real templates produce.
##########
maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateStatsAndMetricsJob.java:
##########
@@ -391,15 +400,22 @@ static Map<String, String> parseArguments(String[] args) {
@VisibleForTesting
static Map<String, String> parseCustomSparkConfigs(String sparkConfJson) {
Review Comment:
**reuse**: This class still hand-rolls its own argument parser
(`parseArguments`, a few lines above this method) instead of calling the shared
`IcebergJobUtils.parseArguments` (used directly by
`IcebergExpireSnapshotsJob`), and the local copy has no boolean-flag branch —
`IcebergJobUtils.parseArguments` treats a trailing `--flag` with no value as
`"true"`; this local copy just drops it. The class also reimplements
`escapeSqlIdentifier` locally instead of delegating to `IcebergJobUtils`, while
`IcebergRewriteDataFilesJob` goes the opposite direction (pure delegation
wrappers) — the three sibling job classes have landed on inconsistent
relationships with the shared utility class.
A future boolean-flag CLI option added to this job would silently behave
differently than the same pattern in the other two Iceberg jobs, and a bug fix
to `IcebergJobUtils.parseArguments`/`escapeSqlIdentifier` wouldn't propagate
here since this class never calls them.
--
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]
