Review feedback.

Author: Marcelo Vanzin

core/src/main/scala/org/apache/spark/deploy/SparkSubmitArguments.scala

@@ -33,7 +33,7 @@ import org.apache.spark.util.Utils
  * The env argument is used for testing.
  */
 private[spark] class SparkSubmitArguments(args: Seq[String], env: Map[String, String] = sys.env)
-    extends SparkSubmitArgumentsParser {
+  extends SparkSubmitArgumentsParser {
   var master: String = null
   var deployMode: String = null
   var executorMemory: String = null
@@ -397,6 +397,8 @@ private[spark] class SparkSubmitArguments(args: Seq[String], env: Map[String, St
   }
 
   /**
+   * Handle unrecognized command line options.
+   *
    * The first unrecognized option is treated as the "primary resource". Everything else is
    * treated as application arguments.
    */
@@ -419,7 +421,6 @@ private[spark] class SparkSubmitArguments(args: Seq[String], env: Map[String, St
     childArgs ++= extra
   }
 
-
   private def printUsageAndExit(exitCode: Int, unknownParam: Any = null): Unit = {
     val outStream = SparkSubmit.printStream
     if (unknownParam != null) {

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

@@ -19,8 +19,7 @@ package org.apache.spark.launcher
 
 /**
  * This class makes SparkSubmitOptionParser visible for Spark code outside of the `launcher`
- * package.
+ * package, since Java doesn't have a feature similar to `private[spark]`, and we don't want
+ * that class to be public.
  */
-private[spark] abstract class SparkSubmitArgumentsParser extends SparkSubmitOptionParser {
-
-}
+private[spark] abstract class SparkSubmitArgumentsParser extends SparkSubmitOptionParser

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

@@ -25,8 +25,9 @@ import scala.collection.JavaConversions._
 import org.apache.spark.deploy.Command
 
 /**
- * This class is used by CommandUtils. It uses some package-private APIs in SparkLauncher and so
- * needs to live in the same package as the rest of the library.
+ * This class is used by CommandUtils. It uses some package-private APIs in SparkLauncher, and since
+ * Java doesn't have a feature similar to `private[spark]`, and we don't want that class to be
+ * public, needs to live in the same package as the rest of the library.
  */
 private[spark] class WorkerCommandBuilder(sparkHome: String, memoryMb: Int, command: Command)
     extends AbstractCommandBuilder {
@@ -38,7 +39,7 @@ private[spark] class WorkerCommandBuilder(sparkHome: String, memoryMb: Int, comm
     val cmd = buildJavaCommand(command.classPathEntries.mkString(File.pathSeparator))
     cmd.add(s"-Xms${memoryMb}M")
     cmd.add(s"-Xmx${memoryMb}M")
-    command.javaOpts.foreach { cmd.add }
+    command.javaOpts.foreach(cmd.add)
     addPermGenSizeOpt(cmd)
     addOptionString(cmd, getenv("SPARK_JAVA_OPTS"))
     cmd

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

@@ -36,7 +36,7 @@
 import static org.apache.spark.launcher.CommandBuilderUtils.*;
 
 /**
- * Abstract command builder that defines common functionality for all builders.
+ * Abstract Spark command builder that defines common functionality.
  */
 abstract class AbstractCommandBuilder {
 
@@ -56,20 +56,16 @@ abstract class AbstractCommandBuilder {
   final Map<String, String> conf;
 
   public AbstractCommandBuilder() {
-    this(Collections.<String, String>emptyMap());
-  }
-
-  public AbstractCommandBuilder(Map<String, String> env) {
     this.appArgs = new ArrayList<String>();
-    this.childEnv = new HashMap<String, String>(env);
+    this.childEnv = new HashMap<String, String>();
     this.conf = new HashMap<String, String>();
     this.files = new ArrayList<String>();
     this.jars = new ArrayList<String>();
     this.pyFiles = new ArrayList<String>();
   }
 
   /**
-   * Builds the command like to execute.
+   * Builds the command to execute.
    *
    * @param env A map containing environment variables for the child process. It may already contain
    *            entries defined by the user (such as SPARK_HOME, or those defined by the
@@ -78,6 +74,16 @@ public AbstractCommandBuilder(Map<String, String> env) {
    */
   abstract List<String> buildCommand(Map<String, String> env) throws IOException;
 
+  /**
+   * Builds a list of arguments to run java.
+   *
+   * This method finds the java executable to use and appends JVM-specific options for running a
+   * class with Spark in the classpath. It also loads options from the "java-opts" file in the
+   * configuration directory being used.
+   *
+   * Callers should still add at least the class to run, as well as any arguments to pass to the
+   * class.
+   */
   List<String> buildJavaCommand(String extraClassPath) throws IOException {
     List<String> cmd = new ArrayList<String>();
     if (javaHome == null) {
@@ -241,7 +247,7 @@ List<String> buildClassPath(String appClassPath) throws IOException {
   /**
    * Adds entries to the classpath.
    *
-   * @param cp List where to appended the new classpath entries.
+   * @param cp List to which the new entries are appended.
    * @param entries New classpath entries (separated by File.pathSeparator).
    */
   private void addToClassPath(List<String> cp, String entries) {
@@ -268,18 +274,15 @@ String getScalaVersion() {
     String sparkHome = getSparkHome();
     File scala210 = new File(sparkHome, "assembly/target/scala-2.10");
     File scala211 = new File(sparkHome, "assembly/target/scala-2.11");
-    if (scala210.isDirectory() && scala211.isDirectory()) {
-      checkState(false,
-        "Presence of build for both scala versions (2.10 and 2.11) detected.\n" +
-        "Either clean one of them or set SPARK_SCALA_VERSION in your environment.");
-    } else if (scala210.isDirectory()) {
+    checkState(!scala210.isDirectory() || !scala211.isDirectory(),
+      "Presence of build for both scala versions (2.10 and 2.11) detected.\n" +
+      "Either clean one of them or set SPARK_SCALA_VERSION in your environment.");
+    if (scala210.isDirectory()) {
       return "2.10";
     } else {
       checkState(scala211.isDirectory(), "Cannot find any assembly build directories.");
       return "2.11";
     }
-
-    throw new IllegalStateException("Should not reach here.");
   }
 
   String getSparkHome() {

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

@@ -64,8 +64,10 @@ static String join(String sep, Iterable<String> elements) {
     return sb.toString();
   }
 
-  /** Returns the first value mapped to the given key in the given maps. */
-  static String find(String key, Map<?, ?>... maps) {
+  /**
+   * Returns the first non-empty value mapped to the given key in the given maps, or null otherwise.
+   */
+  static String firstNonEmptyValue(String key, Map<?, ?>... maps) {
     for (Map<?, ?> map : maps) {
       String value = (String) map.get(key);
       if (!isEmpty(value)) {
@@ -75,7 +77,7 @@ static String find(String key, Map<?, ?>... maps) {
     return null;
   }
 
-  /** Returns the first non-empty, non-null string in the given list. */
+  /** Returns the first non-empty, non-null string in the given list, or null otherwise. */
   static String firstNonEmpty(String... candidates) {
     for (String s : candidates) {
       if (!isEmpty(s)) {
@@ -106,8 +108,8 @@ static boolean isWindows() {
   }
 
   /**
-   * Updates the user environment to contain the merged value of "envKey" after appending
-   * the given path list.
+   * Updates the user environment, appending the given pathList to the existing value of the given
+   * environment variable (or setting it if it hasn't yet been set).
    */
   static void mergeEnvPathList(Map<String, String> userEnv, String envKey, String pathList) {
     if (!isEmpty(pathList)) {
@@ -117,8 +119,8 @@ static void mergeEnvPathList(Map<String, String> userEnv, String envKey, String
   }
 
   /**
-   * Parse a string as if it were a list of arguments, in the way that a shell would.
-   * This tries to follow the way bash parses strings. For example:
+   * Parse a string as if it were a list of arguments, following bash semantics.
+   * For example:
    *
    * Input: "\"ab cd\" efgh 'i \" j'"
    * Output: [ "ab cd", "efgh", "i \" j" ]
@@ -130,6 +132,8 @@ static List<String> parseOptionString(String s) {
     boolean inSingleQuote = false;
     boolean inDoubleQuote = false;
     boolean escapeNext = false;
+
+    // This is needed to detect when a quoted empty string is used as an argument ("" or '').
     boolean hasData = false;
 
     for (int i = 0; i < s.length(); i++) {
@@ -161,7 +165,7 @@ static List<String> parseOptionString(String s) {
           }
           break;
         default:
-          if (inSingleQuote || inDoubleQuote || !Character.isWhitespace(c)) {
+          if (!Character.isWhitespace(c) || inSingleQuote || inDoubleQuote) {
             opt.appendCodePoint(c);
           } else {
             opts.add(opt.toString());
@@ -225,4 +229,56 @@ static void checkState(boolean check, String msg, Object... args) {
     }
   }
 
+  /**
+   * Quote a command argument for a command to be run by a Windows batch script, if the argument
+   * needs quoting. Arguments only seem to need quotes in batch scripts if they have whitespace.
+   *
+   *  For example:
+   *    original single argument: ab "cde" fgh
+   *    quoted: "ab ""cde"" fgh"
+   */
+  static String quoteForBatchScript(String arg) {
+    boolean needsQuotes = false;
+    for (int i = 0; i < arg.length(); i++) {
+      if (Character.isWhitespace(arg.codePointAt(i)) || arg.codePointAt(i) == '"') {
+        needsQuotes = true;
+        break;
+      }
+    }
+    if (!needsQuotes) {
+      return arg;
+    }
+    StringBuilder quoted = new StringBuilder();
+    quoted.append("\"");
+    for (int i = 0; i < arg.length(); i++) {
+      int cp = arg.codePointAt(i);
+      if (cp == '\"') {
+        quoted.append("\"");
+      }
+      quoted.appendCodePoint(cp);
+    }
+    quoted.append("\"");
+    return quoted.toString();
+  }
+
+  /**
+   * Quotes a string so that it can be used in a command string and be parsed back into a single
+   * argument by python's "shlex.split()" function.
+   *
+   * Basically, just add simple escapes. E.g.:
+   *    original single argument : ab "cd" ef
+   *    after: "ab \"cd\" ef"
+   */
+  static String quoteForPython(String s) {
+    StringBuilder quoted = new StringBuilder().append('"');
+    for (int i = 0; i < s.length(); i++) {
+      int cp = s.codePointAt(i);
+      if (cp == '"' || cp == '\\') {
+        quoted.appendCodePoint('\\');
+      }
+      quoted.appendCodePoint(cp);
+    }
+    return quoted.append('"').toString();
+  }
+
 }

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

@@ -75,9 +75,10 @@ public static void main(String[] argsArray) throws Exception {
     }
 
     if (isWindows()) {
-      System.out.println(prepareForWindows(cmd, env));
+      System.out.println(prepareWindowsCommand(cmd, env));
     } else {
-      List<String> bashCmd = prepareForBash(cmd, env);
+      // In bash, use NULL as the arg separator since it cannot be used in an argument.
+      List<String> bashCmd = prepareBashCommand(cmd, env);
       for (String c : bashCmd) {
         System.out.print(c);
         System.out.print('\0');
@@ -95,7 +96,7 @@ public static void main(String[] argsArray) throws Exception {
    * The command is executed using "cmd /c" and formatted in single line, since that's the
    * easiest way to consume this from a batch script (see spark-class2.cmd).
    */
-  private static String prepareForWindows(List<String> cmd, Map<String, String> childEnv) {
+  private static String prepareWindowsCommand(List<String> cmd, Map<String, String> childEnv) {
     StringBuilder cmdline = new StringBuilder("cmd /c \"");
     for (Map.Entry<String, String> e : childEnv.entrySet()) {
       cmdline.append(String.format("set %s=%s", e.getKey(), e.getValue()));
@@ -113,7 +114,7 @@ private static String prepareForWindows(List<String> cmd, Map<String, String> ch
    * Prepare the command for execution from a bash script. The final command will have commands to
    * set up any needed environment variables needed by the child process.
    */
-  private static List<String> prepareForBash(List<String> cmd, Map<String, String> childEnv) {
+  private static List<String> prepareBashCommand(List<String> cmd, Map<String, String> childEnv) {
     if (childEnv.isEmpty()) {
       return cmd;
     }
@@ -128,34 +129,6 @@ private static List<String> prepareForBash(List<String> cmd, Map<String, String>
     return newCmd;
   }
 
-  /**
-   * Quote a command argument for a command to be run by a Windows batch script, if the argument
-   * needs quoting. Arguments only seem to need quotes in batch scripts if they have whitespace.
-   */
-  private static String quoteForBatchScript(String arg) {
-    boolean needsQuotes = false;
-    for (int i = 0; i < arg.length(); i++) {
-      if (Character.isWhitespace(arg.codePointAt(i))) {
-        needsQuotes = true;
-        break;
-      }
-    }
-    if (!needsQuotes) {
-      return arg;
-    }
-    StringBuilder quoted = new StringBuilder();
-    quoted.append("\"");
-    for (int i = 0; i < arg.length(); i++) {
-      int cp = arg.codePointAt(i);
-      if (cp == '\"') {
-        quoted.append("\"");
-      }
-      quoted.appendCodePoint(cp);
-    }
-    quoted.append("\"");
-    return quoted.toString();
-  }
-
   /**
    * Internal launcher used when command line parsing fails. This will behave differently depending
    * on the platform:

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

@@ -89,9 +89,6 @@ public List<String> buildCommand(Map<String, String> env) throws IOException {
         toolsDir.getAbsolutePath(), className);
 
       javaOptsKeys.add("SPARK_JAVA_OPTS");
-    } else {
-      // Any classes not explicitly listed above are submitted using SparkSubmit.
-      return createSparkSubmitCommand(env);
     }
 
     List<String> cmd = buildJavaCommand(extraClassPath);
@@ -108,48 +105,4 @@ public List<String> buildCommand(Map<String, String> env) throws IOException {
     return cmd;
   }
 
-  private List<String> createSparkSubmitCommand(Map<String, String> env) throws IOException {
-    final List<String> sparkSubmitArgs = new ArrayList<String>();
-    final List<String> appArgs = new ArrayList<String>();
-
-    // Parse the command line and special-case the HELP command line argument, allowing it to be
-    // propagated to the app being launched.
-    SparkSubmitOptionParser parser = new SparkSubmitOptionParser() {
-
-      @Override
-      protected boolean handle(String opt, String value) {
-        if (opt.equals(HELP)) {
-          appArgs.add(opt);
-        } else {
-          sparkSubmitArgs.add(opt);
-          sparkSubmitArgs.add(value);
-        }
-        return true;
-      }
-
-      @Override
-      protected boolean handleUnknown(String opt) {
-        appArgs.add(opt);
-        return true;
-      }
-
-      @Override
-      protected void handleExtraArgs(List<String> extra) {
-        appArgs.addAll(extra);
-      }
-
-    };
-
-    parser.parse(classArgs);
-    sparkSubmitArgs.add(parser.CLASS);
-    sparkSubmitArgs.add(className);
-
-    SparkSubmitCommandBuilder builder = new SparkSubmitCommandBuilder(true, sparkSubmitArgs);
-    builder.appResource = "spark-internal";
-    for (String arg: appArgs) {
-      builder.appArgs.add(arg);
-    }
-    return builder.buildCommand(env);
-  }
-
 }