lasdf1234 commented on code in PR #13134:
URL: https://github.com/apache/gravitino/pull/13134#discussion_r4013880111
##########
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:
Thank you very much for your review.
It was changed to IcebergJobUtils.requireIcebergSparkRuntimeOrExit(spark),
and the main of all three jobs were all modified to call this line; in case of
failure, it still prints the error, spark.stop(), and exit(1). From now on,
only this part will be modified.
##########
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:
Thank you very much for your review.
I have made the necessary changes according to your suggestions, and it has
been updated to Class.forName(... ) , false, ...) Just check that the class can
be loaded, without running the static initialization.
##########
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:
Thank you very much for your review.
Move `requireClassForTest` / `requireClass` to the end of the class, leaving
only the public methods at the beginning, which is in line with the principle
of "putting private members at the end".
##########
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:
Thank you very much for your review.
Shared flags (catalog / table / spark-conf) receive IcebergJobUtils;
Expire/Rewrite/UpdateStats read parameters, errors, and templates all use the
same constant, no longer manually writing strings while using constants.
##########
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:
Thank you very much for your review.
Configs has adopted the same "omit" approach as environments: the
{{spark_executor_instances}} that is not passed will be discarded and will no
longer be included exactly as it is in --conf.
--
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]
