This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 35c184d2de [#13133] fix(maintenance): Fail Iceberg jobs when Spark
runtime is missing (#13134)
35c184d2de is described below
commit 35c184d2dec23b193aac72191ddd7560c31db828
Author: MaSai <[email protected]>
AuthorDate: Thu Sep 17 14:28:29 2026 +0800
[#13133] fix(maintenance): Fail Iceberg jobs when Spark runtime is missing
(#13134)
### What changes were proposed in this pull request?
Addresses #13133:
1. **Fail when Iceberg Spark runtime is missing**
Built-in Iceberg jobs check for `IcebergSparkSessionExtensions` /
`SparkCatalog` after `SparkSession` creation and exit non-zero when
absent.
2. **Document the Iceberg Spark runtime prerequisite**
Optimizer docs state that operators must supply a matching
`iceberg-spark-runtime` (for example via `spark.jars` / Spark env), and
troubleshooting reflects fail-fast behavior.
3. **Clearer optional-arg / JSON errors for callers**
Iceberg JSON parse failures name the real CLI flags
(`--updater-options`, `--spark-conf`, `--options`). Docs note that
omitting optional `jobConf` keys can leave dangling `--flag`
placeholders, so callers/UIs should send explicit values.
### Why are the changes needed?
Without Iceberg Spark runtime, Spark only warns and jobs continue as if
Iceberg were available. Docs did not make the runtime prerequisite
discoverable. Optional template placeholders and opaque JSON parse
errors also made failed submissions harder to diagnose.
Fix: #13133
### Does this PR introduce _any_ user-facing change?
- Built-in Iceberg jobs now exit non-zero when Iceberg Spark runtime
classes are missing.
- Docs clarify the `iceberg-spark-runtime` prerequisite (`spark.jars` /
Spark env), fail-fast behavior, and dangling optional template flags.
- JSON option parse errors name the real `--flag` names.
### How was this patch tested?
```bash
./gradlew :maintenance:jobs:test \
--tests org.apache.gravitino.maintenance.jobs.iceberg.TestIcebergJobUtils
\
--tests
org.apache.gravitino.maintenance.jobs.iceberg.TestIcebergUpdateStatsJob \
--tests
org.apache.gravitino.maintenance.jobs.iceberg.TestIcebergRewriteDataFilesJob \
--tests
org.apache.gravitino.maintenance.jobs.iceberg.TestIcebergExpireSnapshotsJob \
-PskipITs -PskipDockerTests=true
```
---------
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 +++
.../jobs/iceberg/IcebergExpireSnapshotsJob.java | 1 +
.../maintenance/jobs/iceberg/IcebergJobUtils.java | 90 ++++++++++++++++------
.../jobs/iceberg/IcebergRewriteDataFilesJob.java | 42 +---------
.../iceberg/IcebergUpdateStatsAndMetricsJob.java | 17 +++-
.../iceberg/TestIcebergExpireSnapshotsJob.java | 2 +-
.../jobs/iceberg/TestIcebergJobUtils.java | 57 ++++++++++++++
.../iceberg/TestIcebergRewriteDataFilesJob.java | 4 +-
.../jobs/iceberg/TestIcebergUpdateStatsJob.java | 43 ++++++++---
12 files changed, 233 insertions(+), 78 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/IcebergExpireSnapshotsJob.java
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergExpireSnapshotsJob.java
index a7495dae64..19b9ccf6dd 100644
---
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergExpireSnapshotsJob.java
+++
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergExpireSnapshotsJob.java
@@ -148,6 +148,7 @@ public class IcebergExpireSnapshotsJob implements
BuiltInJob {
}
SparkSession spark = sparkBuilder.getOrCreate();
+ IcebergJobUtils.requireIcebergSparkRuntimeOrExit(spark);
try {
// Build the procedure call SQL
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
index 19cdc2f051..9f95e92405 100644
---
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
@@ -18,19 +18,21 @@
*/
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;
+import org.apache.spark.sql.SparkSession;
/**
* 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 IcebergJobUtils() {}
/**
@@ -105,29 +107,71 @@ public final class IcebergJobUtils {
* @throws IllegalArgumentException if JSON parsing fails
*/
public static Map<String, String> parseCustomSparkConfigs(String
sparkConfJson) {
- if (sparkConfJson == null || sparkConfJson.isEmpty()) {
- return new HashMap<>();
- }
+ 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 {
- ObjectMapper mapper = new ObjectMapper();
- Map<String, Object> parsedMap =
- mapper.readValue(sparkConfJson, new TypeReference<Map<String,
Object>>() {});
+ requireIcebergSparkRuntime();
+ } catch (IllegalStateException e) {
+ System.err.println("Error: " + e.getMessage());
+ spark.stop();
+ System.exit(1);
+ return;
+ }
+ }
- 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());
+ /** 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.
+ }
}
- return configs;
- } catch (Exception e) {
- throw new IllegalArgumentException(
- "Failed to parse Spark configurations JSON: "
- + sparkConfJson
- + ". Error: "
- + e.getMessage(),
- e);
+ 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 6a34c48161..c3e3453bea 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,11 +18,8 @@
*/
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;
import java.util.List;
import java.util.Map;
import org.apache.gravitino.job.JobTemplateProvider;
@@ -160,6 +157,7 @@ public class IcebergRewriteDataFilesJob implements
BuiltInJob {
}
SparkSession spark = sparkBuilder.getOrCreate();
+ IcebergJobUtils.requireIcebergSparkRuntimeOrExit(spark);
try {
// Build the procedure call SQL
@@ -357,47 +355,15 @@ public class IcebergRewriteDataFilesJob implements
BuiltInJob {
}
/**
- * Parse options from JSON string using Jackson for robust parsing.
+ * Parse options from a flat JSON map.
*
- * <p>Expected format: {"key1": "value1", "key2": "value2"}
- *
- * <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/TestIcebergExpireSnapshotsJob.java
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergExpireSnapshotsJob.java
index 8650e6425e..90f71478b4 100644
---
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergExpireSnapshotsJob.java
+++
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergExpireSnapshotsJob.java
@@ -471,7 +471,7 @@ public class TestIcebergExpireSnapshotsJob {
IcebergJobUtils.parseCustomSparkConfigs("{invalid 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/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 20e5f98a80..9f99e673ed 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"));
}
}
@@ -706,7 +706,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
