[SPARK-14702] Make environment of SparkLauncher launched process more configurable

## What changes were proposed in this pull request?

Adds a few public methods to `SparkLauncher` to allow configuring some extra features of the `ProcessBuilder`, including the working directory, output and error stream redirection.

## How was this patch tested?

Unit testing + simple Spark driver programs

Author: Andrew Duffy <root@aduffy.org>

Closes #14201 from andreweduffy/feature/launcher.

1 files changed, 63 insertions, 4 deletions

diff --git a/core/src/test/java/org/apache/spark/launcher/SparkLauncherSuite.java b/core/src/test/java/org/apache/spark/launcher/SparkLauncherSuite.java
index 8ca54b24d8..e393db06a0 100644
--- a/core/src/test/java/org/apache/spark/launcher/SparkLauncherSuite.java
+++ b/core/src/test/java/org/apache/spark/launcher/SparkLauncherSuite.java
@@ -21,6 +21,7 @@ import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
 
+import org.junit.Before;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -40,10 +41,15 @@ public class SparkLauncherSuite {
   private static final Logger LOG = LoggerFactory.getLogger(SparkLauncherSuite.class);
   private static final NamedThreadFactory TF = new NamedThreadFactory("SparkLauncherSuite-%d");
 
+  private SparkLauncher launcher;
+
+  @Before
+  public void configureLauncher() {
+    launcher = new SparkLauncher().setSparkHome(System.getProperty("spark.test.home"));
+  }
+
   @Test
   public void testSparkArgumentHandling() throws Exception {
-    SparkLauncher launcher = new SparkLauncher()
-      .setSparkHome(System.getProperty("spark.test.home"));
     SparkSubmitOptionParser opts = new SparkSubmitOptionParser();
 
     launcher.addSparkArg(opts.HELP);
@@ -85,14 +91,67 @@ public class SparkLauncherSuite {
     assertEquals("bar", launcher.builder.conf.get("spark.foo"));
   }
 
+  @Test(expected=IllegalStateException.class)
+  public void testRedirectTwiceFails() throws Exception {
+    launcher.setAppResource("fake-resource.jar")
+      .setMainClass("my.fake.class.Fake")
+      .redirectError()
+      .redirectError(ProcessBuilder.Redirect.PIPE)
+      .launch();
+  }
+
+  @Test(expected=IllegalStateException.class)
+  public void testRedirectToLogWithOthersFails() throws Exception {
+    launcher.setAppResource("fake-resource.jar")
+      .setMainClass("my.fake.class.Fake")
+      .redirectToLog("fakeLog")
+      .redirectError(ProcessBuilder.Redirect.PIPE)
+      .launch();
+  }
+
+  @Test
+  public void testRedirectErrorToOutput() throws Exception {
+    launcher.redirectError();
+    assertTrue(launcher.redirectErrorStream);
+  }
+
+  @Test
+  public void testRedirectsSimple() throws Exception {
+    launcher.redirectError(ProcessBuilder.Redirect.PIPE);
+    assertNotNull(launcher.errorStream);
+    assertEquals(launcher.errorStream.type(), ProcessBuilder.Redirect.Type.PIPE);
+
+    launcher.redirectOutput(ProcessBuilder.Redirect.PIPE);
+    assertNotNull(launcher.outputStream);
+    assertEquals(launcher.outputStream.type(), ProcessBuilder.Redirect.Type.PIPE);
+  }
+
+  @Test
+  public void testRedirectLastWins() throws Exception {
+    launcher.redirectError(ProcessBuilder.Redirect.PIPE)
+      .redirectError(ProcessBuilder.Redirect.INHERIT);
+    assertEquals(launcher.errorStream.type(), ProcessBuilder.Redirect.Type.INHERIT);
+
+    launcher.redirectOutput(ProcessBuilder.Redirect.PIPE)
+      .redirectOutput(ProcessBuilder.Redirect.INHERIT);
+    assertEquals(launcher.outputStream.type(), ProcessBuilder.Redirect.Type.INHERIT);
+  }
+
+  @Test
+  public void testRedirectToLog() throws Exception {
+    launcher.redirectToLog("fakeLogger");
+    assertTrue(launcher.redirectToLog);
+    assertTrue(launcher.builder.getEffectiveConfig()
+      .containsKey(SparkLauncher.CHILD_PROCESS_LOGGER_NAME));
+  }
+
   @Test
   public void testChildProcLauncher() throws Exception {
     SparkSubmitOptionParser opts = new SparkSubmitOptionParser();
     Map<String, String> env = new HashMap<>();
     env.put("SPARK_PRINT_LAUNCH_COMMAND", "1");
 
-    SparkLauncher launcher = new SparkLauncher(env)
-      .setSparkHome(System.getProperty("spark.test.home"))
+    launcher
       .setMaster("local")
       .setAppResource(SparkLauncher.NO_RESOURCE)
       .addSparkArg(opts.CONF,