Update SparkSubmitSuite.scala

Author: AngersZhuuuu

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

@@ -1631,6 +1631,22 @@ 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 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",
+      unusedJar.toString
+    )
+    // The test application throws SparkUserAppException with exit code 42,
+    // so SparkContext.stop(42) should be called in k8s mode
+    assertResult(42)(runSparkSubmit(args, expectFailure = true))
+  }
+
   private def testRemoteResources(
       enableHttpFs: Boolean,
       forceDownloadSchemes: Seq[String] = Nil): Unit = {
@@ -2038,3 +2054,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
+    }
+  }
+}