[SPARK-53176][DEPLOY] Spark launcher should respect --load-spark-defaults

### What changes were proposed in this pull request?

SPARK-48392 introduces `--load-spark-defaults`, but does not apply correctly for the Spark launcher process, this mainly affects the driver when Spark runs in local/client mode.

let's say we have
```
$ cat > conf/spark-defaults.conf <<EOF
spark.driver.memory=4g
EOF
$ cat > conf/spark-local.conf <<EOF
spark.master=local[4]
EOF
```

```
$ bin/spark-shell --properties-file conf/spark-local.conf --load-spark-defaults
...
scala> spark.sql("SET spark.driver.memory").show()
+-------------------+-----+
|                key|value|
+-------------------+-----+
|spark.driver.memory|   4g|
+-------------------+-----+
```
even the spark conf reports that driver uses 4GiB heap memory, but if we check the Java process, the config actually does not take effect, the default 1GiB is used instead.
```
$ jinfo <spark-submit-pid>
...
VM Arguments:
jvm_args: -Dscala.usejavacp=true -Xmx1g ...
```

### Why are the changes needed?

Bug fix.

### Does this PR introduce _any_ user-facing change?

Yes, bug fix.

### How was this patch tested?

UT is modified to cover the change, plus manual tests for the above cases.

```
$ jinfo <spark-submit-pid>
...
VM Arguments:
jvm_args: -Dscala.usejavacp=true -Xmx4g ...
```

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #51905 from pan3793/SPARK-53176.

Authored-by: Cheng Pan <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
(cherry picked from commit b82957c)
Signed-off-by: Liang-Chi Hsieh <[email protected]>

Author: pan3793

launcher/src/main/java/org/apache/spark/launcher/AbstractCommandBuilder.java

@@ -49,6 +49,7 @@ abstract class AbstractCommandBuilder {
   String master;
   String remote;
   protected String propertiesFile;
+  protected boolean loadSparkDefaults;
   final List<String> appArgs;
   final List<String> jars;
   final List<String> files;
@@ -362,21 +363,35 @@ Map<String, String> getEffectiveConfig() throws IOException {
   }
 
   /**
-   * Loads the configuration file for the application, if it exists. This is either the
-   * user-specified properties file, or the spark-defaults.conf file under the Spark configuration
-   * directory.
+   * Load the configuration file(s) for the application - from the user-specified properties
+   * file, and/or the spark-defaults.conf file under the Spark configuration directory, if exists.
+   * Configurations from user-specified properties file take precedence over spark-defaults.conf.
    */
   private Properties loadPropertiesFile() throws IOException {
     Properties props = new Properties();
-    File propsFile;
     if (propertiesFile != null) {
-      propsFile = new File(propertiesFile);
+      File propsFile = new File(propertiesFile);
       checkArgument(propsFile.isFile(), "Invalid properties file '%s'.", propertiesFile);
-    } else {
-      propsFile = new File(getConfDir(), DEFAULT_PROPERTIES_FILE);
+      props = loadPropertiesFile(propsFile);
     }
 
-    if (propsFile.isFile()) {
+    Properties defaultsProps = new Properties();
+    if (propertiesFile == null || loadSparkDefaults) {
+      defaultsProps = loadPropertiesFile(new File(getConfDir(), DEFAULT_PROPERTIES_FILE));
+    }
+
+    for (Map.Entry<Object, Object> entry : defaultsProps.entrySet()) {
+      if (!props.containsKey(entry.getKey())) {
+        props.put(entry.getKey(), entry.getValue());
+      }
+    }
+
+    return props;
+  }
+
+  private Properties loadPropertiesFile(File propsFile) throws IOException {
+    Properties props = new Properties();
+    if (propsFile != null && propsFile.isFile()) {
       try (InputStreamReader isr = new InputStreamReader(
           new FileInputStream(propsFile), StandardCharsets.UTF_8)) {
         props.load(isr);

launcher/src/main/java/org/apache/spark/launcher/SparkSubmitCommandBuilder.java

@@ -251,6 +251,10 @@ List<String> buildSparkSubmitArgs(boolean includeRemote) {
       args.add(propertiesFile);
     }
 
+    if (loadSparkDefaults) {
+      args.add(parser.LOAD_SPARK_DEFAULTS);
+    }
+
     if (isExample) {
       jars.addAll(findExamplesJars());
     }
@@ -550,6 +554,7 @@ protected boolean handle(String opt, String value) {
         }
         case DEPLOY_MODE -> deployMode = value;
         case PROPERTIES_FILE -> propertiesFile = value;
+        case LOAD_SPARK_DEFAULTS -> loadSparkDefaults = true;
         case DRIVER_MEMORY -> conf.put(SparkLauncher.DRIVER_MEMORY, value);
         case DRIVER_JAVA_OPTIONS -> conf.put(SparkLauncher.DRIVER_EXTRA_JAVA_OPTIONS, value);
         case DRIVER_LIBRARY_PATH -> conf.put(SparkLauncher.DRIVER_EXTRA_LIBRARY_PATH, value);

launcher/src/test/java/org/apache/spark/launcher/SparkSubmitCommandBuilderSuite.java

@@ -38,20 +38,91 @@ public class SparkSubmitCommandBuilderSuite extends BaseSuite {
 
   private static File dummyPropsFile;
   private static File connectPropsFile;
+  private static File driverMemPropsFile;
   private static SparkSubmitOptionParser parser;
 
   @BeforeAll
   public static void setUp() throws Exception {
     dummyPropsFile = File.createTempFile("spark", "properties");
     connectPropsFile = File.createTempFile("spark", "properties");
     Files.writeString(connectPropsFile.toPath(), "spark.remote=sc://connect-server:15002");
+    driverMemPropsFile = File.createTempFile("spark", "properties");
+    Files.writeString(driverMemPropsFile.toPath(),
+        "spark.driver.memory=4g\nspark.driver.memoryOverhead=768m");
     parser = new SparkSubmitOptionParser();
   }
 
   @AfterAll
   public static void cleanUp() throws Exception {
     dummyPropsFile.delete();
     connectPropsFile.delete();
+    driverMemPropsFile.delete();
+  }
+
+  @Test
+  public void testGetEffectiveConfig() throws Exception {
+    doTestGetEffectiveConfig(null, true, true);
+    doTestGetEffectiveConfig(null, true, false);
+    doTestGetEffectiveConfig(null, false, true);
+    doTestGetEffectiveConfig(null, false, false);
+    doTestGetEffectiveConfig(driverMemPropsFile, true, true);
+    doTestGetEffectiveConfig(driverMemPropsFile, true, false);
+    doTestGetEffectiveConfig(driverMemPropsFile, false, true);
+    doTestGetEffectiveConfig(driverMemPropsFile, false, false);
+  }
+
+  private void doTestGetEffectiveConfig(
+      File propertiesFile, boolean loadSparkDefaults, boolean confDriverMemory) throws Exception {
+    SparkSubmitCommandBuilder launcher =
+        newCommandBuilder(Collections.emptyList());
+    launcher.loadSparkDefaults = loadSparkDefaults;
+    launcher.conf.put("spark.foo", "bar");
+    launcher.childEnv.put("SPARK_CONF_DIR", System.getProperty("spark.test.home")
+        + "/launcher/src/test/resources");
+
+    if (propertiesFile != null) {
+      launcher.setPropertiesFile(propertiesFile.getAbsolutePath());
+    }
+
+    if (confDriverMemory) {
+      launcher.conf.put(SparkLauncher.DRIVER_MEMORY, "2g");
+    }
+
+    Map<String, String> effectiveConfig = launcher.getEffectiveConfig();
+
+    assertEquals("bar", effectiveConfig.get("spark.foo"));
+    if (confDriverMemory) {
+      assertEquals("2g", effectiveConfig.get(SparkLauncher.DRIVER_MEMORY));
+    } else if (propertiesFile != null) {
+      try (FileReader reader = new FileReader(propertiesFile, StandardCharsets.UTF_8)) {
+        Properties props = new Properties();
+        props.load(reader);
+        if (props.containsKey(SparkLauncher.DRIVER_MEMORY)) {
+          assertEquals(props.getProperty(SparkLauncher.DRIVER_MEMORY),
+            effectiveConfig.get(SparkLauncher.DRIVER_MEMORY));
+        }
+      }
+    } else {
+      assertEquals("1g", effectiveConfig.get(SparkLauncher.DRIVER_MEMORY));
+    }
+
+    if (propertiesFile != null) {
+      try (FileReader reader = new FileReader(propertiesFile, StandardCharsets.UTF_8)) {
+        Properties props = new Properties();
+        props.load(reader);
+        if (props.containsKey("spark.driver.memoryOverhead")) {
+          assertEquals(props.getProperty("spark.driver.memoryOverhead"),
+              effectiveConfig.get("spark.driver.memoryOverhead"));
+        }
+      }
+      if (loadSparkDefaults) {
+        assertEquals("/driver", effectiveConfig.get(SparkLauncher.DRIVER_EXTRA_CLASSPATH));
+      } else {
+        assertFalse(effectiveConfig.containsKey(SparkLauncher.DRIVER_EXTRA_CLASSPATH));
+      }
+    } else {
+      assertEquals("/driver", effectiveConfig.get(SparkLauncher.DRIVER_EXTRA_CLASSPATH));
+    }
   }
 
   @Test