Call SparkSubmit from spark-class launcher for unknown classes.

For new-style launchers, do the launching using SparkSubmit; hopefully
this will be the preferred method of launching new daemons (if any).
Currently it handles the thrift server daemon.

Author: Marcelo Vanzin

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

@@ -194,15 +194,15 @@ protected List<String> buildClassPath(String appClassPath) throws IOException {
     }
 
     boolean prependClasses = !isEmpty(getenv("SPARK_PREPEND_CLASSES"));
-    boolean isTesting = !isEmpty(getenv("SPARK_TESTING"));
+    boolean isTesting = "1".equals(getenv("SPARK_TESTING"));
     if (prependClasses || isTesting) {
-      System.err.println(
-        "NOTE: SPARK_PREPEND_CLASSES is set, placing locally compiled Spark classes ahead of " +
-        "assembly.");
       List<String> projects = Arrays.asList("core", "repl", "mllib", "bagel", "graphx",
         "streaming", "tools", "sql/catalyst", "sql/core", "sql/hive", "sql/hive-thriftserver",
         "yarn", "launcher");
       if (prependClasses) {
+        System.err.println(
+          "NOTE: SPARK_PREPEND_CLASSES is set, placing locally compiled Spark classes ahead of " +
+          "assembly.");
         for (String project : projects) {
           addToClassPath(cp, String.format("%s/%s/target/scala-%s/classes", sparkHome, project,
             scala));

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

@@ -42,8 +42,6 @@ class SparkClassLauncher extends AbstractLauncher<SparkClassLauncher> {
 
   @Override
   protected List<String> buildLauncherCommand() throws IOException {
-    List<String> cmd = createJavaCommand();
-
     List<String> javaOptsKeys = new ArrayList<String>();
     String memKey = null;
     String extraClassPath = null;
@@ -88,8 +86,12 @@ protected List<String> buildLauncherCommand() throws IOException {
         toolsDir.getAbsolutePath(), className);
 
       javaOptsKeys.add("SPARK_JAVA_OPTS");
+    } else {
+      // Any classes not explicitly listed above are submitted using SparkSubmit.
+      return buildSparkSubmitCommand();
     }
 
+    List<String> cmd = createJavaCommand();
     for (String key : javaOptsKeys) {
       addOptionString(cmd, System.getenv(key));
     }
@@ -104,4 +106,14 @@ protected List<String> buildLauncherCommand() throws IOException {
     return prepareForOs(cmd, null, Collections.<String, String>emptyMap());
   }
 
+  private List<String> buildSparkSubmitCommand() throws IOException {
+    List<String> sparkSubmitArgs = new ArrayList<String>(classArgs);
+    sparkSubmitArgs.add(SparkSubmitOptionParser.CLASS);
+    sparkSubmitArgs.add(className);
+
+    SparkSubmitCliLauncher launcher = new SparkSubmitCliLauncher(true, sparkSubmitArgs);
+    launcher.setAppResource("spark-internal");
+    return launcher.buildLauncherCommand();
+  }
+
 }

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

@@ -49,16 +49,16 @@ public class SparkSubmitCliLauncher extends SparkLauncher {
   }
 
   private final List<String> driverArgs;
-  private boolean isShell;
+  private boolean hasMixedArguments;
 
   SparkSubmitCliLauncher(List<String> args) {
     this(false, args);
   }
 
-  SparkSubmitCliLauncher(boolean isShell, List<String> args) {
+  SparkSubmitCliLauncher(boolean hasMixedArguments, List<String> args) {
     boolean sparkSubmitOptionsEnded = false;
     this.driverArgs = new ArrayList<String>();
-    this.isShell = isShell;
+    this.hasMixedArguments = hasMixedArguments;
     new OptionParser().parse(args);
   }
 
@@ -126,7 +126,7 @@ protected boolean handle(String opt, String value) {
         // non-spark-submit arguments.
         setClass(value);
         if (shells.containsKey(value)) {
-          isShell = true;
+          hasMixedArguments = true;
           setAppResource(shells.get(value));
         }
       } else {
@@ -137,10 +137,10 @@ protected boolean handle(String opt, String value) {
 
     @Override
     protected boolean handleUnknown(String opt) {
-      // When running a shell, add unrecognized parameters directly to the user arguments list.
+      // When mixing arguments, add unrecognized parameters directly to the user arguments list.
       // In normal mode, any unrecognized parameter triggers the end of command line parsing.
       // The remaining params will be appended to the list of SparkSubmit arguments.
-      if (isShell) {
+      if (hasMixedArguments) {
         addArgs(opt);
         return true;
       } else {

sbin/spark-daemon.sh

@@ -124,7 +124,7 @@ fi
 
 case $option in
 
-  (start|spark-submit)
+  (start)
 
     mkdir -p "$SPARK_PID_DIR"
 
@@ -142,14 +142,7 @@ case $option in
 
     spark_rotate_log "$log"
     echo starting $command, logging to $log
-    if [ $option == spark-submit ]; then
-      source "$SPARK_HOME"/bin/utils.sh
-      gatherSparkSubmitOpts "$@"
-      nohup nice -n $SPARK_NICENESS "$SPARK_PREFIX"/bin/spark-submit --class $command \
-        "${SUBMISSION_OPTS[@]}" spark-internal "${APPLICATION_OPTS[@]}" >> "$log" 2>&1 < /dev/null &
-    else
-      nohup nice -n $SPARK_NICENESS "$SPARK_PREFIX"/bin/spark-class $command "$@" >> "$log" 2>&1 < /dev/null &
-    fi
+    nohup nice -n $SPARK_NICENESS "$SPARK_PREFIX"/bin/spark-class $command "$@" >> "$log" 2>&1 < /dev/null &
     newpid=$!
     echo $newpid > $pid
     sleep 2

sbin/start-thriftserver.sh

@@ -50,4 +50,4 @@ fi
 
 export SUBMIT_USAGE_FUNCTION=usage
 
-exec "$FWDIR"/sbin/spark-daemon.sh spark-submit $CLASS 1 "$@"
+exec "$FWDIR"/sbin/spark-daemon.sh start $CLASS 1 "$@"