[LIVY-1059][FOLLOWUP] Fix missing validation condition and add unit test
## What changes were proposed in this pull request?
This is a follow-up fix for [LIVY-1059](https://github.com/apache/livy/pull/529).
When a client sends `"queue": ""` in the session create JSON, Livy was treating it as a valid queue value instead of falling back to the server default `livy.spark.yarn.queue`.
This change treats an empty queue string the same as a missing queue, so the default YARN queue from Livy config is used when the client passes "".
Updated in both batch and interactive session creation:
`request.queue.filterNot(_.isEmpty).orElse(livyConf.sparkYarnQueue())`
## How was this patch tested?
- Added unit test in `BatchSessionSpec` for `queue = Some("")` and verified the default queue is passed to spark-submit
- Added unit test in `InteractiveSessionSpec` for `queue = Some("")` and verified the session uses the default queue from `LivyConf`
- Existing tests for `queue = None` and user-provided queue values still passdiff --git a/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala b/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala
index ec0f961..de77115 100644
--- a/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala
+++ b/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala
@@ -83,7 +83,7 @@
request.executorMemory.foreach(builder.executorMemory)
request.executorCores.foreach(builder.executorCores)
request.numExecutors.foreach(builder.numExecutors)
- request.queue.orElse(livyConf.sparkYarnQueue()).foreach(builder.queue)
+ request.queue.filterNot(_.isEmpty).orElse(livyConf.sparkYarnQueue()).foreach(builder.queue)
request.name.foreach(builder.name)
sessionStore.save(BatchSession.RECOVERY_SESSION_TYPE, s.recoveryMetadata)
diff --git a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
index 0c7cb0b..e578326 100644
--- a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
+++ b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
@@ -107,7 +107,7 @@
SparkLauncher.EXECUTOR_MEMORY -> request.executorMemory.map(_.toString),
"spark.executor.instances" -> request.numExecutors.map(_.toString),
"spark.app.name" -> request.name.map(_.toString),
- "spark.yarn.queue" -> request.queue.orElse(livyConf.sparkYarnQueue())
+ "spark.yarn.queue" -> request.queue.filterNot(_.isEmpty).orElse(livyConf.sparkYarnQueue())
)
userOpts.foreach { case (key, opt) =>
@@ -152,7 +152,7 @@
request.jars,
request.numExecutors,
request.pyFiles,
- request.queue.orElse(livyConf.sparkYarnQueue()),
+ request.queue.filterNot(_.isEmpty).orElse(livyConf.sparkYarnQueue()),
mockApp)
}
diff --git a/server/src/test/scala/org/apache/livy/server/batch/BatchSessionSpec.scala b/server/src/test/scala/org/apache/livy/server/batch/BatchSessionSpec.scala
index 807b20f..9ecfd57 100644
--- a/server/src/test/scala/org/apache/livy/server/batch/BatchSessionSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/batch/BatchSessionSpec.scala
@@ -33,7 +33,7 @@
import org.apache.livy.server.AccessManager
import org.apache.livy.server.recovery.SessionStore
import org.apache.livy.sessions.SessionState
-import org.apache.livy.utils.{AppInfo, Clock, SparkApp, SparkProcessBuilder}
+import org.apache.livy.utils.{AppInfo, Clock, SparkApp}
class BatchSessionSpec
extends FunSpec
@@ -175,6 +175,24 @@
batch.logLines().mkString should not include "livy-default-batch-queue"
}
+ it("should pass default YARN queue when request queue is empty string") {
+ val req = new CreateBatchRequest()
+ req.file = script.toString
+ req.queue = Some("")
+ req.conf = Map("spark.driver.extraClassPath" -> sys.props("java.class.path"))
+
+ val conf = new LivyConf()
+ .set(LivyConf.LOCAL_FS_WHITELIST, sys.props("java.io.tmpdir"))
+ .set(LivyConf.SPARK_YARN_QUEUE, "livy-default-batch-queue")
+
+ val accessManager = new AccessManager(conf)
+ val batch = BatchSession.create(30, None, req, conf, accessManager, null, None, sessionStore)
+ batch.start()
+
+ Utils.waitUntil({ () => !batch.state.isActive }, Duration(10, TimeUnit.SECONDS))
+ batch.logLines().mkString should include("livy-default-batch-queue")
+ }
+
def testRecoverSession(name: Option[String]): Unit = {
val conf = new LivyConf()
val req = new CreateBatchRequest()
diff --git a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionSpec.scala b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionSpec.scala
index 31daefa..72a2a31 100644
--- a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionSpec.scala
@@ -337,18 +337,8 @@
val mockClient = Some(mock[RSCClient])
val s = InteractiveSession.create(
- id = 101,
- name = None,
- owner = "systest",
- proxyUser = None,
- livyConf = testLivyConf,
- accessManager = accessManager,
- request = req,
- sessionStore = mock[SessionStore],
- ttl = None,
- idleTimeout = None,
- mockApp = None,
- mockClient = mockClient
+ 101, None, "systest", None, testLivyConf, accessManager,
+ req, mock[SessionStore], None, None, None, mockClient
)
s.queue shouldBe Some("livy-default-queue")
@@ -367,21 +357,29 @@
val mockClient = Some(mock[RSCClient])
val s = InteractiveSession.create(
- id = 102,
- name = None,
- owner = "systest",
- proxyUser = None,
- livyConf = testLivyConf,
- accessManager = accessManager,
- request = req,
- sessionStore = mock[SessionStore],
- ttl = None,
- idleTimeout = None,
- mockApp = None,
- mockClient = mockClient
+ 102, None, "systest", None, testLivyConf, accessManager,
+ req, mock[SessionStore], None, None, None, mockClient
)
s.queue shouldBe Some("user-custom-queue")
}
+
+ it("should inherit default YARN queue when request queue is empty string") {
+ val testLivyConf = new LivyConf()
+ .set(LivyConf.REPL_JARS, "dummy.jar")
+ .set(LivyConf.SPARK_YARN_QUEUE, "livy-default-queue")
+
+ val req = new CreateInteractiveRequest()
+ req.kind = Spark
+ req.queue = Some("") // explicitly empty string
+ req.conf = Map(RSCConf.Entry.LIVY_JARS.key() -> "")
+
+ val s = InteractiveSession.create(
+ 103, None, "systest", None, testLivyConf, accessManager,
+ req, mock[SessionStore], None, None, None, Some(mock[RSCClient])
+ )
+
+ s.queue shouldBe Some("livy-default-queue")
+ }
}
}
