[KYUUBI #2308][SUB-TASK][KPIP-4] Batch job configuration ignore list and pre-defined configuration in server-side

### _Why are the changes needed?_

To close #2308

Support to ignore some batch configuration items and pre-define some configuration in server side.

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

- [ ] Add screenshots for manual tests if appropriate

- [ ] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #2386 from turboFei/KYUUBI_2308_kpip4_batch_conf.

Closes #2308

2e517ec [Fei Wang] [KYUUBI #2308][SUB-TASK][KPIP-4] Batch job configuration ignore list and server predefined configurations

Authored-by: Fei Wang <fwang12@ebay.com>
Signed-off-by: Fei Wang <fwang12@ebay.com>

Author: turboFei

docs/deployment/settings.md

@@ -159,6 +159,7 @@ Key | Default | Meaning | Type | Since
 Key | Default | Meaning | Type | Since
 --- | --- | --- | --- | ---
 <code>kyuubi.batch.application.check.interval</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>PT5S</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>The interval to check batch job application information.</div>|<div style='width: 30pt'>duration</div>|<div style='width: 20pt'>1.6.0</div>
+<code>kyuubi.batch.conf.ignore.list</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'></div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>A comma separated list of ignored keys for batch conf. If the batch conf contains any of them, the key and the corresponding value will be removed silently during batch job submission. Note that this rule is for server-side protection defined via administrators to prevent some essential configs from tampering. You can also pre-define some config for batch job submission with prefix: kyuubi.batchConf.[batchType]. For example, you can pre-define `spark.master` for spark batch job with key `kyuubi.batchConf.spark.spark.master`.</div>|<div style='width: 30pt'>seq</div>|<div style='width: 20pt'>1.6.0</div>
 
 
 ### Credentials

kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala

@@ -103,6 +103,11 @@ case class KyuubiConf(loadSysDefault: Boolean = true) extends Logging {
     sys.env ++ getAllWithPrefix(KYUUBI_ENGINE_ENV_PREFIX, "")
   }
 
+  /** Get all batch conf as map */
+  def getBatchConf(batchType: String): Map[String, String] = {
+    getAllWithPrefix(s"$KYUUBI_BATCH_CONF_PREFIX.${batchType.toLowerCase(Locale.ROOT)}", "")
+  }
+
   /**
    * Retrieve key-value pairs from [[KyuubiConf]] starting with `dropped.remainder`, and put them to
    * the result map with the `dropped` of key being dropped.
@@ -165,6 +170,7 @@ object KyuubiConf {
   final val KYUUBI_CONF_FILE_NAME = "kyuubi-defaults.conf"
   final val KYUUBI_HOME = "KYUUBI_HOME"
   final val KYUUBI_ENGINE_ENV_PREFIX = "kyuubi.engineEnv"
+  final val KYUUBI_BATCH_CONF_PREFIX = "kyuubi.batchConf"
 
   val kyuubiConfEntries: java.util.Map[String, ConfigEntry[_]] =
     java.util.Collections.synchronizedMap(new java.util.HashMap[String, ConfigEntry[_]]())
@@ -813,6 +819,21 @@ object KyuubiConf {
       .timeConf
       .createWithDefaultString("PT5S")
 
+  val BATCH_CONF_IGNORE_LIST: ConfigEntry[Seq[String]] =
+    buildConf("kyuubi.batch.conf.ignore.list")
+      .doc("A comma separated list of ignored keys for batch conf. If the batch conf contains" +
+        " any of them, the key and the corresponding value will be removed silently during batch" +
+        " job submission." +
+        " Note that this rule is for server-side protection defined via administrators to" +
+        " prevent some essential configs from tampering." +
+        " You can also pre-define some config for batch job submission with prefix:" +
+        " kyuubi.batchConf.[batchType]. For example, you can pre-define `spark.master`" +
+        " for spark batch job with key `kyuubi.batchConf.spark.spark.master`.")
+      .version("1.6.0")
+      .stringConf
+      .toSequence()
+      .createWithDefault(Nil)
+
   val SERVER_EXEC_POOL_SIZE: ConfigEntry[Int] =
     buildConf("kyuubi.backend.server.exec.pool.size")
       .doc("Number of threads in the operation execution thread pool of Kyuubi server")

kyuubi-common/src/main/scala/org/apache/kyuubi/session/SessionManager.scala

@@ -162,10 +162,13 @@ abstract class SessionManager(name: String) extends CompositeService(name) {
 
   private var _confRestrictList: Set[String] = _
   private var _confIgnoreList: Set[String] = _
+  private var _batchConfIgnoreList: Set[String] = _
   private lazy val _confRestrictMatchList: Set[String] =
     _confRestrictList.filter(_.endsWith(".*")).map(_.stripSuffix(".*"))
   private lazy val _confIgnoreMatchList: Set[String] =
     _confIgnoreList.filter(_.endsWith(".*")).map(_.stripSuffix(".*"))
+  private lazy val _batchConfIgnoreMatchList: Set[String] =
+    _batchConfIgnoreList.filter(_.endsWith(".*")).map(_.stripSuffix(".*"))
 
   // strip prefix and validate whether if key is restricted, ignored or valid
   def validateKey(key: String, value: String): Option[(String, String)] = {
@@ -206,6 +209,20 @@ abstract class SessionManager(name: String) extends CompositeService(name) {
     case (k, v) => validateKey(k, v)
   }
 
+  // validate whether if a batch key should be ignored
+  def validateBatchKey(key: String, value: String): Option[(String, String)] = {
+    if (_batchConfIgnoreMatchList.exists(key.startsWith(_)) || _batchConfIgnoreList.contains(key)) {
+      warn(s"$key is a ignored batch key according to the server-side configuration")
+      None
+    } else {
+      Some((key, value))
+    }
+  }
+
+  def validateBatchConf(config: Map[String, String]): Map[String, String] = config.flatMap {
+    case (k, v) => validateBatchKey(k, v)
+  }
+
   override def initialize(conf: KyuubiConf): Unit = synchronized {
     addService(operationManager)
     initOperationLogRootDir()
@@ -232,6 +249,7 @@ abstract class SessionManager(name: String) extends CompositeService(name) {
 
     _confRestrictList = conf.get(SESSION_CONF_RESTRICT_LIST).toSet
     _confIgnoreList = conf.get(SESSION_CONF_IGNORE_LIST).toSet
+    _batchConfIgnoreList = conf.get(BATCH_CONF_IGNORE_LIST).toSet
 
     execPool = ThreadUtils.newDaemonQueuedThreadPool(
       poolSize,

kyuubi-common/src/test/scala/org/apache/kyuubi/config/KyuubiConfSuite.scala

@@ -171,4 +171,12 @@ class KyuubiConfSuite extends KyuubiFunSuite {
     kyuubiConf.set(ENGINE_SHARE_LEVEL_SUBDOMAIN.key, path)
     assert(kyuubiConf.get(ENGINE_SHARE_LEVEL_SUBDOMAIN).get == path)
   }
+
+  test("get pre-defined batch conf for different batch types") {
+    val kyuubiConf = KyuubiConf()
+    kyuubiConf.set(s"$KYUUBI_BATCH_CONF_PREFIX.spark.spark.yarn.tags", "kyuubi")
+    kyuubiConf.set(s"$KYUUBI_BATCH_CONF_PREFIX.flink.yarn.tags", "kyuubi")
+    assert(kyuubiConf.getBatchConf("spark") == Map("spark.yarn.tags" -> "kyuubi"))
+    assert(kyuubiConf.getBatchConf("flink") == Map("yarn.tags" -> "kyuubi"))
+  }
 }

kyuubi-server/src/main/scala/org/apache/kyuubi/operation/BatchJobSubmission.scala

@@ -82,11 +82,12 @@ class BatchJobSubmission(session: KyuubiBatchSessionImpl, batchRequest: BatchReq
   private def submitBatchJob(): Unit = {
     builder = Option(batchRequest.batchType).map(_.toUpperCase(Locale.ROOT)) match {
       case Some("SPARK") =>
+        val batchSparkConf = session.sessionConf.getBatchConf("spark")
         new SparkBatchProcessBuilder(
           session.user,
           session.sessionConf,
           session.batchId,
-          batchRequest,
+          batchRequest.copy(conf = batchSparkConf ++ batchRequest.conf),
           getOperationLog)
 
       case _ =>

kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiBatchSessionImpl.scala

@@ -40,8 +40,12 @@ class KyuubiBatchSessionImpl(
   override val handle: SessionHandle = sessionManager.newBatchSessionHandle(protocol)
   val batchId: String = handle.identifier.toString
 
+  // TODO: Support batch conf advisor
+  override val normalizedConf: Map[String, String] =
+    sessionManager.validateBatchConf(Option(batchRequest.conf).getOrElse(Map.empty))
+
   private[kyuubi] lazy val batchJobSubmissionOp = sessionManager.operationManager
-    .newBatchJobSubmissionOperation(this, batchRequest)
+    .newBatchJobSubmissionOperation(this, batchRequest.copy(conf = normalizedConf))
 
   private val sessionEvent = KyuubiSessionEvent(this)
   EventBus.post(sessionEvent)

kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiBatchYarnClusterSuite.scala

@@ -32,7 +32,10 @@ import org.apache.kyuubi.session.{KyuubiBatchSessionImpl, KyuubiSessionManager}
 class KyuubiBatchYarnClusterSuite extends WithKyuubiServerOnYarn {
   override protected val connectionConf: Map[String, String] = Map.empty
 
-  override protected val kyuubiServerConf: KyuubiConf = KyuubiConf()
+  override protected val kyuubiServerConf: KyuubiConf = {
+    KyuubiConf().set(s"$KYUUBI_BATCH_CONF_PREFIX.spark.spark.master", "yarn")
+      .set(BATCH_CONF_IGNORE_LIST, Seq("spark.master"))
+  }
 
   private def sessionManager(): KyuubiSessionManager =
     server.backendService.sessionManager.asInstanceOf[KyuubiSessionManager]
@@ -47,7 +50,7 @@ class KyuubiBatchYarnClusterSuite extends WithKyuubiServerOnYarn {
       sparkProcessBuilder.mainClass,
       "spark-batch-submission",
       Map(
-        "spark.master" -> "yarn",
+        "spark.master" -> "local",
         s"spark.${ENGINE_SPARK_MAX_LIFETIME.key}" -> "5000",
         s"spark.${ENGINE_CHECK_INTERVAL.key}" -> "1000"),
       Seq.empty[String])