[KYUUBI #1988] Tune initialization of vars depend on system level confs

<!--
Thanks for sending a pull request!

Here are some tips for you:
  1. If this is your first time, please read our contributor guidelines: https://kyuubi.readthedocs.io/en/latest/community/contributions.html
  2. If the PR is related to an issue in https://github.com/apache/incubator-kyuubi/issues, add '[KYUUBI #XXXX]' in your PR title, e.g., '[KYUUBI #XXXX] Your PR title ...'.
  3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][KYUUBI #XXXX] Your PR title ...'.
-->

### _Why are the changes needed?_
<!--
Please clarify why the changes are needed. For instance,
  1. If you add a feature, you can talk about the use case of it.
  2. If you fix a bug, you can clarify why it is a bug.
-->
The `new SparkConf()` and `KyuubiConf()` executes early than UT, and can not re-initialize after changing system properties, which makes conf overwriting does not work.

After the change, the `sparkConf.setIfMissing("spark.sql.catalogImplementation", defaultCat)` work as expected when we overwriting `spark.sql.catalogImplementation` in UT, and Hudi uses `in-memory` catalog to create table.

<img width="891" alt="Xnip2022-03-02_19-10-06" src="https://user-images.githubusercontent.com/26535726/156351805-bb403ee8-0b89-4db2-b91f-d7b2b9957838.png">

### _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

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

Closes #1995 from pan3793/lazy.

Closes #1988

c264a22 [Cheng Pan] Fix
d1a23fb [Cheng Pan] Lazy initialize vars depends on system level confs

Authored-by: Cheng Pan <chengpan@apache.org>
Signed-off-by: ulysses-you <ulyssesyou@apache.org>

Author: pan3793

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/SparkSQLEngine.scala

@@ -71,41 +71,48 @@ case class SparkSQLEngine(spark: SparkSession) extends Serverable("SparkSQLEngin
 
 object SparkSQLEngine extends Logging {
 
-  val sparkConf: SparkConf = new SparkConf()
+  private var _sparkConf: SparkConf = _
 
-  val kyuubiConf: KyuubiConf = KyuubiConf()
+  private var _kyuubiConf: KyuubiConf = _
+
+  def kyuubiConf: KyuubiConf = _kyuubiConf
 
   var currentEngine: Option[SparkSQLEngine] = None
 
-  private val user = currentUser
+  private lazy val user = currentUser
 
   private val countDownLatch = new CountDownLatch(1)
 
+  SignalRegister.registerLogger(logger)
+  setupConf()
+
   def setupConf(): Unit = {
-    val rootDir = sparkConf.getOption("spark.repl.classdir").getOrElse(getLocalDir(sparkConf))
+    _sparkConf = new SparkConf()
+    _kyuubiConf = KyuubiConf()
+    val rootDir = _sparkConf.getOption("spark.repl.classdir").getOrElse(getLocalDir(_sparkConf))
     val outputDir = Utils.createTempDir(root = rootDir, namePrefix = "repl")
-    sparkConf.setIfMissing("spark.sql.execution.topKSortFallbackThreshold", "10000")
-    sparkConf.setIfMissing("spark.sql.legacy.castComplexTypesToString.enabled", "true")
-    sparkConf.setIfMissing("spark.master", "local")
-    sparkConf.setIfMissing("spark.ui.port", "0")
+    _sparkConf.setIfMissing("spark.sql.execution.topKSortFallbackThreshold", "10000")
+    _sparkConf.setIfMissing("spark.sql.legacy.castComplexTypesToString.enabled", "true")
+    _sparkConf.setIfMissing("spark.master", "local")
+    _sparkConf.setIfMissing("spark.ui.port", "0")
     // register the repl's output dir with the file server.
     // see also `spark.repl.classdir`
-    sparkConf.set("spark.repl.class.outputDir", outputDir.toFile.getAbsolutePath)
-    sparkConf.setIfMissing(
+    _sparkConf.set("spark.repl.class.outputDir", outputDir.toFile.getAbsolutePath)
+    _sparkConf.setIfMissing(
       "spark.hadoop.mapreduce.input.fileinputformat.list-status.num-threads",
       "20")
 
     val appName = s"kyuubi_${user}_spark_${Instant.now}"
-    sparkConf.setIfMissing("spark.app.name", appName)
+    _sparkConf.setIfMissing("spark.app.name", appName)
     val defaultCat = if (KyuubiSparkUtil.hiveClassesArePresent) "hive" else "in-memory"
-    sparkConf.setIfMissing("spark.sql.catalogImplementation", defaultCat)
+    _sparkConf.setIfMissing("spark.sql.catalogImplementation", defaultCat)
 
     kyuubiConf.setIfMissing(KyuubiConf.FRONTEND_THRIFT_BINARY_BIND_PORT, 0)
     kyuubiConf.setIfMissing(HA_ZK_CONN_RETRY_POLICY, RetryPolicies.N_TIME.toString)
 
     // Pass kyuubi config from spark with `spark.kyuubi`
     val sparkToKyuubiPrefix = "spark.kyuubi."
-    sparkConf.getAllWithPrefix(sparkToKyuubiPrefix).foreach { case (k, v) =>
+    _sparkConf.getAllWithPrefix(sparkToKyuubiPrefix).foreach { case (k, v) =>
       kyuubiConf.set(s"kyuubi.$k", v)
     }
 
@@ -117,7 +124,7 @@ object SparkSQLEngine extends Logging {
   }
 
   def createSpark(): SparkSession = {
-    val session = SparkSession.builder.config(sparkConf).getOrCreate
+    val session = SparkSession.builder.config(_sparkConf).getOrCreate
     (kyuubiConf.get(ENGINE_INITIALIZE_SQL) ++ kyuubiConf.get(ENGINE_SESSION_INITIALIZE_SQL))
       .foreach { sqlStr =>
         session.sparkContext.setJobGroup(
@@ -174,8 +181,6 @@ object SparkSQLEngine extends Logging {
   }
 
   def main(args: Array[String]): Unit = {
-    SignalRegister.registerLogger(logger)
-    setupConf()
     val startedTime = System.currentTimeMillis()
     val submitTime = kyuubiConf.getOption(KYUUBI_ENGINE_SUBMIT_TIME_KEY) match {
       case Some(t) => t.toLong

externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/WithSparkSQLEngine.scala

@@ -28,7 +28,7 @@ trait WithSparkSQLEngine extends KyuubiFunSuite {
   protected var engine: SparkSQLEngine = _
   // conf will be loaded until start spark engine
   def withKyuubiConf: Map[String, String]
-  val kyuubiConf: KyuubiConf = SparkSQLEngine.kyuubiConf
+  def kyuubiConf: KyuubiConf = SparkSQLEngine.kyuubiConf
 
   protected var connectionUrl: String = _
 
@@ -61,7 +61,6 @@ trait WithSparkSQLEngine extends KyuubiFunSuite {
     System.setProperty("spark.ui.enabled", "false")
     withKyuubiConf.foreach { case (k, v) =>
       System.setProperty(k, v)
-      kyuubiConf.set(k, v)
     }
 
     SparkSession.clearActiveSession()
@@ -82,7 +81,6 @@ trait WithSparkSQLEngine extends KyuubiFunSuite {
     // we need to clean up conf since it's the global config in same jvm.
     withKyuubiConf.foreach { case (k, _) =>
       System.clearProperty(k)
-      kyuubiConf.unset(k)
     }
 
     if (engine != null) {

externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/operation/SparkDeltaOperationSuite.scala

@@ -29,11 +29,4 @@ class SparkDeltaOperationSuite extends WithSparkSQLEngine
   override protected def jdbcUrl: String = getJdbcUrl
 
   override def withKyuubiConf: Map[String, String] = extraConfigs
-
-  override def afterAll(): Unit = {
-    super.afterAll()
-    for ((k, _) <- withKyuubiConf) {
-      System.clearProperty(k)
-    }
-  }
 }

externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/operation/SparkHudiOperationSuite.scala

@@ -27,11 +27,4 @@ class SparkHudiOperationSuite extends WithSparkSQLEngine with HudiMetadataTests
   override protected def jdbcUrl: String = getJdbcUrl
 
   override def withKyuubiConf: Map[String, String] = extraConfigs
-
-  override def afterAll(): Unit = {
-    super.afterAll()
-    for ((k, _) <- withKyuubiConf) {
-      System.clearProperty(k)
-    }
-  }
 }

externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/operation/SparkIcebergOperationSuite.scala

@@ -29,11 +29,4 @@ class SparkIcebergOperationSuite extends WithSparkSQLEngine
   override protected def jdbcUrl: String = getJdbcUrl
 
   override def withKyuubiConf: Map[String, String] = extraConfigs
-
-  override def afterAll(): Unit = {
-    super.afterAll()
-    for ((k, _) <- withKyuubiConf) {
-      System.clearProperty(k)
-    }
-  }
 }