From 3ec73adf8e44715350f048230a9fc11d0aeea89d Mon Sep 17 00:00:00 2001
From: liangbowen <liangbowen@gf.com.cn>
Date: Thu, 26 Oct 2023 08:43:41 +0800
Subject: [PATCH] [KYUUBI #5522] [BATCH] Ignore main class for PySpark batch
 job submission

### _Why are the changes needed?_

For PySpark batch jobs, the main class is not required for spark submission and should be ignored.

- Spark submit doc: https://spark.apache.org/docs/latest/submitting-applications.html
- Also reported in : https://github.com/apache/kyuubi/pull/5498/files#diff-a215909f2588cb42eb307d1bdaa73eef9e05259bb0a99099f29859235bec19bbR30

### _How was this patch tested?_
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [x] [Run test](https://kyuubi.readthedocs.io/en/master/contributing/code/testing.html#running-tests) locally before make a pull request

### _Was this patch authored or co-authored using generative AI tooling?_

No.

Closes #5522 from bowenliang123/pyspark-submit-ignore.

Closes #5522

c28ef86da [liangbowen] ignore main class for pyspark batch submission

Authored-by: liangbowen <liangbowen@gf.com.cn>
Signed-off-by: Bowen Liang <liangbowen@gf.com.cn>
---
 .../kyuubi/operation/BatchJobSubmission.scala | 29 +++++++++----------
 1 file changed, 14 insertions(+), 15 deletions(-)

diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/operation/BatchJobSubmission.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/operation/BatchJobSubmission.scala
index 779dc48ae6a..af6242ae1c7 100644
--- a/kyuubi-server/src/main/scala/org/apache/kyuubi/operation/BatchJobSubmission.scala
+++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/operation/BatchJobSubmission.scala
@@ -84,22 +84,21 @@ class BatchJobSubmission(
 
   @VisibleForTesting
   private[kyuubi] val builder: ProcBuilder = {
-    Option(batchType).map(_.toUpperCase(Locale.ROOT)) match {
-      case Some("SPARK") | Some("PYSPARK") =>
-        new SparkBatchProcessBuilder(
-          session.user,
-          session.sessionConf,
-          batchId,
-          batchName,
-          Option(resource),
-          className,
-          batchConf,
-          batchArgs,
-          getOperationLog)
-
-      case _ =>
-        throw new UnsupportedOperationException(s"Batch type $batchType unsupported")
+    val mainClass = Option(batchType).map(_.toUpperCase(Locale.ROOT)) match {
+      case Some("SPARK") => className
+      case Some("PYSPARK") => null
+      case _ => throw new UnsupportedOperationException(s"Batch type $batchType unsupported")
     }
+    new SparkBatchProcessBuilder(
+      session.user,
+      session.sessionConf,
+      batchId,
+      batchName,
+      Option(resource),
+      mainClass,
+      batchConf,
+      batchArgs,
+      getOperationLog)
   }
 
   override def currentApplicationInfo(): Option[ApplicationInfo] = {