[SPARK-17742][core] Handle child process exit in SparkLauncher.

Currently the launcher handle does not monitor the child spark-submit
process it launches; this means that if the child exits with an error,
the handle's state will never change, and an application will not know
that the application has failed.

This change adds code to monitor the child process, and changes the
handle state appropriately when the child process exits.

Tested with added unit tests.

core/src/test/java/org/apache/spark/launcher/SparkLauncherSuite.java

@@ -116,11 +116,11 @@ public void testChildProcLauncher() throws Exception {
       .setConf(SparkLauncher.DRIVER_EXTRA_CLASSPATH, System.getProperty("java.class.path"))
       .addSparkArg(opts.CLASS, "ShouldBeOverriddenBelow")
       .setMainClass(SparkLauncherTestApp.class.getName())
+      .redirectError()
       .addAppArgs("proc");
     final Process app = launcher.launch();
 
-    new OutputRedirector(app.getInputStream(), TF);
-    new OutputRedirector(app.getErrorStream(), TF);
+    new OutputRedirector(app.getInputStream(), getClass().getName() + ".child", TF);
     assertEquals(0, app.waitFor());
   }
 

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

@@ -34,7 +34,7 @@ class ChildProcAppHandle implements SparkAppHandle {
   private final String secret;
   private final LauncherServer server;
 
-  private Process childProc;
+  private volatile Process childProc;
   private boolean disposed;
   private LauncherConnection connection;
   private List<Listener> listeners;
@@ -96,9 +96,7 @@ public synchronized void disconnect() {
 
   @Override
   public synchronized void kill() {
-    if (!disposed) {
-      disconnect();
-    }
+    disconnect();
     if (childProc != null) {
       try {
         childProc.exitValue();
@@ -118,14 +116,40 @@ void setChildProc(Process childProc, String loggerName, InputStream logStream) {
     this.childProc = childProc;
     if (logStream != null) {
       this.redirector = new OutputRedirector(logStream, loggerName,
-        SparkLauncher.REDIRECTOR_FACTORY);
+        SparkLauncher.REDIRECTOR_FACTORY, this);
+    } else {
+      // If there is no log redirection, spawn a thread that will wait for the child process
+      // to finish.
+      Thread waiter = SparkLauncher.REDIRECTOR_FACTORY.newThread(this::monitorChild);
+      waiter.setDaemon(true);
+      waiter.start();
     }
   }
 
   void setConnection(LauncherConnection connection) {
     this.connection = connection;
   }
 
+  /**
+   * Callback for when the child process exits. Forcefully put the application in a final state,
+   * overwriting the current final state unless it is already FAILED.
+   */
+  synchronized void childProcessExited() {
+    disconnect();
+
+    int ec;
+    try {
+      ec = childProc.exitValue();
+    } catch (Exception e) {
+      ec = 1;
+    }
+
+    if (!state.isFinal() || (ec != 0 && state != State.FAILED)) {
+      state = State.LOST;
+      fireEvent(false);
+    }
+  }
+
   LauncherServer getServer() {
     return server;
   }
@@ -134,7 +158,7 @@ LauncherConnection getConnection() {
     return connection;
   }
 
-  void setState(State s) {
+  synchronized void setState(State s) {
     if (!state.isFinal()) {
       state = s;
       fireEvent(false);
@@ -144,17 +168,12 @@ void setState(State s) {
     }
   }
 
-  void setAppId(String appId) {
+  synchronized void setAppId(String appId) {
     this.appId = appId;
     fireEvent(true);
   }
 
-  // Visible for testing.
-  boolean isRunning() {
-    return childProc == null || childProc.isAlive() || (redirector != null && redirector.isAlive());
-  }
-
-  private synchronized void fireEvent(boolean isInfoChanged) {
+  private void fireEvent(boolean isInfoChanged) {
     if (listeners != null) {
       for (Listener l : listeners) {
         if (isInfoChanged) {
@@ -166,4 +185,15 @@ private synchronized void fireEvent(boolean isInfoChanged) {
     }
   }
 
+  private void monitorChild() {
+    while (childProc.isAlive()) {
+      try {
+        childProc.waitFor();
+      } catch (Exception e) {
+        // Try again.
+      }
+    }
+    childProcessExited();
+  }
+
 }

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

@@ -34,18 +34,24 @@ class OutputRedirector {
   private final BufferedReader reader;
   private final Logger sink;
   private final Thread thread;
+  private final ChildProcAppHandle callback;
 
   private volatile boolean active;
 
-  OutputRedirector(InputStream in, ThreadFactory tf) {
-    this(in, OutputRedirector.class.getName(), tf);
+  OutputRedirector(InputStream in, String loggerName, ThreadFactory tf) {
+    this(in, loggerName, tf, null);
   }
 
-  OutputRedirector(InputStream in, String loggerName, ThreadFactory tf) {
+  OutputRedirector(
+      InputStream in,
+      String loggerName,
+      ThreadFactory tf,
+      ChildProcAppHandle callback) {
     this.active = true;
     this.reader = new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8));
     this.thread = tf.newThread(this::redirect);
     this.sink = Logger.getLogger(loggerName);
+    this.callback = callback;
     thread.start();
   }
 
@@ -59,6 +65,10 @@ private void redirect() {
       }
     } catch (IOException e) {
       sink.log(Level.FINE, "Error reading child process output.", e);
+    } finally {
+      if (callback != null) {
+        callback.childProcessExited();
+      }
     }
   }
 

launcher/src/test/java/org/apache/spark/launcher/OutputRedirectionSuite.java → launcher/src/test/java/org/apache/spark/launcher/ChildProcAppHandleSuite.java

@@ -25,6 +25,7 @@
 import java.util.Arrays;
 import java.util.EnumSet;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import static java.nio.file.attribute.PosixFilePermission.*;
 
@@ -39,7 +40,7 @@
 
 import static org.apache.spark.launcher.CommandBuilderUtils.*;
 
-public class OutputRedirectionSuite extends BaseSuite {
+public class ChildProcAppHandleSuite extends BaseSuite {
 
   private static final List<String> MESSAGES = new ArrayList<>();
 
@@ -99,7 +100,8 @@ public void testRedirectLastWins() throws Exception {
   public void testRedirectToLog() throws Exception {
     assumeFalse(isWindows());
 
-    ChildProcAppHandle handle = (ChildProcAppHandle) new TestSparkLauncher().startApplication();
+    SparkAppHandle handle = (ChildProcAppHandle) new TestSparkLauncher()
+      .startApplication();
     waitFor(handle);
 
     assertTrue(MESSAGES.contains("output"));
@@ -112,7 +114,7 @@ public void testRedirectErrorToLog() throws Exception {
 
     Path err = Files.createTempFile("stderr", "txt");
 
-    ChildProcAppHandle handle = (ChildProcAppHandle) new TestSparkLauncher()
+    SparkAppHandle handle = (ChildProcAppHandle) new TestSparkLauncher()
       .redirectError(err.toFile())
       .startApplication();
     waitFor(handle);
@@ -127,7 +129,7 @@ public void testRedirectOutputToLog() throws Exception {
 
     Path out = Files.createTempFile("stdout", "txt");
 
-    ChildProcAppHandle handle = (ChildProcAppHandle) new TestSparkLauncher()
+    SparkAppHandle handle = (ChildProcAppHandle) new TestSparkLauncher()
       .redirectOutput(out.toFile())
       .startApplication();
     waitFor(handle);
@@ -173,17 +175,37 @@ public void testRedirectErrorTwiceFails() throws Exception {
       .waitFor();
   }
 
-  private void waitFor(ChildProcAppHandle handle) throws Exception {
+  @Test
+  public void testProcMonitorWithOutputRedirection() throws Exception {
+    File err = Files.createTempFile("out", "txt").toFile();
+    SparkAppHandle handle = new TestSparkLauncher()
+      .redirectError()
+      .redirectOutput(err)
+      .startApplication();
+    waitFor(handle);
+    assertEquals(SparkAppHandle.State.LOST, handle.getState());
+  }
+
+  @Test
+  public void testProcMonitorWithLogRedirection() throws Exception {
+    SparkAppHandle handle = new TestSparkLauncher()
+      .redirectToLog(getClass().getName())
+      .startApplication();
+    waitFor(handle);
+    assertEquals(SparkAppHandle.State.LOST, handle.getState());
+  }
+
+  private void waitFor(SparkAppHandle handle) throws Exception {
+    long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
     try {
-      while (handle.isRunning()) {
-        Thread.sleep(10);
+      while (!handle.getState().isFinal()) {
+        assertTrue("Timed out waiting for handle to transition to final state.",
+          System.nanoTime() < deadline);
+        TimeUnit.MILLISECONDS.sleep(10);
       }
     } finally {
-      // Explicit unregister from server since the handle doesn't yet do that when the
-      // process finishes by itself.
-      LauncherServer server = LauncherServer.getServerInstance();
-      if (server != null) {
-        server.unregister(handle);
+      if (!handle.getState().isFinal()) {
+        handle.kill();
       }
     }
   }

launcher/src/test/resources/log4j.properties

@@ -29,7 +29,7 @@ log4j.appender.childproc.target=System.err
 log4j.appender.childproc.layout=org.apache.log4j.PatternLayout
 log4j.appender.childproc.layout.ConversionPattern=%t: %m%n
 
-log4j.appender.outputredirtest=org.apache.spark.launcher.OutputRedirectionSuite$LogAppender
+log4j.appender.outputredirtest=org.apache.spark.launcher.ChildProcAppHandleSuite$LogAppender
 log4j.logger.org.apache.spark.launcher.app.outputredirtest=INFO, outputredirtest
 log4j.logger.org.apache.spark.launcher.app.outputredirtest.additivity=false