Fixed spark-submit application options handling logic

Any options in the application option list with the same option name
that SparkSubmitArguments recognizes (e.g., --help) are stolen by
SparkSubmit instead of passed to the application.

Author: liancheng

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

@@ -204,8 +204,9 @@ private[spark] class SparkSubmitArguments(args: Seq[String]) {
 
   /** Fill in values by parsing user options. */
   private def parseOpts(opts: Seq[String]): Unit = {
+    val EQ_SEPARATED_OPT = """(--[^=]+)=(.+)""".r
+
     // Delineates parsing of Spark options from parsing of user options.
-    var inSparkOpts = true
     parse(opts)
 
     def parse(opts: Seq[String]): Unit = opts match {
@@ -306,33 +307,21 @@ private[spark] class SparkSubmitArguments(args: Seq[String]) {
         verbose = true
         parse(tail)
 
+      case EQ_SEPARATED_OPT(opt, value) :: tail =>
+        // convert --foo=bar to --foo bar
+        parse(opt :: value :: tail)
+
+      case value :: tail if value.startsWith("-") =>
+        SparkSubmit.printErrorAndExit(s"Unrecognized option '$value'.")
+
       case value :: tail =>
-        if (inSparkOpts) {
-          value match {
-            // convert --foo=bar to --foo bar
-            case v if v.startsWith("--") && v.contains("=") && v.split("=").size == 2 =>
-              val parts = v.split("=")
-              parse(Seq(parts(0), parts(1)) ++ tail)
-            case v if v.startsWith("-") =>
-              val errMessage = s"Unrecognized option '$value'."
-              SparkSubmit.printErrorAndExit(errMessage)
-            case v =>
-              primaryResource =
-                if (!SparkSubmit.isShell(v) && !SparkSubmit.isInternal(v)) {
-                  Utils.resolveURI(v).toString
-                } else {
-                  v
-                }
-              inSparkOpts = false
-              isPython = SparkSubmit.isPython(v)
-              parse(tail)
-          }
+        primaryResource = if (!SparkSubmit.isShell(value) && !SparkSubmit.isInternal(value)) {
+          Utils.resolveURI(value).toString
         } else {
-          if (!value.isEmpty) {
-            childArgs += value
-          }
-          parse(tail)
+          value
         }
+        isPython = SparkSubmit.isPython(value)
+        childArgs ++= tail.filter(_.nonEmpty)
 
       case Nil =>
     }

core/src/test/scala/org/apache/spark/deploy/SparkSubmitSuite.scala

@@ -101,9 +101,10 @@ class SparkSubmitSuite extends FunSuite with Matchers {
       "--class", "Foo",
       "userjar.jar",
       "some",
+      "--class", "Bar",
       "--weird", "args")
     val appArgs = new SparkSubmitArguments(clArgs)
-    appArgs.childArgs should be (Seq("some", "--weird", "args"))
+    appArgs.childArgs should be (Seq("some", "--class", "Bar", "--weird", "args"))
   }
 
   test("handles YARN cluster mode") {