[SPARK-35949][CORE]Add keep-spark-context-alive arg for to prevent closing spark context after invoking main for some case

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

@@ -957,8 +957,8 @@ private[spark] class SparkSubmit extends Logging {
       case t: Throwable =>
         throw findCause(t)
     } finally {
-      if (!isShell(args.primaryResource) && !isSqlShell(args.mainClass) &&
-        !isThriftServer(args.mainClass)) {
+      if (!args.keepSparkContextAlive && !isShell(args.primaryResource) &&
+        !isSqlShell(args.mainClass) && !isThriftServer(args.mainClass)) {
         try {
           SparkContext.getActive.foreach(_.stop())
         } catch {

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

@@ -82,6 +82,7 @@ private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, S
   var submissionToKill: String = null
   var submissionToRequestStatusFor: String = null
   var useRest: Boolean = false // used internally
+  var keepSparkContextAlive: Boolean = false // keep spark context alive after invoking main
 
   /** Default properties present in the currently defined defaults file. */
   lazy val defaultSparkProperties: HashMap[String, String] = {
@@ -445,6 +446,9 @@ private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, S
       case USAGE_ERROR =>
         printUsageAndExit(1)
 
+      case KEEP_SPARK_CONTEXT_ALIVE =>
+        keepSparkContextAlive = value.toBoolean
+
       case _ =>
         error(s"Unexpected argument '$opt'.")
     }
@@ -535,6 +539,11 @@ private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, S
         |  --proxy-user NAME           User to impersonate when submitting the application.
         |                              This argument does not work with --principal / --keytab.
         |
+        |  --keep-spark-context-alive  True or False. Whether to keep the spark context alive after
+        |                              invoking main method. Default is false. If value is true, spark
+        |                              context will not close after invoking main method. Otherwise
+        |                              the spark context will close in shut down hook.
+        |
         |  --help, -h                  Show this help message and exit.
         |  --verbose, -v               Print additional debug output.
         |  --version,                  Print the version of current Spark.

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

@@ -1525,6 +1525,18 @@ class SparkSubmitSuite
       conf.get(k) should be (v)
     }
   }
+
+  test("handles arguments with --keep-spark-context-alive") {
+    val clArgs = Seq(
+      "--name=myApp",
+      "--class=org.FooBar",
+      "--keep-spark-context-alive", "true",
+      "test.jar"
+    )
+    val appArgs = new SparkSubmitArguments(clArgs)
+    appArgs.name should be ("myApp")
+    appArgs.keepSparkContextAlive should be (true)
+  }
 }
 
 object SparkSubmitSuite extends SparkFunSuite with TimeLimits {

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

@@ -58,6 +58,7 @@ class SparkSubmitOptionParser {
   protected final String REPOSITORIES = "--repositories";
   protected final String STATUS = "--status";
   protected final String TOTAL_EXECUTOR_CORES = "--total-executor-cores";
+  protected final String KEEP_SPARK_CONTEXT_ALIVE = "--keep-spark-context-alive";
 
   // Options that do not take arguments.
   protected final String HELP = "--help";
@@ -115,6 +116,7 @@ class SparkSubmitOptionParser {
     { REPOSITORIES },
     { STATUS },
     { TOTAL_EXECUTOR_CORES },
+    { KEEP_SPARK_CONTEXT_ALIVE },
   };
 
   /**