yuqi1129 commented on code in PR #13982:
URL: https://github.com/apache/gravitino/pull/13982#discussion_r4374949849
##########
maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergJobUtils.java:
##########
@@ -173,16 +202,69 @@ public static Map<String, String> parseArguments(
* @throws IllegalArgumentException if JSON parsing fails
*/
public static Map<String, String> parseCustomSparkConfigs(String
sparkConfJson) {
- return IcebergSparkConfigUtils.parseFlatJsonMap(sparkConfJson,
"spark-conf");
+ return filterCustomSparkConfigs(
+ IcebergSparkConfigUtils.parseFlatJsonMap(sparkConfJson, "spark-conf"));
+ }
+
+ /**
+ * Filter the custom Spark configurations of a job, which the driver applies
when it creates the
+ * SparkSession, against the configurations spark-submit has already set for
the driver.
+ *
+ * <p>{@code spark.master} and {@code spark.submit.deployMode} are dropped
if spark-submit has set
+ * them, as the job would otherwise run with another master than it was
submitted to. The {@code
+ * spark.kubernetes.*} configurations are dropped, as whoever runs a job
sets them here, while the
+ * driver creates the executor pods from them. A blank value leaves a
configuration unset.
+ *
+ * @param configs the custom Spark configurations
+ * @return the custom Spark configurations to apply
+ */
+ public static Map<String, String> filterCustomSparkConfigs(Map<String,
String> configs) {
+ // new SparkConf() reads the spark.* system properties on purpose:
spark-submit sets the
+ // configurations it submitted the job with as system properties before it
calls main().
+ return filterCustomSparkConfigs(configs, new SparkConf());
+ }
+
+ @VisibleForTesting
+ static Map<String, String> filterCustomSparkConfigs(
+ Map<String, String> configs, SparkConf submittedConf) {
+ Map<String, String> filtered = new LinkedHashMap<>();
+ configs.forEach(
+ (key, value) -> {
+ // A blank value leaves a configuration unset, as the k8s job
executor does.
+ if (value == null || value.trim().isEmpty()) {
Review Comment:
[Nit] Could we use the same blank-value predicate here as in
`SparkApplicationUtils.getJobSparkConf()`? The submit side uses
`StringUtils.isBlank(value)`, while `trim().isEmpty()` here does not recognize
all the same whitespace characters. For example, with `spark_conf` containing
`{"spark.executor.memory":"\u2003"}`, the k8s executor skips the value and
retains the template/executor default, but the driver keeps it and applies it
again when building the SparkSession. This makes the two stages disagree on the
documented “blank means unset” rule. A consistent predicate and a
Unicode-whitespace regression case would close this small edge case.
Verified with a temporary JUnit probe against `filterCustomSparkConfigs(...,
new SparkConf(false))`: the U+2003 value remains in the returned map instead of
being omitted. I did not reproduce the subsequent behavior on a real Kubernetes
cluster. This is non-blocking; ordinary empty strings and ASCII whitespace
behave consistently.
--
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]
