[SPARK-17443][SPARK-11035] Stop Spark Application if launcher goes down and use reflection #15009

kishorvpatil · 2016-09-08T03:34:34Z

What changes were proposed in this pull request?

SparkLauncher provides stopIfInterrupted method to allow ensuring SparkApplication stops if tracking process dies.
SparkLauncher provides startApplicationInProcess method to start application using reflection in another Thread instead of launching childProcess and relying on spark-submit binary.

(Please fill in changes proposed in this fix)

How was this patch tested?

I ran following Manual tests:

startApplication with request to stopIfInterrupted, then send interrupt signal to the process.
Result: The childprocess kills Spark Application and terminates.
startApplicationInProcess with request to stopIfInterrupted, then send interrupt signal to the process.
Result: The shutdownHook in LauncherBackend kills Spark Application and before termination.
startApplication without stopIfInterrupted, then send interrupt signal to the process.
Result: The childprocess terminates without killing the Spark Application.
startApplicationInProcess without stopIfInterrupted, then send interrupt signal to the process.
Result: The process terminates without killing Spark Application.

tgravescs · 2016-09-08T13:26:18Z

Jenkins, test this please

tgravescs · 2016-09-08T13:29:02Z

core/src/main/scala/org/apache/spark/launcher/LauncherBackend.scala

+  }
+
+  def connect(port: Int, secret: String): Unit = {
+    if (port != None && secret != None) {


doing the same check here as line 46 above, don't need both.

SparkQA · 2016-09-08T13:29:26Z

Test build #65103 has finished for PR 15009 at commit 8986aa9.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- public abstract class AbstractSparkAppHandle implements SparkAppHandle
- class ChildProcAppHandle extends AbstractSparkAppHandle
- public class ChildThreadAppHandle extends AbstractSparkAppHandle
- class SparkSubmitRunner implements Runnable

tgravescs · 2016-09-08T13:31:16Z

core/src/main/scala/org/apache/spark/launcher/LauncherBackend.scala

    if (port != None && secret != None) {
-      val s = new Socket(InetAddress.getLoopbackAddress(), port.get)
+      if(stopFlag != None) {
+        connect(port.get, secret.get, stopFlag.get)


can't we just set this.stopFlag here and have one connect method since that is all this one is doing?

ignore this I didn't see it being called from Client

tgravescs · 2016-09-08T13:53:34Z

@kishorvpatil Can you please describe in more detail what manual testing you did?

tgravescs · 2016-09-08T13:53:54Z

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

+  private String appId;
+
+  OutputRedirector redirector;
+


remove extra blank newlines

tgravescs · 2016-09-08T17:41:47Z

Jenkins, test this please

tgravescs · 2016-09-08T17:42:06Z

made a few comments but haven't made it all the way through yet, hopefully later this afternoon

SparkQA · 2016-09-08T17:48:51Z

Test build #65108 has finished for PR 15009 at commit 99b1d1b.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

tgravescs · 2016-09-09T14:01:00Z

Jenkins, this is okay to test

tgravescs · 2016-09-09T14:12:32Z

Jenkins, test this please

tgravescs · 2016-09-09T14:14:06Z

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

+    if (!disposed) {
+      disconnect();
+    }
+    if (childThread!= null) {


space before !=

SparkQA · 2016-09-09T14:26:13Z

Test build #65150 has finished for PR 15009 at commit bac5cd2.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

tgravescs · 2016-09-09T14:26:46Z

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

+        childThread = null;
+      }
+    }
+


remove extra new line

tgravescs · 2016-09-09T19:52:45Z

Jenkins, test this please

SparkQA · 2016-09-09T22:12:02Z

Test build #65166 has finished for PR 15009 at commit 67f8de7.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

tgravescs · 2016-09-12T13:58:12Z

core/src/main/scala/org/apache/spark/launcher/LauncherBackend.scala

@@ -71,6 +104,9 @@ private[spark] abstract class LauncherBackend {
    if (connection != null && lastState != state) {
      connection.send(new SetState(state))
      lastState = state
+      if (!_isConnected && stopFlag) {


do we still need this if its being done in close() now?

kishorvpatil · 2017-05-17T15:18:18Z

@vanzin I modified the test with sleep for SparkApp to run.

vanzin · 2017-05-19T18:42:47Z

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

+      .addAppArgs("thread");
+    final SparkAppHandle app = launcher.startApplication();
+    sleep(3000);
+    assertEquals(false, app.getState().isFinal());


You still haven't fixed this. This is what I wrote the very first time I commented on this:

Hmm... you need to wait for the app to finish, or kill it eventually. And check that it actually ran.

You are not waiting for the app to finish. You're checking that app.getState().isFinal() is false here, so basically this test waits 3 seconds and still expects the app to be running after that.

You want exactly the opposite. You want the app to finish at some point, and not just finish, but finish successfully. That means checking for a specific status.

SparkQA · 2017-05-23T22:06:27Z

Test build #77263 has finished for PR 15009 at commit d906072.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

kishorvpatil · 2017-05-24T13:09:57Z

@vanzin I added waitFor method to ChildSparkHandle allowing it to wait for launched process or thread. Now the test waits for thread to finish.

vanzin · 2017-06-01T21:54:35Z

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

+      .addAppArgs("thread");
+    final SparkAppHandle app = launcher.startApplication();
+    sleep(3000);
+    AbstractSparkAppHandle handle = (AbstractSparkAppHandle)app;


You don't need this. SparkAppHandle is a AbstractSparkAppHandle, isn't it? You should also make waitFor() package private instead of protected.

waitFor is a package private now. This method is added for unit test in AbstractSparkAppHandle not part of the interface SparkAppHandle.

vanzin · 2017-06-01T21:55:57Z

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

+    final SparkAppHandle app = launcher.startApplication();
+    sleep(3000);
+    AbstractSparkAppHandle handle = (AbstractSparkAppHandle)app;
+    handle.waitFor();


How are you testing that the application actually succeeded instead of, e.g., throwing an exception?

In this unit test the main application is Simple application. There is no spark application running in background to track status, so no way to use handle.getState. All I am checking is for launched thread/process is complete.

Seems like there is a bug somewhere; if the app finishes without reporting state (either in child process or thread mode), the handle should change its state to some form of error (e.g. LOST). Anyway that's unrelated to your changes, but please file a separate bug.

You can still check that the state is not FAILED, right? Because if the child application throws an exception in thread mode, the handle should at least be updated to the failed state. (That would be a nice candidate for a new unit test, too.)

vanzin · 2017-06-01T21:56:09Z

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

  public static class SparkLauncherTestApp {

    public static void main(String[] args) throws Exception {
      assertEquals(1, args.length);
-      assertEquals("proc", args[0]);
-      assertEquals("bar", System.getProperty("foo"));
+      if("proc".equalsIgnoreCase(args[0])) {


missing space after if.

vanzin · 2017-06-01T21:57:32Z

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

+
+  protected void waitFor() throws InterruptedException {
+    if(this.childThread.isAlive()) {
+      this.childThread.wait();


Why wait and not join? Who's doing a notify on the thread to wake up this call?

vanzin · 2017-06-01T21:57:41Z

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

+  }
+
+  protected void waitFor() throws InterruptedException {
+    if(this.childThread.isAlive()) {


Missing space after if.

SparkQA · 2017-06-14T15:34:52Z

Test build #78058 has finished for PR 15009 at commit 81fd297.

This patch fails RAT tests.
This patch merges cleanly.
This patch adds no public classes.

vanzin · 2017-06-14T17:46:12Z

retest this please

SparkQA · 2017-06-14T20:35:52Z

Test build #78060 has finished for PR 15009 at commit 81fd297.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-06-14T22:24:52Z

Test build #78064 has finished for PR 15009 at commit 10513ec.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

vanzin · 2017-07-11T22:20:27Z

@kishorvpatil are you planning to address the last bit of feedback remaining here? It shouldn't be that hard to make that test better.

justinuang · 2017-08-11T17:17:18Z

@kishorvpatil this will be quite useful for us! To avoid the 3s cost of spinning up a new jvm just for yarn-cluster

vanzin · 2017-08-11T18:51:21Z

I kinda gave up waiting for feedback to be addressed here. I'm fixing some things in the launcher library and will work on SPARK-11035 next, replacing part of this PR.

justinuang · 2017-08-16T00:10:18Z

That would be incredible. Launching a new jvm and loading all of hadoop takes about 4 seconds extra each time, versus reusing the launcher jvm, which is really significant for us since we launch a lot of jobs and users have to wait on this.

…e External Shuffle Service

Stop Spark Application if launcher goes down and use reflection

8986aa9

tgravescs reviewed Sep 8, 2016
View reviewed changes

Fixing coding style issues and review comments

99b1d1b

kishorvpatil added 2 commits September 8, 2016 18:36

formatting fixes

70a67fb

Remove extra empty lines

bac5cd2

kishorvpatil changed the title ~~[SPARK-17443] Stop Spark Application if launcher goes down and use reflection~~ [SPARK-17443][SPARK-11035] Stop Spark Application if launcher goes down and use reflection Sep 8, 2016

tgravescs reviewed Sep 9, 2016
View reviewed changes

kishorvpatil added 2 commits September 9, 2016 10:12

Formatting and code review comments

38a7e3a

Remove unwanted conditional to null check Boolean value

67f8de7

tgravescs reviewed Sep 12, 2016
View reviewed changes

vanzin reviewed May 19, 2017

View reviewed changes

Fix SparkLauncherSuite with waitFor method.

d906072

vanzin reviewed Jun 1, 2017

View reviewed changes

Fixing SparkLauncherSuite unit test

81fd297

Fixing review comments

10513ec

vanzin mentioned this pull request Sep 26, 2017

[BUILD] Close stale PRs #19348

Closed

asfgit closed this in ceaec93 Sep 27, 2017

zheniantoushipashi added a commit to zheniantoushipashi/spark that referenced this pull request Oct 9, 2019

apache#15009 Remove the reliance of dynamic resource allocation on Th…

cbd3e91

…e External Shuffle Service

zheniantoushipashi added a commit to zheniantoushipashi/spark that referenced this pull request Oct 11, 2019

apache#15009 Remove the reliance of dynamic resource allocation on Th…

6575d4a

…e External Shuffle Service

[SPARK-17443][SPARK-11035] Stop Spark Application if launcher goes down and use reflection #15009

[SPARK-17443][SPARK-11035] Stop Spark Application if launcher goes down and use reflection #15009

Conversation

kishorvpatil commented Sep 8, 2016 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

tgravescs commented Sep 8, 2016

Choose a reason for hiding this comment

SparkQA commented Sep 8, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tgravescs commented Sep 8, 2016

tgravescs Sep 8, 2016 • edited Loading

Choose a reason for hiding this comment

tgravescs commented Sep 8, 2016

tgravescs commented Sep 8, 2016

SparkQA commented Sep 8, 2016

tgravescs commented Sep 9, 2016

tgravescs commented Sep 9, 2016

Choose a reason for hiding this comment

SparkQA commented Sep 9, 2016

Choose a reason for hiding this comment

tgravescs commented Sep 9, 2016

SparkQA commented Sep 9, 2016

Choose a reason for hiding this comment

kishorvpatil commented May 17, 2017

Choose a reason for hiding this comment

SparkQA commented May 23, 2017

kishorvpatil commented May 24, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Jun 14, 2017

vanzin commented Jun 14, 2017

SparkQA commented Jun 14, 2017

SparkQA commented Jun 14, 2017

vanzin commented Jul 11, 2017

justinuang commented Aug 11, 2017

vanzin commented Aug 11, 2017 • edited Loading

justinuang commented Aug 16, 2017

kishorvpatil commented Sep 8, 2016 •

edited

Loading

tgravescs Sep 8, 2016 •

edited

Loading

vanzin commented Aug 11, 2017 •

edited

Loading