(spark) branch master updated: [SPARK-54774][CORE] Submit failed should keep same exit code with app exit code in K8s mode

[ch...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

chengpan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/master by this push:
     new f9bb48cbea01 [SPARK-54774][CORE] Submit failed should keep same exit code with app exit code in K8s mode
f9bb48cbea01 is described below

commit f9bb48cbea010b415169d9976836f6ab36120cbd
Author: Angerszhuuuu <an...@gmail.com>
AuthorDate: Fri Apr 10 10:23:31 2026 +0800

    [SPARK-54774][CORE] Submit failed should keep same exit code with app exit code in K8s mode
    
    ### What changes were proposed in this pull request?
    When submit failed, k8s case will stop  SparkContext.
    Here we already know the exit code, so here we stop SC should also pass the exitcode too.
    
    ### Why are the changes needed?
    keep same exit code for SC and whole app, help when checking the log
    
    ### Does this PR introduce _any_ user-facing change?
    No
    
    ### How was this patch tested?
    Added UT
    
    ### Was this patch authored or co-authored using generative AI tooling?
    No
    
    Closes #53539 from AngersZhuuuu/SPARK-54774.
    
    Authored-by: Angerszhuuuu <an...@gmail.com>
    Signed-off-by: Cheng Pan <ch...@apache.org>
---
 .../org/apache/spark/deploy/SparkSubmit.scala      |  2 +-
 .../org/apache/spark/deploy/SparkSubmitSuite.scala | 52 ++++++++++++++++++++++
 2 files changed, 53 insertions(+), 1 deletion(-)

diff --git a/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala b/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
index 43fd1a47ec29..d1e63fab00ff 100644
--- a/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
@@ -1050,7 +1050,7 @@ private[spark] class SparkSubmit extends Logging {
           !isSqlShell(args.mainClass) && !isThriftServer(args.mainClass) &&
           !isConnectServer(args.mainClass)) {
         try {
-          SparkContext.getActive.foreach(_.stop())
+          SparkContext.getActive.foreach(_.stop(exitCode))
         } catch {
           case e: Throwable => logError("Failed to close SparkContext", e)
         }
diff --git a/core/src/test/scala/org/apache/spark/deploy/SparkSubmitSuite.scala b/core/src/test/scala/org/apache/spark/deploy/SparkSubmitSuite.scala
index 7f115d1ddd75..761e34571cf4 100644
--- a/core/src/test/scala/org/apache/spark/deploy/SparkSubmitSuite.scala
+++ b/core/src/test/scala/org/apache/spark/deploy/SparkSubmitSuite.scala
@@ -1632,6 +1632,38 @@ class SparkSubmitSuite
     assertResult(3)(runSparkSubmit(args, expectFailure = true))
   }
 
+  test("SPARK-54774: k8s submit failed should keep same exit code with user code") {
+    val unusedJar = TestUtils.createJarWithClasses(Seq.empty)
+    val fileSystem = Utils.getHadoopFileSystem("/",
+      SparkHadoopUtil.get.newConfiguration(new SparkConf()))
+    withTempDir { testDir =>
+      val testDirPath = new Path(testDir.getAbsolutePath())
+      val args = Seq(
+        "--class", K8sExitCodeTestApplication.getClass.getName.stripSuffix("$"),
+        "--name", "testApp",
+        "--master", "k8s://host:port",
+        "--conf", "spark.ui.enabled=false",
+        "--conf", "spark.master.rest.enabled=false",
+        "--conf", "spark.kubernetes.authenticate.driver.serviceAccountName=default",
+        "--conf", "spark.eventLog.enabled=true",
+        "--conf", "spark.eventLog.rolling.enabled=false",
+        "--conf", "spark.eventLog.testing=true",
+        "--conf", s"spark.eventLog.dir=${testDirPath.toUri.toString}",
+        unusedJar.toString
+      )
+      // The test application throws SparkUserAppException with exit code 42,
+      // so SparkContext.stop(42) should be called in k8s mode
+      runSparkSubmit(args, expectFailure = true)
+      val listStatus = fileSystem.listStatus(testDirPath)
+      val logData = EventLogFileReader.openEventLog(listStatus.last.getPath, fileSystem)
+      Source.fromInputStream(logData)(Codec.UTF8).getLines().filter { line =>
+        line.contains("SparkListenerApplicationEnd")
+      }.foreach { line =>
+        assert(line.contains("\"ExitCode\":42"))
+      }
+    }
+  }
+
   private def testRemoteResources(
       enableHttpFs: Boolean,
       forceDownloadSchemes: Seq[String] = Nil): Unit = {
@@ -2120,3 +2152,23 @@ class TestSparkApplication extends SparkApplication with Matchers {
   }
 
 }
+
+object K8sExitCodeTestApplication {
+  def main(args: Array[String]): Unit = {
+    TestUtils.configTestLog4j2("INFO")
+    // Use local master to ensure SparkContext can be created in test environment
+    // The k8s master is set in SparkSubmit args, which triggers the finally block logic
+    val conf = new SparkConf().setMaster("local[2]")
+    val sc = new SparkContext(conf)
+    try {
+      // Create a simple RDD to ensure SparkContext is active
+      sc.parallelize(1 to 10).count()
+      // Throw SparkUserAppException with a specific exit code
+      // This simulates a user application failure
+      throw new SparkUserAppException(42)
+    } finally {
+      // Note: In k8s mode, SparkSubmit should call sc.stop(42) in the finally block
+      // We don't call stop() here to let SparkSubmit handle it
+    }
+  }
+}


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org