This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 1ac4a35897 [Cherry-pick to branch-1.3] [#13133] fix(maintenance): Fail
Iceberg jobs when Spark runtime is missing (#13134) (#13268)
1ac4a35897 is described below
commit 1ac4a358976d255657fc58bd1f0f3a1d6ba5def1
Author: MaSai <[email protected]>
AuthorDate: Thu Sep 17 16:07:49 2026 +0800
[Cherry-pick to branch-1.3] [#13133] fix(maintenance): Fail Iceberg jobs
when Spark runtime is missing (#13134) (#13268)
**Cherry-pick Information:**
- Original commit: 35c184d2dec23b193aac72191ddd7560c31db828
- Target branch: `branch-1.3`
- Status: Conflicts resolved
Co-authored-by: Cursor <[email protected]>
---
.../optimizer-cli-reference.md | 16 ++
.../optimizer-configuration.md | 12 ++
.../optimizer-troubleshooting.md | 16 ++
docs/table-maintenance-service/optimizer.md | 11 ++
.../maintenance/jobs/iceberg/IcebergJobUtils.java | 177 +++++++++++++++++++++
.../jobs/iceberg/IcebergRewriteDataFilesJob.java | 66 +-------
.../iceberg/IcebergUpdateStatsAndMetricsJob.java | 17 +-
.../jobs/iceberg/TestIcebergJobUtils.java | 57 +++++++
.../iceberg/TestIcebergRewriteDataFilesJob.java | 4 +-
.../jobs/iceberg/TestIcebergUpdateStatsJob.java | 43 +++--
10 files changed, 342 insertions(+), 77 deletions(-)
diff --git a/docs/table-maintenance-service/optimizer-cli-reference.md
b/docs/table-maintenance-service/optimizer-cli-reference.md
index fcfa9130a0..2389cbb956 100644
--- a/docs/table-maintenance-service/optimizer-cli-reference.md
+++ b/docs/table-maintenance-service/optimizer-cli-reference.md
@@ -244,6 +244,22 @@ Three job templates ship with the service, and they are
complementary rather tha
Each can be submitted directly over REST, and the first two are also what the
policy-driven workflow submits on your behalf. See [Quick
Start](./optimizer.md#walkthrough) for the policy-driven path.
+These templates set Iceberg Spark session and catalog classes, but they do not
list an Iceberg Spark
+runtime in `jars`. `gravitino-jobs` also excludes that runtime from its shaded
JAR, so the version
+that runs with your Spark cluster is yours to supply. Provide a matching
+`iceberg-spark-runtime-<sparkMajor>_<scala>` JAR on the Spark classpath used
by the job executor —
+commonly through `spark.jars` in `spark_conf`, or by installing it into
`SPARK_HOME`. Align the
+artifact with the Spark, Scala, and Iceberg versions you actually run. A
reference coordinate used
+in Gravitino's own jobs tests is
`org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.11.0`.
+Without that runtime, built-in Iceberg jobs fail after Spark starts instead of
continuing without
+Iceberg support.
+
+Optional template arguments are still listed as `--flag` + `{{placeholder}}`
pairs. If `jobConf`
+omits a key (or leaves the placeholder unresolved), the flag remains on the
process command line as
+a dangling argument (for example `--updater-options` with no value before
`--spark-conf`). Callers
+and UIs should supply every placeholder they care about with an explicit
value, including optional
+ones they intentionally disable or leave at a documented default, rather than
omitting the key.
+
## Update Statistics
`builtin-iceberg-update-stats` reads a table and writes back the statistics
and metrics that policies evaluate. Compaction policies read
`custom-data-file-mse` and `custom-delete-file-number`, so nothing else will
fire until this job has run at least once.
diff --git a/docs/table-maintenance-service/optimizer-configuration.md
b/docs/table-maintenance-service/optimizer-configuration.md
index 409a0d70e5..cc379da68b 100644
--- a/docs/table-maintenance-service/optimizer-configuration.md
+++ b/docs/table-maintenance-service/optimizer-configuration.md
@@ -79,6 +79,18 @@ A direct job submission carries its own `jobConf`. This is
`builtin-iceberg-upda
`updater_options` and `spark_conf` are JSON strings inside a JSON object, so
their quotes are escaped. That nesting is the most common source of malformed
submissions.
+Built-in Iceberg templates list optional keys as `--flag` + `{{placeholder}}`.
Omitting a key from
+`jobConf` does not remove that flag from the submitted command; it can leave a
dangling flag such as
+`--updater-options` with no value. Prefer sending an explicit value for each
placeholder you use
+(or a documented default) instead of dropping the key. See
+[Built-in Job Templates](./optimizer-cli-reference.md#built-in-job-templates).
+
+Built-in Iceberg templates also need an Iceberg Spark runtime on the Spark
classpath. They do not
+ship that JAR or fill template `jars`, so include it yourself — for example
+`"spark.jars":"/path/to/iceberg-spark-runtime-....jar"` inside `spark_conf`.
Match the artifact to
+your Spark, Scala, and Iceberg versions. Details are under
+[Built-in Job Templates](./optimizer-cli-reference.md#built-in-job-templates).
+
`warehouse_location` may be empty for local filesystem testing. Set it to the
warehouse URI for HDFS or cloud object storage.
## Running Against a Local Filesystem
diff --git a/docs/table-maintenance-service/optimizer-troubleshooting.md
b/docs/table-maintenance-service/optimizer-troubleshooting.md
index 3a2111d69d..e9d382e7e0 100644
--- a/docs/table-maintenance-service/optimizer-troubleshooting.md
+++ b/docs/table-maintenance-service/optimizer-troubleshooting.md
@@ -72,6 +72,22 @@ spark.hadoop.fs.defaultFS=file:///
}
```
+**Built-in Iceberg jobs fail with `Missing Iceberg Spark session extensions`**
—
+Spark only warns when `IcebergSparkSessionExtensions` is missing, so built-in
jobs check the
+classpath after `SparkSession` starts and exit with a non-zero status when the
Iceberg Spark
+runtime is absent. The templates configure Iceberg classes but leave `jars`
empty, and
+`gravitino-jobs` does not bundle `iceberg-spark-runtime`. A stock Spark
install is not enough.
+Put a version-matched Iceberg Spark runtime on the job classpath, for example:
+
+```json
+{
+ "spark.jars": "/path/to/iceberg-spark-runtime-3.5_2.12-1.11.0.jar"
+}
+```
+
+Use the Spark, Scala, and Iceberg versions that match your cluster. See
+[Built-in Job Templates](./optimizer-cli-reference.md#built-in-job-templates).
+
**Rewrite fails on a multi-level partition** — in release `1.2.0`, rewriting a
table partitioned by an identity transform combined with a time transform, such
as `PARTITIONED BY (p, days(ts))`, fails with:
```text
diff --git a/docs/table-maintenance-service/optimizer.md
b/docs/table-maintenance-service/optimizer.md
index 22c3362528..4a0acb8d77 100644
--- a/docs/table-maintenance-service/optimizer.md
+++ b/docs/table-maintenance-service/optimizer.md
@@ -67,6 +67,17 @@ Each step ends with a check. If a check fails, stop there,
since every step depe
- A running Gravitino server with a metalake. The examples use `test`.
- Spark available to the job executor, through either `SPARK_HOME` or
`gravitino.jobExecutor.local.sparkHome`.
+- An Iceberg Spark runtime on that Spark classpath. Built-in Iceberg templates
configure
+ `IcebergSparkSessionExtensions` and `SparkCatalog`, but `gravitino-jobs`
does not ship the
+ Iceberg Spark runtime and the templates leave `jars` empty so your Spark and
Iceberg versions
+ stay under your control. A stock Spark distribution is not enough. Put a
matching
+ `iceberg-spark-runtime-*` JAR on the job classpath — for example with
`spark.jars` in
+ `spark_conf`, or by installing it into your Spark environment. Pick the
artifact that matches
+ your Spark, Scala, and Iceberg versions. The jobs module is built and tested
against Spark 3.5.x
+ and Iceberg 1.11.0 (for example
+ `org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.11.0`). Without it, the
job fails after
+ Spark starts with an error naming the missing Iceberg classes. See
+ [Troubleshooting](./optimizer-troubleshooting.md#job-execution-failures).
- `gravitino.job.statusPullIntervalInMs` lowered to `10000` and the server
restarted. The default is five minutes, which makes every status check in this
walkthrough feel broken.
If your Iceberg REST backend runs in memory, do not restart it partway
through. Restarting resets both metadata and data files, and you start over.
diff --git
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergJobUtils.java
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergJobUtils.java
new file mode 100644
index 0000000000..9f95e92405
--- /dev/null
+++
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergJobUtils.java
@@ -0,0 +1,177 @@
+/*
+ * 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.maintenance.jobs.iceberg;
+
+import java.util.HashMap;
+import java.util.Map;
+import
org.apache.gravitino.maintenance.optimizer.common.util.IcebergSparkConfigUtils;
+import org.apache.spark.sql.SparkSession;
+
+/**
+ * Shared utility methods for Iceberg maintenance jobs.
+ *
+ * <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 IcebergJobUtils() {}
+
+ /**
+ * Escape single quotes in SQL string literals by replacing ' with ''.
+ *
+ * @param value the string value to escape
+ * @return escaped string safe for use in SQL string literals
+ */
+ public static String escapeSqlString(String value) {
+ if (value == null) {
+ return null;
+ }
+ return value.replace("'", "''");
+ }
+
+ /**
+ * Escape and quote a SQL identifier with backticks.
+ *
+ * <p>Internal backticks are doubled to prevent breaking out of the quoted
identifier. The result
+ * is wrapped in backticks so that identifiers containing special characters
(whitespace, dots,
+ * semicolons) are treated as a single identifier token.
+ *
+ * @param identifier the SQL identifier to escape and quote
+ * @return backtick-quoted identifier safe for use in SQL, or null if input
is null
+ */
+ public static String escapeSqlIdentifier(String identifier) {
+ if (identifier == null) {
+ return null;
+ }
+ String escaped = identifier.replace("`", "``");
+ return "`" + escaped + "`";
+ }
+
+ /**
+ * Parse command line arguments in --key value format.
+ *
+ * <p>Supports boolean flags (--flag without a value) by storing them with a
"true" value.
+ *
+ * @param args command line arguments
+ * @return map of argument names to values
+ */
+ public static Map<String, String> parseArguments(String[] args) {
+ Map<String, String> argMap = new HashMap<>();
+
+ for (int i = 0; i < args.length; i++) {
+ if (args[i].startsWith("--")) {
+ String key = args[i].substring(2); // Remove "--" prefix
+
+ // Check if there's a value for this key (not another flag)
+ if (i + 1 < args.length && !args[i + 1].startsWith("--")) {
+ String value = args[i + 1];
+ // Only add non-empty values
+ if (value != null && !value.trim().isEmpty()) {
+ argMap.put(key, value);
+ }
+ i++; // Skip the value in next iteration
+ } else {
+ // Boolean flag with no value - treat as "true"
+ argMap.put(key, "true");
+ }
+ }
+ }
+
+ return argMap;
+ }
+
+ /**
+ * Parse custom Spark configurations from JSON string.
+ *
+ * @param sparkConfJson JSON string containing Spark configurations
+ * @return map of Spark configuration keys to values
+ * @throws IllegalArgumentException if JSON parsing fails
+ */
+ public static Map<String, String> parseCustomSparkConfigs(String
sparkConfJson) {
+ return IcebergSparkConfigUtils.parseFlatJsonMap(sparkConfJson,
"spark-conf");
+ }
+
+ /**
+ * 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");
+ }
+
+ /**
+ * Checks the Iceberg Spark runtime after the session is created, then stops
Spark and exits the
+ * process on failure.
+ *
+ * @param spark Spark session created for this job; stopped if the runtime
check fails
+ */
+ public static void requireIcebergSparkRuntimeOrExit(SparkSession spark) {
+ try {
+ requireIcebergSparkRuntime();
+ } catch (IllegalStateException e) {
+ System.err.println("Error: " + e.getMessage());
+ spark.stop();
+ System.exit(1);
+ return;
+ }
+ }
+
+ /** 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 {
+ // initialize=false: only verify the class is loadable; avoid running
<clinit> here and
+ // misreporting init-time LinkageError as a missing
iceberg-spark-runtime jar.
+ Class.forName(className, false, contextLoader != null ? contextLoader :
fallbackLoader);
+ } catch (ClassNotFoundException | LinkageError first) {
+ if (contextLoader != null && contextLoader != fallbackLoader) {
+ try {
+ Class.forName(className, false, fallbackLoader);
+ return;
+ } catch (ClassNotFoundException | LinkageError ignored) {
+ // Fall through to the user-facing error.
+ }
+ }
+ throw new IllegalStateException(
+ String.format(
+ "Missing %s (%s). Built-in Iceberg jobs need
iceberg-spark-runtime on the Spark "
+ + "classpath (for example via spark.jars in spark_conf, or
installed into the "
+ + "Spark environment). A stock Spark distribution does not
include it. Match "
+ + "the artifact to your Spark, Scala, and Iceberg versions.",
+ description, className),
+ first);
+ }
+ }
+}
diff --git
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteDataFilesJob.java
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteDataFilesJob.java
index c2788281b0..071537e418 100644
---
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteDataFilesJob.java
+++
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteDataFilesJob.java
@@ -18,8 +18,6 @@
*/
package org.apache.gravitino.maintenance.jobs.iceberg;
-import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -160,6 +158,7 @@ public class IcebergRewriteDataFilesJob implements
BuiltInJob {
}
SparkSession spark = sparkBuilder.getOrCreate();
+ IcebergJobUtils.requireIcebergSparkRuntimeOrExit(spark);
try {
// Build the procedure call SQL
@@ -360,30 +359,7 @@ public class IcebergRewriteDataFilesJob implements
BuiltInJob {
* @throws IllegalArgumentException if JSON parsing fails
*/
static Map<String, String> parseCustomSparkConfigs(String sparkConfJson) {
- if (sparkConfJson == null || sparkConfJson.isEmpty()) {
- return new HashMap<>();
- }
-
- try {
- ObjectMapper mapper = new ObjectMapper();
- Map<String, Object> parsedMap =
- mapper.readValue(sparkConfJson, new TypeReference<Map<String,
Object>>() {});
-
- Map<String, String> configs = new HashMap<>();
- for (Map.Entry<String, Object> entry : parsedMap.entrySet()) {
- String key = entry.getKey();
- Object value = entry.getValue();
- configs.put(key, value == null ? "" : value.toString());
- }
- return configs;
- } catch (Exception e) {
- throw new IllegalArgumentException(
- "Failed to parse Spark configurations JSON: "
- + sparkConfJson
- + ". Error: "
- + e.getMessage(),
- e);
- }
+ return IcebergJobUtils.parseCustomSparkConfigs(sparkConfJson);
}
/** Print usage information. */
@@ -426,47 +402,15 @@ public class IcebergRewriteDataFilesJob implements
BuiltInJob {
}
/**
- * Parse options from JSON string using Jackson for robust parsing.
- *
- * <p>Expected format: {"key1": "value1", "key2": "value2"}
+ * Parse options from a flat JSON map.
*
- * <p>This method uses Jackson ObjectMapper to properly handle:
- *
- * <ul>
- * <li>Escaped quotes in values
- * <li>Colons and commas in values
- * <li>Complex JSON structures
- * <li>Various data types (strings, numbers, booleans)
- * </ul>
+ * <p>Parse errors name the real CLI flag {@code --options}. Nested
objects/arrays are rejected.
*
* @param optionsJson JSON string
* @return map of option keys to values
*/
static Map<String, String> parseOptionsJson(String optionsJson) {
- Map<String, String> options = new HashMap<>();
- if (optionsJson == null || optionsJson.isEmpty()) {
- return options;
- }
-
- try {
- ObjectMapper mapper = new ObjectMapper();
- // Parse JSON into a Map<String, Object> to handle various value types
- Map<String, Object> parsedMap =
- mapper.readValue(optionsJson, new TypeReference<Map<String,
Object>>() {});
-
- // Convert all values to strings
- for (Map.Entry<String, Object> entry : parsedMap.entrySet()) {
- String key = entry.getKey();
- Object value = entry.getValue();
- // Convert value to string - handles strings, numbers, booleans, etc.
- options.put(key, value == null ? "" : value.toString());
- }
- } catch (Exception e) {
- throw new IllegalArgumentException(
- "Failed to parse options JSON: " + optionsJson + ". Error: " +
e.getMessage(), e);
- }
-
- return options;
+ return IcebergSparkConfigUtils.parseFlatJsonMap(optionsJson, "options");
}
/**
diff --git
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateStatsAndMetricsJob.java
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateStatsAndMetricsJob.java
index 22a81caf00..7253cb07ec 100644
---
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateStatsAndMetricsJob.java
+++
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateStatsAndMetricsJob.java
@@ -96,7 +96,8 @@ public class IcebergUpdateStatsAndMetricsJob implements
BuiltInJob {
System.exit(1);
}
- Map<String, String> updaterOptions =
parseJsonOptions(argMap.get("updater-options"));
+ Map<String, String> updaterOptions =
+ parseJsonOptions(argMap.get("updater-options"), "updater-options");
String sparkConfJson = argMap.get("spark-conf");
SparkSession.Builder sparkBuilder =
@@ -110,6 +111,7 @@ public class IcebergUpdateStatsAndMetricsJob implements
BuiltInJob {
}
SparkSession spark = sparkBuilder.getOrCreate();
+ IcebergJobUtils.requireIcebergSparkRuntimeOrExit(spark);
StatisticsUpdater statisticsUpdater = null;
MetricsUpdater metricsUpdater = null;
try {
@@ -391,15 +393,22 @@ public class IcebergUpdateStatsAndMetricsJob implements
BuiltInJob {
@VisibleForTesting
static Map<String, String> parseCustomSparkConfigs(String sparkConfJson) {
- return parseJsonOptions(sparkConfJson);
+ return parseJsonOptions(sparkConfJson, "spark-conf");
}
+ /**
+ * Parse a flat JSON option map and report parse errors with the real CLI
flag name.
+ *
+ * @param json JSON string; null or empty yields an empty map
+ * @param optionName CLI flag name without {@code --} (for example {@code
updater-options})
+ * @return parsed flat string map
+ */
@VisibleForTesting
- static Map<String, String> parseJsonOptions(String json) {
+ static Map<String, String> parseJsonOptions(String json, String optionName) {
if (json == null || json.isEmpty()) {
return new HashMap<>();
}
- return IcebergSparkConfigUtils.parseFlatJsonMap(json, "json-options");
+ return IcebergSparkConfigUtils.parseFlatJsonMap(json, optionName);
}
@VisibleForTesting
diff --git
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergJobUtils.java
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergJobUtils.java
new file mode 100644
index 0000000000..5f03f58e3e
--- /dev/null
+++
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergJobUtils.java
@@ -0,0 +1,57 @@
+/*
+ * 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.maintenance.jobs.iceberg;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.junit.jupiter.api.Test;
+
+public class TestIcebergJobUtils {
+
+ @Test
+ public void testRequireIcebergSparkRuntimeSucceedsWhenPresent() {
+ // Test classpath includes iceberg-spark-runtime.
+ assertDoesNotThrow(IcebergJobUtils::requireIcebergSparkRuntime);
+ }
+
+ @Test
+ public void testRequireClassFailsWithActionableMessage() {
+ IllegalStateException ex =
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ IcebergJobUtils.requireClassForTest(
+ "org.apache.gravitino.does.not.ExistIcebergExtension",
+ "Iceberg Spark session extensions"));
+ assertTrue(ex.getMessage().contains("Missing Iceberg Spark session
extensions"));
+ assertTrue(ex.getMessage().contains("iceberg-spark-runtime"));
+ assertTrue(ex.getMessage().contains("spark.jars"));
+ }
+
+ @Test
+ public void testParseCustomSparkConfigsUsesSparkConfFlagName() {
+ IllegalArgumentException ex =
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> IcebergJobUtils.parseCustomSparkConfigs("{not_json}"));
+ assertTrue(ex.getMessage().contains("--spark-conf"));
+ }
+}
diff --git
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteDataFilesJob.java
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteDataFilesJob.java
index 51bb2ba108..c60a689a67 100644
---
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteDataFilesJob.java
+++
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteDataFilesJob.java
@@ -375,7 +375,7 @@ public class TestIcebergRewriteDataFilesJob {
IcebergRewriteDataFilesJob.parseOptionsJson(json);
fail("Expected IllegalArgumentException for invalid JSON");
} catch (IllegalArgumentException e) {
- assertTrue(e.getMessage().contains("Failed to parse options JSON"));
+ assertTrue(e.getMessage().contains("--options"));
}
}
@@ -705,7 +705,7 @@ public class TestIcebergRewriteDataFilesJob {
IcebergRewriteDataFilesJob.parseCustomSparkConfigs(json);
fail("Expected IllegalArgumentException for invalid JSON");
} catch (IllegalArgumentException e) {
- assertTrue(e.getMessage().contains("Failed to parse Spark configurations
JSON"));
+ assertTrue(e.getMessage().contains("--spark-conf"));
}
}
}
diff --git
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateStatsJob.java
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateStatsJob.java
index 3f0080265e..9fd9307e55 100644
---
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateStatsJob.java
+++
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateStatsJob.java
@@ -148,20 +148,43 @@ public class TestIcebergUpdateStatsJob {
public void testParseJsonOptions() {
Map<String, String> parsed =
IcebergUpdateStatsAndMetricsJob.parseJsonOptions(
- "{\"a\":\"b\",\"x\":1,\"flag\":true,\"nil\":null}");
+ "{\"a\":\"b\",\"x\":1,\"flag\":true,\"nil\":null}",
"updater-options");
assertEquals("b", parsed.get("a"));
assertEquals("1", parsed.get("x"));
assertEquals("true", parsed.get("flag"));
assertEquals("", parsed.get("nil"));
- assertThrows(
- IllegalArgumentException.class,
- () -> IcebergUpdateStatsAndMetricsJob.parseJsonOptions("{not_json}"));
- assertThrows(
- IllegalArgumentException.class,
- () ->
IcebergUpdateStatsAndMetricsJob.parseJsonOptions("{\"nested\":{\"a\":1}}"));
- assertThrows(
- IllegalArgumentException.class,
- () ->
IcebergUpdateStatsAndMetricsJob.parseJsonOptions("{\"array\":[1,2,3]}"));
+
+ IllegalArgumentException invalidJson =
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ IcebergUpdateStatsAndMetricsJob.parseJsonOptions("{not_json}",
"updater-options"));
+ assertTrue(invalidJson.getMessage().contains("--updater-options"));
+
+ IllegalArgumentException nested =
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ IcebergUpdateStatsAndMetricsJob.parseJsonOptions(
+ "{\"nested\":{\"a\":1}}", "updater-options"));
+ assertTrue(nested.getMessage().contains("--updater-options"));
+
+ IllegalArgumentException array =
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ IcebergUpdateStatsAndMetricsJob.parseJsonOptions(
+ "{\"array\":[1,2,3]}", "spark-conf"));
+ assertTrue(array.getMessage().contains("--spark-conf"));
+ }
+
+ @Test
+ public void testParseCustomSparkConfigsUsesSparkConfFlagName() {
+ IllegalArgumentException ex =
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
IcebergUpdateStatsAndMetricsJob.parseCustomSparkConfigs("{not_json}"));
+ assertTrue(ex.getMessage().contains("--spark-conf"));
}
@Test
