[KYUUBI #7153] Share JAAS configuration for Zookeeper client to avoid server OOM

### Why are the changes needed?

Sharing jaas configuration for zookeeper client with same keytab and principal to avoid server oom due to nested jaas configuration.

fix issue https://github.com/apache/kyuubi/issues/7153

### How was this patch tested?

ut

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

no

Closes #7154 from Z1Wu/fix/comm_reuse_zk_jass.

Closes #7153

3b0169a00 [Cheng Pan] Update kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/zookeeper/ZookeeperClientProvider.scala
5873d12f3 [Cheng Pan] Update kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/zookeeper/ZookeeperClientProvider.scala
0d8a18a4e [wuziyi] nit
ffa7d29fc [wuziyi] [fix] share jaas configuration for zookeeper client with same keytab and principal to avoid server oom due to recursive jaas configuration.

Lead-authored-by: wuziyi <wuziyi02@corp.netease.com>
Co-authored-by: Cheng Pan <pan3793@gmail.com>
Signed-off-by: Cheng Pan <chengpan@apache.org>

.github/workflows/style.yml

@@ -132,9 +132,3 @@ jobs:
         with:
           name: super-linter-log
           path: super-linter.log
-      - name: check bin directory
-        uses: ludeeus/action-shellcheck@1.1.0
-        with:
-          # TODO: enable for all folders step by step
-          scandir: './bin'
-          severity: error

dev/dependencyList

@@ -97,7 +97,7 @@ jetty-util-ajax/9.4.57.v20241219//jetty-util-ajax-9.4.57.v20241219.jar
 jetty-util/9.4.57.v20241219//jetty-util-9.4.57.v20241219.jar
 jline/2.14.6//jline-2.14.6.jar
 jul-to-slf4j/1.7.36//jul-to-slf4j-1.7.36.jar
-kafka-clients/3.5.2//kafka-clients-3.5.2.jar
+kafka-clients/3.9.1//kafka-clients-3.9.1.jar
 kubernetes-client-api/6.13.5//kubernetes-client-api-6.13.5.jar
 kubernetes-client/6.13.5//kubernetes-client-6.13.5.jar
 kubernetes-httpclient-okhttp/6.13.5//kubernetes-httpclient-okhttp-6.13.5.jar

docs/deployment/migration-guide.md

@@ -23,6 +23,8 @@
 
 * Since Kyuubi 1.11, if the engine is running in cluster mode, Kyuubi will respect the `kyuubi.session.engine.startup.waitCompletion` config to determine whether to wait for the engine completion or not. If the engine is running in client mode, Kyuubi will always wait for the engine completion. And for Spark engine, Kyuubi will append the `spark.yarn.submit.waitAppCompletion` and `spark.kubernetes.submission.waitAppCompletion` configs to the engine conf based on the value of `kyuubi.session.engine.startup.waitCompletion`.
 
+* Since Kyuubi 1.11, the configuration `kyuubi.session.engine.spark.initialize.sql` set by the client (via session configuration) is now correctly applied to every session in shared engines (USER, GROUP, SERVER). Previously, only the value set on the server side was applied and only for the first session when the engine started. Now, session-level settings provided by each client are respected.
+
 ## Upgrading from Kyuubi 1.9 to 1.10
 
 * Since Kyuubi 1.10, `beeline` is deprecated and will be removed in the future, please use `kyuubi-beeline` instead.

extensions/spark/kyuubi-spark-authz/pom.xml

@@ -391,6 +391,12 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.kyuubi</groupId>
+            <artifactId>kyuubi-spark-lineage_${scala.binary.version}</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

extensions/spark/kyuubi-spark-authz/src/test/scala/org/apache/kyuubi/plugin/spark/authz/ranger/RangerSparkExtensionSuite.scala

@@ -37,6 +37,8 @@ import org.scalatest.BeforeAndAfterAll
 import org.scalatest.funsuite.AnyFunSuite
 
 import org.apache.kyuubi.Utils
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkSQLLineageParseHelper
 import org.apache.kyuubi.plugin.spark.authz.{AccessControlException, SparkSessionProvider}
 import org.apache.kyuubi.plugin.spark.authz.MysqlContainerEnv
 import org.apache.kyuubi.plugin.spark.authz.RangerTestNamespace._
@@ -1513,4 +1515,31 @@ class HiveCatalogRangerSparkExtensionSuite extends RangerSparkExtensionSuite {
       }
     }
   }
+
+  test("Test view lineage") {
+    def extractLineage(sql: String): Lineage = {
+      val parsed = spark.sessionState.sqlParser.parsePlan(sql)
+      val qe = spark.sessionState.executePlan(parsed)
+      val analyzed = qe.analyzed
+      SparkSQLLineageParseHelper(spark).transformToLineage(0, analyzed).get
+    }
+
+    val db1 = defaultDb
+    val table1 = "table1"
+    val view1 = "view1"
+    withSingleCallEnabled {
+      withCleanTmpResources(Seq((s"$db1.$table1", "table"), (s"$db1.$view1", "view"))) {
+        doAs(admin, sql(s"CREATE TABLE IF NOT EXISTS $db1.$table1 (id int, scope int)"))
+        doAs(admin, sql(s"CREATE VIEW $db1.$view1 AS SELECT * FROM $db1.$table1"))
+
+        val lineage = doAs(
+          admin,
+          extractLineage(s"SELECT id FROM $db1.$view1 WHERE id > 1"))
+        assert(lineage.inputTables.size == 1)
+        assert(lineage.inputTables.head === s"spark_catalog.$db1.$table1")
+        assert(lineage.columnLineage.size == 1)
+        assert(lineage.columnLineage.head.originalColumns.head === s"spark_catalog.$db1.$table1.id")
+      }
+    }
+  }
 }

extensions/spark/kyuubi-spark-lineage/README.md

@@ -26,7 +26,7 @@
 ## Build
 
 ```shell
-build/mvn clean package -DskipTests -pl :kyuubi-spark-lineage_2.12 -am -Dspark.version=3.2.1
+build/mvn clean package -DskipTests -pl :kyuubi-spark-lineage_2.12 -am -Dspark.version=3.5.1
 ```
 
 ### Supported Apache Spark Versions
@@ -37,6 +37,4 @@ build/mvn clean package -DskipTests -pl :kyuubi-spark-lineage_2.12 -am -Dspark.v
 - [x] 3.5.x (default)
 - [x] 3.4.x
 - [x] 3.3.x
-- [x] 3.2.x
-- [x] 3.1.x
 

extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/helper/SparkSQLLineageParseHelper.scala

@@ -46,6 +46,9 @@ trait LineageParser {
   val SUBQUERY_COLUMN_IDENTIFIER = "__subquery__"
   val AGGREGATE_COUNT_COLUMN_IDENTIFIER = "__count__"
   val LOCAL_TABLE_IDENTIFIER = "__local__"
+  val METADATA_COL_ATTR_KEY = "__metadata_col"
+  val ORIGINAL_ROW_ID_VALUE_PREFIX: String = "__original_row_id_"
+  val OPERATION_COLUMN: String = "__row_operation"
 
   type AttributeMap[A] = ListMap[Attribute, A]
 
@@ -307,7 +310,37 @@ trait LineageParser {
         extractColumnsLineage(getQuery(plan), parentColumnsLineage).map { case (k, v) =>
           k.withName(s"$table.${k.name}") -> v
         }
+      case p if p.nodeName == "MergeRows" =>
+        val instructionsOutputs =
+          getField[Seq[Expression]](p, "matchedInstructions")
+            .map(extractInstructionOutputs) ++
+            getField[Seq[Expression]](p, "notMatchedInstructions")
+              .map(extractInstructionOutputs) ++
+            getField[Seq[Expression]](p, "notMatchedBySourceInstructions")
+              .map(extractInstructionOutputs)
+        val nextColumnsLineage = ListMap(p.output.indices.map { index =>
+          val keyAttr = p.output(index)
+          val instructionOutputs = instructionsOutputs.map(_(index))
+          (keyAttr, instructionOutputs)
+        }.collect {
+          case (keyAttr: Attribute, instructionsOutput)
+              if instructionsOutput
+                .exists(_.references.nonEmpty) =>
+            val attributeSet = AttributeSet.apply(instructionsOutput)
+            keyAttr -> attributeSet
+        }: _*)
+        p.children.map(
+          extractColumnsLineage(_, nextColumnsLineage)).reduce(mergeColumnsLineage)
 
+      case p if p.nodeName == "WriteDelta" || p.nodeName == "ReplaceData" =>
+        val table = getV2TableName(getField[NamedRelation](plan, "table"))
+        val query = getQuery(plan)
+        val columnsLineage = extractColumnsLineage(query, parentColumnsLineage)
+        columnsLineage
+          .filter { case (k, _) => !isMetadataAttr(k) }
+          .map { case (k, v) =>
+            k.withName(s"$table.${k.name}") -> v
+          }
       case p if p.nodeName == "MergeIntoTable" =>
         val matchedActions = getField[Seq[MergeAction]](plan, "matchedActions")
         val notMatchedActions = getField[Seq[MergeAction]](plan, "notMatchedActions")
@@ -448,6 +481,13 @@ trait LineageParser {
             })
         }
 
+      // PermanentViewMarker is introduced by kyuubi authz plugin, which is a wrapper of View,
+      // so we just extract the columns lineage from its inner children (original view)
+      case pvm if pvm.nodeName == "PermanentViewMarker" =>
+        pvm.innerChildren.asInstanceOf[Seq[LogicalPlan]]
+          .map(extractColumnsLineage(_, parentColumnsLineage))
+          .reduce(mergeColumnsLineage)
+
       case p: View =>
         if (!p.isTempView && SparkContextHelper.getConf(
             LineageConf.SKIP_PARSING_PERMANENT_VIEW_ENABLED)) {
@@ -507,6 +547,19 @@ trait LineageParser {
       case _ => qualifiedName
     }
   }
+
+  private def isMetadataAttr(attr: Attribute): Boolean = {
+    attr.metadata.contains(METADATA_COL_ATTR_KEY) ||
+    attr.name.startsWith(ORIGINAL_ROW_ID_VALUE_PREFIX) ||
+    attr.name.startsWith(OPERATION_COLUMN)
+  }
+
+  private def extractInstructionOutputs(instruction: Expression): Seq[Expression] = {
+    instruction match {
+      case p if p.nodeName == "Split" => getField[Seq[Expression]](p, "otherOutput")
+      case p => getField[Seq[Expression]](p, "output")
+    }
+  }
 }
 
 case class SparkSQLLineageParseHelper(sparkSession: SparkSession) extends LineageParser

extensions/spark/kyuubi-spark-lineage/src/test/scala/org/apache/kyuubi/plugin/lineage/helper/RowLevelCatalogLineageParserSuite.scala

@@ -0,0 +1,221 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.helper
+
+import org.apache.kyuubi.plugin.lineage.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper.SPARK_RUNTIME_VERSION
+
+class RowLevelCatalogLineageParserSuite extends SparkSQLLineageParserHelperSuite {
+
+  override def catalogName: String = {
+    "org.apache.spark.sql.connector.catalog.InMemoryRowLevelOperationTableCatalog"
+  }
+
+  test("columns lineage extract - WriteDelta") {
+    assume(
+      SPARK_RUNTIME_VERSION >= "3.5",
+      "WriteDelta is only supported in SPARK_RUNTIME_VERSION >= 3.5")
+    val ddls =
+      """
+        |create table v2_catalog.db.target_t(pk int not null, name string, price float)
+        | TBLPROPERTIES ('supports-deltas'='true');
+        |create table v2_catalog.db.source_t(pk int not null, name string, price float)
+        | TBLPROPERTIES ('supports-deltas'='true');
+        |create table v2_catalog.db.pivot_t(pk int not null, price float)
+        | TBLPROPERTIES ('supports-deltas'='true')
+        |""".stripMargin
+    ddls.split(";").filter(_.nonEmpty).foreach(spark.sql(_).collect())
+
+    withTable("v2_catalog.db.target_t", "v2_catalog.db.source_t", "v2_catalog.db.pivot_t") { _ =>
+      val ret0 = extractLineageWithoutExecuting(
+        "MERGE INTO v2_catalog.db.target_t AS target " +
+          "USING v2_catalog.db.source_t AS source " +
+          "ON target.pk = source.pk " +
+          "WHEN MATCHED THEN " +
+          "  UPDATE SET target.name = source.name, target.price = source.price " +
+          "WHEN NOT MATCHED THEN " +
+          "  INSERT (pk, name, price) VALUES (cast(source.pk as int), source.name, source.price)" +
+          "WHEN NOT MATCHED BY SOURCE THEN  UPDATE SET target.name = 'abc' ")
+      assert(ret0 == Lineage(
+        List("v2_catalog.db.source_t", "v2_catalog.db.target_t"),
+        List("v2_catalog.db.target_t"),
+        List(
+          (
+            "v2_catalog.db.target_t.pk",
+            Set("v2_catalog.db.source_t.pk", "v2_catalog.db.target_t.pk")),
+          ("v2_catalog.db.target_t.name", Set("v2_catalog.db.source_t.name")),
+          (
+            "v2_catalog.db.target_t.price",
+            Set("v2_catalog.db.source_t.price", "v2_catalog.db.target_t.price")))))
+
+      val ret1 = extractLineageWithoutExecuting(
+        "MERGE INTO v2_catalog.db.target_t AS target " +
+          "USING v2_catalog.db.source_t AS source " +
+          "ON target.pk = source.pk " +
+          "WHEN MATCHED THEN " +
+          "  UPDATE SET * " +
+          "WHEN NOT MATCHED THEN " +
+          "  INSERT *")
+      assert(ret1 == Lineage(
+        List("v2_catalog.db.source_t"),
+        List("v2_catalog.db.target_t"),
+        List(
+          ("v2_catalog.db.target_t.pk", Set("v2_catalog.db.source_t.pk")),
+          ("v2_catalog.db.target_t.name", Set("v2_catalog.db.source_t.name")),
+          ("v2_catalog.db.target_t.price", Set("v2_catalog.db.source_t.price")))))
+
+      val ret2 = extractLineageWithoutExecuting(
+        "MERGE INTO v2_catalog.db.target_t AS target " +
+          "USING (select a.pk, a.name, b.price " +
+          "from v2_catalog.db.source_t a join " +
+          "v2_catalog.db.pivot_t b) AS source " +
+          "ON target.pk = source.pk " +
+          "WHEN MATCHED THEN " +
+          "  UPDATE SET * " +
+          "WHEN NOT MATCHED THEN " +
+          "  INSERT *")
+
+      assert(ret2 == Lineage(
+        List("v2_catalog.db.source_t", "v2_catalog.db.pivot_t"),
+        List("v2_catalog.db.target_t"),
+        List(
+          ("v2_catalog.db.target_t.pk", Set("v2_catalog.db.source_t.pk")),
+          ("v2_catalog.db.target_t.name", Set("v2_catalog.db.source_t.name")),
+          ("v2_catalog.db.target_t.price", Set("v2_catalog.db.pivot_t.price")))))
+
+      val ret3 = extractLineageWithoutExecuting(
+        "update v2_catalog.db.target_t AS set name='abc' where price < 10 ")
+      assert(ret3 == Lineage(
+        List("v2_catalog.db.target_t"),
+        List("v2_catalog.db.target_t"),
+        List(
+          ("v2_catalog.db.target_t.pk", Set("v2_catalog.db.target_t.pk")),
+          ("v2_catalog.db.target_t.name", Set()),
+          ("v2_catalog.db.target_t.price", Set("v2_catalog.db.target_t.price")))))
+    }
+  }
+
+  test("columns lineage extract - ReplaceData") {
+    assume(
+      SPARK_RUNTIME_VERSION >= "3.5",
+      "ReplaceData[SPARK-43963] for merge into is supported in SPARK_RUNTIME_VERSION >= 3.5")
+    val ddls =
+      """
+        |create table v2_catalog.db.target_t(id int, name string, price float)
+        |create table v2_catalog.db.source_t(id int, name string, price float)
+        |create table v2_catalog.db.pivot_t(id int, price float)
+        |""".stripMargin
+    ddls.split("\n").filter(_.nonEmpty).foreach(spark.sql(_).collect())
+    withTable("v2_catalog.db.target_t", "v2_catalog.db.source_t", "v2_catalog.db.pivot_t") { _ =>
+      val ret0 = extractLineageWithoutExecuting("MERGE INTO v2_catalog.db.target_t AS target " +
+        "USING v2_catalog.db.source_t AS source " +
+        "ON target.id = source.id " +
+        "WHEN MATCHED THEN " +
+        "  UPDATE SET target.name = source.name, target.price = source.price " +
+        "WHEN NOT MATCHED THEN " +
+        "  INSERT (id, name, price) VALUES (cast(source.id as int), source.name, source.price)")
+
+      /**
+       * The ReplaceData operation requires that target records which are read but do not match
+       * any of the MATCHED or NOT MATCHED BY SOURCE clauses also be copied.
+       * (refer to [[RewriteMergeIntoTable#buildReplaceDataMergeRowsPlan]])
+       */
+      assert(ret0 == Lineage(
+        List("v2_catalog.db.source_t", "v2_catalog.db.target_t"),
+        List("v2_catalog.db.target_t"),
+        List(
+          (
+            "v2_catalog.db.target_t.id",
+            Set("v2_catalog.db.source_t.id", "v2_catalog.db.target_t.id")),
+          (
+            "v2_catalog.db.target_t.name",
+            Set("v2_catalog.db.source_t.name", "v2_catalog.db.target_t.name")),
+          (
+            "v2_catalog.db.target_t.price",
+            Set("v2_catalog.db.source_t.price", "v2_catalog.db.target_t.price")))))
+
+      val ret1 = extractLineageWithoutExecuting("MERGE INTO v2_catalog.db.target_t AS target " +
+        "USING v2_catalog.db.source_t AS source " +
+        "ON target.id = source.id " +
+        "WHEN MATCHED THEN " +
+        "  UPDATE SET * " +
+        "WHEN NOT MATCHED THEN " +
+        "  INSERT *")
+      assert(ret1 == Lineage(
+        List("v2_catalog.db.source_t", "v2_catalog.db.target_t"),
+        List("v2_catalog.db.target_t"),
+        List(
+          (
+            "v2_catalog.db.target_t.id",
+            Set("v2_catalog.db.source_t.id", "v2_catalog.db.target_t.id")),
+          (
+            "v2_catalog.db.target_t.name",
+            Set("v2_catalog.db.source_t.name", "v2_catalog.db.target_t.name")),
+          (
+            "v2_catalog.db.target_t.price",
+            Set("v2_catalog.db.source_t.price", "v2_catalog.db.target_t.price")))))
+
+      val ret2 = extractLineageWithoutExecuting("MERGE INTO v2_catalog.db.target_t AS target " +
+        "USING (select a.id, a.name, b.price " +
+        "from v2_catalog.db.source_t a join v2_catalog.db.pivot_t b) AS source " +
+        "ON target.id = source.id " +
+        "WHEN MATCHED THEN " +
+        "  UPDATE SET * " +
+        "WHEN NOT MATCHED THEN " +
+        "  INSERT *")
+
+      assert(ret2 == Lineage(
+        List("v2_catalog.db.source_t", "v2_catalog.db.target_t", "v2_catalog.db.pivot_t"),
+        List("v2_catalog.db.target_t"),
+        List(
+          (
+            "v2_catalog.db.target_t.id",
+            Set("v2_catalog.db.source_t.id", "v2_catalog.db.target_t.id")),
+          (
+            "v2_catalog.db.target_t.name",
+            Set("v2_catalog.db.source_t.name", "v2_catalog.db.target_t.name")),
+          (
+            "v2_catalog.db.target_t.price",
+            Set("v2_catalog.db.pivot_t.price", "v2_catalog.db.target_t.price")))))
+
+      val ret3 = extractLineageWithoutExecuting(
+        "update v2_catalog.db.target_t AS set name='abc' where price < 10 ")
+      // For tables that do not support row-level deletion,
+      // duplicate data of the same group may be included when writing.
+      // plan is:
+      // ReplaceData
+      // +- Project [if ((price#1160 < cast(10 as float))) id#1158 else id#1158 AS id#1163,
+      //    if ((price#1160 < cast(10 as float))) abc else name#1159 AS name#1164,
+      //    if ((price#1160 < cast(10 as float))) price#1160 else price#1160 AS price#1165,
+      //    _partition#1162]
+      // +- RelationV2[id#1158, name#1159, price#1160, _partition#1162]
+      //    v2_catalog.db.target_t v2_catalog.db.target_t
+      assert(ret3 == Lineage(
+        List("v2_catalog.db.target_t"),
+        List("v2_catalog.db.target_t"),
+        List(
+          (
+            "v2_catalog.db.target_t.id",
+            Set("v2_catalog.db.target_t.price", "v2_catalog.db.target_t.id")),
+          (
+            "v2_catalog.db.target_t.name",
+            Set("v2_catalog.db.target_t.price", "v2_catalog.db.target_t.name")),
+          ("v2_catalog.db.target_t.price", Set("v2_catalog.db.target_t.price")))))
+    }
+  }
+}

extensions/spark/kyuubi-spark-lineage/src/test/scala/org/apache/kyuubi/plugin/lineage/helper/SparkSQLLineageParserHelperSuite.scala

@@ -31,12 +31,13 @@ import org.apache.kyuubi.KyuubiFunSuite
 import org.apache.kyuubi.plugin.lineage.Lineage
 import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper.SPARK_RUNTIME_VERSION
 
-class SparkSQLLineageParserHelperSuite extends KyuubiFunSuite
+abstract class SparkSQLLineageParserHelperSuite extends KyuubiFunSuite
   with SparkListenerExtensionTest {
 
-  val catalogName =
-    if (SPARK_RUNTIME_VERSION <= "3.1") "org.apache.spark.sql.connector.InMemoryTableCatalog"
+  def catalogName: String = {
+    if (SPARK_RUNTIME_VERSION <= "3.3") "org.apache.spark.sql.connector.InMemoryTableCatalog"
     else "org.apache.spark.sql.connector.catalog.InMemoryTableCatalog"
+  }
 
   val DEFAULT_CATALOG = LineageConf.DEFAULT_CATALOG
   override protected val catalogImpl: String = "hive"
@@ -169,65 +170,6 @@ class SparkSQLLineageParserHelperSuite extends KyuubiFunSuite
     }
   }
 
-  test("columns lineage extract - MergeIntoTable") {
-    val ddls =
-      """
-        |create table v2_catalog.db.target_t(id int, name string, price float)
-        |create table v2_catalog.db.source_t(id int, name string, price float)
-        |create table v2_catalog.db.pivot_t(id int, price float)
-        |""".stripMargin
-    ddls.split("\n").filter(_.nonEmpty).foreach(spark.sql(_).collect())
-    withTable("v2_catalog.db.target_t", "v2_catalog.db.source_t") { _ =>
-      val ret0 = extractLineageWithoutExecuting("MERGE INTO v2_catalog.db.target_t AS target " +
-        "USING v2_catalog.db.source_t AS source " +
-        "ON target.id = source.id " +
-        "WHEN MATCHED THEN " +
-        "  UPDATE SET target.name = source.name, target.price = source.price " +
-        "WHEN NOT MATCHED THEN " +
-        "  INSERT (id, name, price) VALUES (cast(source.id as int), source.name, source.price)")
-      assert(ret0 == Lineage(
-        List("v2_catalog.db.source_t"),
-        List("v2_catalog.db.target_t"),
-        List(
-          ("v2_catalog.db.target_t.id", Set("v2_catalog.db.source_t.id")),
-          ("v2_catalog.db.target_t.name", Set("v2_catalog.db.source_t.name")),
-          ("v2_catalog.db.target_t.price", Set("v2_catalog.db.source_t.price")))))
-
-      val ret1 = extractLineageWithoutExecuting("MERGE INTO v2_catalog.db.target_t AS target " +
-        "USING v2_catalog.db.source_t AS source " +
-        "ON target.id = source.id " +
-        "WHEN MATCHED THEN " +
-        "  UPDATE SET * " +
-        "WHEN NOT MATCHED THEN " +
-        "  INSERT *")
-      assert(ret1 == Lineage(
-        List("v2_catalog.db.source_t"),
-        List("v2_catalog.db.target_t"),
-        List(
-          ("v2_catalog.db.target_t.id", Set("v2_catalog.db.source_t.id")),
-          ("v2_catalog.db.target_t.name", Set("v2_catalog.db.source_t.name")),
-          ("v2_catalog.db.target_t.price", Set("v2_catalog.db.source_t.price")))))
-
-      val ret2 = extractLineageWithoutExecuting("MERGE INTO v2_catalog.db.target_t AS target " +
-        "USING (select a.id, a.name, b.price " +
-        "from v2_catalog.db.source_t a join v2_catalog.db.pivot_t b) AS source " +
-        "ON target.id = source.id " +
-        "WHEN MATCHED THEN " +
-        "  UPDATE SET * " +
-        "WHEN NOT MATCHED THEN " +
-        "  INSERT *")
-
-      assert(ret2 == Lineage(
-        List("v2_catalog.db.source_t", "v2_catalog.db.pivot_t"),
-        List("v2_catalog.db.target_t"),
-        List(
-          ("v2_catalog.db.target_t.id", Set("v2_catalog.db.source_t.id")),
-          ("v2_catalog.db.target_t.name", Set("v2_catalog.db.source_t.name")),
-          ("v2_catalog.db.target_t.price", Set("v2_catalog.db.pivot_t.price")))))
-    }
-
-  }
-
   test("columns lineage extract - CreateViewCommand") {
     withView("createviewcommand", "createviewcommand1", "createviewcommand2") { _ =>
       val ret0 = extractLineage(
@@ -1451,32 +1393,36 @@ class SparkSQLLineageParserHelperSuite extends KyuubiFunSuite
   test("test directory to table") {
     val inputFile = getClass.getResource("/").getPath + "input_file"
     val sourceFile = File(inputFile).createFile()
-    spark.sql(
-      s"""
-         |CREATE OR REPLACE TEMPORARY VIEW temp_view (
-         | `a` STRING COMMENT '',
-         | `b` STRING COMMENT ''
-         |) USING csv OPTIONS(
-         |  sep='\t',
-         |  path='${sourceFile.path}'
-         |);
-         |""".stripMargin).collect()
+    withView("temp_view") { _ =>
+      {
+        spark.sql(
+          s"""
+             |CREATE OR REPLACE TEMPORARY VIEW temp_view (
+             | `a` STRING COMMENT '',
+             | `b` STRING COMMENT ''
+             |) USING csv OPTIONS(
+             |  sep='\t',
+             |  path='${sourceFile.path}'
+             |);
+             |""".stripMargin).collect()
 
-    val ret0 = extractLineageWithoutExecuting(
-      s"""
-         |INSERT OVERWRITE TABLE test_db.test_table_from_dir
-         |SELECT `a`, `b` FROM temp_view
-         |""".stripMargin)
+        val ret0 = extractLineageWithoutExecuting(
+          s"""
+             |INSERT OVERWRITE TABLE test_db.test_table_from_dir
+             |SELECT `a`, `b` FROM temp_view
+             |""".stripMargin)
 
-    assert(ret0 == Lineage(
-      List(),
-      List(s"spark_catalog.test_db.test_table_from_dir"),
-      List(
-        (s"spark_catalog.test_db.test_table_from_dir.a0", Set()),
-        (s"spark_catalog.test_db.test_table_from_dir.b0", Set()))))
+        assert(ret0 == Lineage(
+          List(),
+          List(s"spark_catalog.test_db.test_table_from_dir"),
+          List(
+            (s"spark_catalog.test_db.test_table_from_dir.a0", Set()),
+            (s"spark_catalog.test_db.test_table_from_dir.b0", Set()))))
+      }
+    }
   }
 
-  private def extractLineageWithoutExecuting(sql: String): Lineage = {
+  protected def extractLineageWithoutExecuting(sql: String): Lineage = {
     val parsed = spark.sessionState.sqlParser.parsePlan(sql)
     val analyzed = spark.sessionState.analyzer.execute(parsed)
     spark.sessionState.analyzer.checkAnalysis(analyzed)

extensions/spark/kyuubi-spark-lineage/src/test/scala/org/apache/kyuubi/plugin/lineage/helper/TableCatalogLineageParserSuite.scala

@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.helper
+
+import org.apache.kyuubi.plugin.lineage.Lineage
+
+class TableCatalogLineageParserSuite extends SparkSQLLineageParserHelperSuite {
+
+  override def catalogName: String = {
+    "org.apache.spark.sql.connector.catalog.InMemoryTableCatalog"
+  }
+
+  test("columns lineage extract - MergeIntoTable") {
+    val ddls =
+      """
+        |create table v2_catalog.db.target_t(id int, name string, price float)
+        |create table v2_catalog.db.source_t(id int, name string, price float)
+        |create table v2_catalog.db.pivot_t(id int, price float)
+        |""".stripMargin
+    ddls.split("\n").filter(_.nonEmpty).foreach(spark.sql(_).collect())
+    withTable("v2_catalog.db.target_t", "v2_catalog.db.source_t", "v2_catalog.db.pivot_t") { _ =>
+      val ret0 = extractLineageWithoutExecuting("MERGE INTO v2_catalog.db.target_t AS target " +
+        "USING v2_catalog.db.source_t AS source " +
+        "ON target.id = source.id " +
+        "WHEN MATCHED THEN " +
+        "  UPDATE SET target.name = source.name, target.price = source.price " +
+        "WHEN NOT MATCHED THEN " +
+        "  INSERT (id, name, price) VALUES (cast(source.id as int), source.name, source.price)")
+      assert(ret0 == Lineage(
+        List("v2_catalog.db.source_t"),
+        List("v2_catalog.db.target_t"),
+        List(
+          ("v2_catalog.db.target_t.id", Set("v2_catalog.db.source_t.id")),
+          ("v2_catalog.db.target_t.name", Set("v2_catalog.db.source_t.name")),
+          ("v2_catalog.db.target_t.price", Set("v2_catalog.db.source_t.price")))))
+
+      val ret1 = extractLineageWithoutExecuting("MERGE INTO v2_catalog.db.target_t AS target " +
+        "USING v2_catalog.db.source_t AS source " +
+        "ON target.id = source.id " +
+        "WHEN MATCHED THEN " +
+        "  UPDATE SET * " +
+        "WHEN NOT MATCHED THEN " +
+        "  INSERT *")
+      assert(ret1 == Lineage(
+        List("v2_catalog.db.source_t"),
+        List("v2_catalog.db.target_t"),
+        List(
+          ("v2_catalog.db.target_t.id", Set("v2_catalog.db.source_t.id")),
+          ("v2_catalog.db.target_t.name", Set("v2_catalog.db.source_t.name")),
+          ("v2_catalog.db.target_t.price", Set("v2_catalog.db.source_t.price")))))
+
+      val ret2 = extractLineageWithoutExecuting("MERGE INTO v2_catalog.db.target_t AS target " +
+        "USING (select a.id, a.name, b.price " +
+        "from v2_catalog.db.source_t a join v2_catalog.db.pivot_t b) AS source " +
+        "ON target.id = source.id " +
+        "WHEN MATCHED THEN " +
+        "  UPDATE SET * " +
+        "WHEN NOT MATCHED THEN " +
+        "  INSERT *")
+
+      assert(ret2 == Lineage(
+        List("v2_catalog.db.source_t", "v2_catalog.db.pivot_t"),
+        List("v2_catalog.db.target_t"),
+        List(
+          ("v2_catalog.db.target_t.id", Set("v2_catalog.db.source_t.id")),
+          ("v2_catalog.db.target_t.name", Set("v2_catalog.db.source_t.name")),
+          ("v2_catalog.db.target_t.price", Set("v2_catalog.db.pivot_t.price")))))
+    }
+
+  }
+}

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

@@ -105,7 +105,7 @@ class SparkSQLSessionManager private (name: String, spark: SparkSession)
     userIsolatedSparkSessionThread.foreach(_.shutdown())
   }
 
-  private def getOrNewSparkSession(user: String): SparkSession = {
+  private def getOrNewSparkSession(user: String, sessionConf: Map[String, String]): SparkSession = {
     if (singleSparkSession) {
       spark
     } else {
@@ -113,8 +113,8 @@ class SparkSQLSessionManager private (name: String, spark: SparkSession)
         // it's unnecessary to create a new spark session in connection share level
         // since the session is only one
         case CONNECTION => spark
-        case USER => newSparkSession(spark)
-        case GROUP | SERVER if userIsolatedSparkSession => newSparkSession(spark)
+        case USER => newSparkSession(spark, sessionConf)
+        case GROUP | SERVER if userIsolatedSparkSession => newSparkSession(spark, sessionConf)
         case GROUP | SERVER =>
           userIsolatedCacheLock.synchronized {
             if (userIsolatedCache.containsKey(user)) {
@@ -123,7 +123,7 @@ class SparkSQLSessionManager private (name: String, spark: SparkSession)
               userIsolatedCache.get(user)
             } else {
               userIsolatedCacheCount.put(user, (1, System.currentTimeMillis()))
-              val newSession = newSparkSession(spark)
+              val newSession = newSparkSession(spark, sessionConf)
               userIsolatedCache.put(user, newSession)
               newSession
             }
@@ -132,11 +132,16 @@ class SparkSQLSessionManager private (name: String, spark: SparkSession)
     }
   }
 
-  private def newSparkSession(rootSparkSession: SparkSession): SparkSession = {
+  private def newSparkSession(
+      rootSparkSession: SparkSession,
+      sessionConf: Map[String, String]): SparkSession = {
     val newSparkSession = rootSparkSession.newSession()
     KyuubiSparkUtil.initializeSparkSession(
       newSparkSession,
-      conf.get(ENGINE_SESSION_SPARK_INITIALIZE_SQL))
+      sessionConf.get(ENGINE_SESSION_SPARK_INITIALIZE_SQL.key)
+        .filter(_.nonEmpty)
+        .map(_.split(";").toSeq)
+        .getOrElse(conf.get(ENGINE_SESSION_SPARK_INITIALIZE_SQL)))
     newSparkSession
   }
 
@@ -150,7 +155,7 @@ class SparkSQLSessionManager private (name: String, spark: SparkSession)
       getSessionOption).getOrElse {
       val sparkSession =
         try {
-          getOrNewSparkSession(user)
+          getOrNewSparkSession(user, conf)
         } catch {
           case e: Exception => throw KyuubiSQLException(e)
         }
@@ -222,4 +227,8 @@ class SparkSQLSessionManager private (name: String, spark: SparkSession)
       opHandle: OperationHandle): Path = {
     new Path(getSessionResultSavePath(sessionHandle), opHandle.identifier.toString)
   }
+
+  override private[kyuubi] def isEngineContextStopped = {
+    spark.sparkContext.isStopped
+  }
 }

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

@@ -24,6 +24,7 @@ import org.apache.spark.sql.{AnalysisException, SparkSession}
 import org.apache.spark.ui.SparkUIUtils.formatDuration
 
 import org.apache.kyuubi.KyuubiSQLException
+import org.apache.kyuubi.config.KyuubiConf.SESSION_IDLE_TIMEOUT
 import org.apache.kyuubi.config.KyuubiReservedKeys.KYUUBI_SESSION_HANDLE_KEY
 import org.apache.kyuubi.engine.spark.events.SessionEvent
 import org.apache.kyuubi.engine.spark.operation.SparkSQLOperationManager
@@ -57,6 +58,13 @@ class SparkSessionImpl(
     }
   }
 
+  override val sessionIdleTimeoutThreshold: Long = {
+    conf.get(SESSION_IDLE_TIMEOUT.key)
+      .map(_.toLong)
+      .getOrElse(
+        sessionManager.getConf.get(SESSION_IDLE_TIMEOUT))
+  }
+
   private val sessionEvent = SessionEvent(this)
 
   override def open(): Unit = {

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

@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.engine.spark.session
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.kyuubi.config.KyuubiConf._
+import org.apache.kyuubi.engine.spark.WithSparkSQLEngine
+import org.apache.kyuubi.operation.HiveJDBCTestHelper
+import org.apache.kyuubi.shaded.hive.service.rpc.thrift.{TExecuteStatementReq, TFetchResultsReq, TOpenSessionReq}
+
+class MultiSessionSuiteInitSQLSuite extends WithSparkSQLEngine with HiveJDBCTestHelper {
+
+  override def withKyuubiConf: Map[String, String] = {
+    Map(
+      ENGINE_SHARE_LEVEL.key -> "SERVER",
+      ENGINE_SINGLE_SPARK_SESSION.key -> "false")
+  }
+
+  override protected def jdbcUrl: String =
+    s"jdbc:hive2://${engine.frontendServices.head.connectionUrl}/;#spark.ui.enabled=false"
+
+  test("isolated user spark session") {
+    Seq("abc", "xyz").foreach { value =>
+      withThriftClient(Some(user)) { client =>
+        val req = new TOpenSessionReq()
+        req.setUsername("user")
+        req.setPassword("anonymous")
+        req.setConfiguration(Map(
+          ENGINE_SHARE_LEVEL.key -> "SERVER",
+          ENGINE_SINGLE_SPARK_SESSION.key -> "false",
+          ENGINE_SESSION_SPARK_INITIALIZE_SQL.key -> s"SET varA=$value").asJava)
+        val tOpenSessionResp = client.OpenSession(req)
+        val tExecuteStatementReq = new TExecuteStatementReq()
+        tExecuteStatementReq.setSessionHandle(tOpenSessionResp.getSessionHandle)
+        tExecuteStatementReq.setStatement("SELECT '${varA}'")
+        tExecuteStatementReq.setRunAsync(false)
+        val tExecuteStatementResp = client.ExecuteStatement(tExecuteStatementReq)
+
+        val operationHandle = tExecuteStatementResp.getOperationHandle
+        val tFetchResultsReq = new TFetchResultsReq()
+        tFetchResultsReq.setOperationHandle(operationHandle)
+        tFetchResultsReq.setFetchType(0)
+        tFetchResultsReq.setMaxRows(1)
+        val tFetchResultsResp = client.FetchResults(tFetchResultsReq)
+        val ret = tFetchResultsResp.getResults.getColumns.get(0).getStringVal.getValues.get(0)
+        assert(ret === value)
+      }
+    }
+  }
+}

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

@@ -345,10 +345,15 @@ abstract class SessionManager(name: String) extends CompositeService(name) {
     if (idleTimeout > 0) {
       val checkTask = new Runnable {
         override def run(): Unit = {
-          if (!shutdown && System.currentTimeMillis() - latestLogoutTime > idleTimeout &&
-            getActiveUserSessionCount <= 0) {
-            info(s"Idled for more than $idleTimeout ms, terminating")
-            stop()
+          if (!shutdown) {
+            if (System.currentTimeMillis() - latestLogoutTime > idleTimeout &&
+              getActiveUserSessionCount <= 0) {
+              info(s"Idled for more than $idleTimeout ms, terminating")
+              stop()
+            } else if (isEngineContextStopped) {
+              error(s"Engine's SparkContext is stopped, terminating")
+              stop()
+            }
           }
         }
       }
@@ -360,4 +365,6 @@ abstract class SessionManager(name: String) extends CompositeService(name) {
         TimeUnit.MILLISECONDS)
     }
   }
+
+  private[kyuubi] def isEngineContextStopped: Boolean = false
 }

kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/zookeeper/ZookeeperClientProvider.scala

@@ -19,6 +19,7 @@ package org.apache.kyuubi.ha.client.zookeeper
 
 import java.io.{File, IOException}
 import java.nio.charset.StandardCharsets
+import java.util.concurrent.ConcurrentHashMap
 import javax.security.auth.login.Configuration
 
 import scala.util.Random
@@ -38,6 +39,13 @@ import org.apache.kyuubi.util.reflect.DynConstructors
 
 object ZookeeperClientProvider extends Logging {
 
+  /**
+   * Share JAAS configuration for Zookeeper client with same keytab and principal to
+   * avoid server OOM due to each new JAAS configuration references the previous instance.
+   * See KYUUBI #7154 for more details.
+   */
+  val jaasConfigurationCache = new ConcurrentHashMap[(String, String), Configuration]()
+
   /**
    * Create a [[CuratorFramework]] instance to be used as the ZooKeeper client
    * Use the [[ZookeeperACLProvider]] to create appropriate ACLs
@@ -113,22 +121,26 @@ object ZookeeperClientProvider extends Logging {
           System.setProperty("zookeeper.server.principal", zkServerPrincipal)
         }
         val zkClientPrincipal = KyuubiHadoopUtils.getServerPrincipal(principal)
-        // HDFS-16591 makes breaking change on JaasConfiguration
-        val jaasConf = DynConstructors.builder()
-          .impl( // Hadoop 3.3.5 and above
-            "org.apache.hadoop.security.authentication.util.JaasConfiguration",
-            classOf[String],
-            classOf[String],
-            classOf[String])
-          .impl( // Hadoop 3.3.4 and previous
-            // scalastyle:off
-            "org.apache.hadoop.security.token.delegation.ZKDelegationTokenSecretManager$JaasConfiguration",
-            // scalastyle:on
-            classOf[String],
-            classOf[String],
-            classOf[String])
-          .build[Configuration]()
-          .newInstance("KyuubiZooKeeperClient", zkClientPrincipal, keytab)
+        val jaasConf = jaasConfigurationCache.computeIfAbsent(
+          (principal, keytab),
+          _ => {
+            // HDFS-16591 makes breaking change on JaasConfiguration
+            DynConstructors.builder()
+              .impl( // Hadoop 3.3.5 and above
+                "org.apache.hadoop.security.authentication.util.JaasConfiguration",
+                classOf[String],
+                classOf[String],
+                classOf[String])
+              .impl( // Hadoop 3.3.4 and previous
+                // scalastyle:off
+                "org.apache.hadoop.security.token.delegation.ZKDelegationTokenSecretManager$JaasConfiguration",
+                // scalastyle:on
+                classOf[String],
+                classOf[String],
+                classOf[String])
+              .build[Configuration]()
+              .newInstance("KyuubiZooKeeperClient", zkClientPrincipal, keytab)
+          })
         Configuration.setConfiguration(jaasConf)
       case _ =>
     }

kyuubi-server/web-ui/package-lock.json

@@ -1598,6 +1598,18 @@
         "node": ">=8"
       }
     },
+    "node_modules/call-bind-apply-helpers": {
+      "version": "1.0.2",
+      "resolved": "https://registry.npmjs.org/call-bind-apply-helpers/-/call-bind-apply-helpers-1.0.2.tgz",
+      "integrity": "sha512-Sp1ablJ0ivDkSzjcaJdxEunN5/XvksFJ2sMBFfq6x0ryhQV/2b/KwFe21cMpmHtPOSij8K99/wSfoEuTObmuMQ==",
+      "dependencies": {
+        "es-errors": "^1.3.0",
+        "function-bind": "^1.1.2"
+      },
+      "engines": {
+        "node": ">= 0.4"
+      }
+    },
     "node_modules/callsites": {
       "version": "3.1.0",
       "resolved": "https://registry.npmjs.org/callsites/-/callsites-3.1.0.tgz",
@@ -1948,6 +1960,19 @@
         "node": ">=12"
       }
     },
+    "node_modules/dunder-proto": {
+      "version": "1.0.1",
+      "resolved": "https://registry.npmjs.org/dunder-proto/-/dunder-proto-1.0.1.tgz",
+      "integrity": "sha512-KIN/nDJBQRcXw0MLVhZE9iQHmG68qAVIBg9CqmUYjmQIhgij9U5MFvrqkUL5FbtyyzZuOeOt0zdeRe4UY7ct+A==",
+      "dependencies": {
+        "call-bind-apply-helpers": "^1.0.1",
+        "es-errors": "^1.3.0",
+        "gopd": "^1.2.0"
+      },
+      "engines": {
+        "node": ">= 0.4"
+      }
+    },
     "node_modules/element-plus": {
       "version": "2.2.13",
       "resolved": "https://registry.npmjs.org/element-plus/-/element-plus-2.2.13.tgz",
@@ -1985,6 +2010,47 @@
         "url": "https://github.com/fb55/entities?sponsor=1"
       }
     },
+    "node_modules/es-define-property": {
+      "version": "1.0.1",
+      "resolved": "https://registry.npmjs.org/es-define-property/-/es-define-property-1.0.1.tgz",
+      "integrity": "sha512-e3nRfgfUZ4rNGL232gUgX06QNyyez04KdjFrF+LTRoOXmrOgFKDg4BCdsjW8EnT69eqdYGmRpJwiPVYNrCaW3g==",
+      "engines": {
+        "node": ">= 0.4"
+      }
+    },
+    "node_modules/es-errors": {
+      "version": "1.3.0",
+      "resolved": "https://registry.npmjs.org/es-errors/-/es-errors-1.3.0.tgz",
+      "integrity": "sha512-Zf5H2Kxt2xjTvbJvP2ZWLEICxA6j+hAmMzIlypy4xcBg1vKVnx89Wy0GbS+kf5cwCVFFzdCFh2XSCFNULS6csw==",
+      "engines": {
+        "node": ">= 0.4"
+      }
+    },
+    "node_modules/es-object-atoms": {
+      "version": "1.1.1",
+      "resolved": "https://registry.npmjs.org/es-object-atoms/-/es-object-atoms-1.1.1.tgz",
+      "integrity": "sha512-FGgH2h8zKNim9ljj7dankFPcICIK9Cp5bm+c2gQSYePhpaG5+esrLODihIorn+Pe6FGJzWhXQotPv73jTaldXA==",
+      "dependencies": {
+        "es-errors": "^1.3.0"
+      },
+      "engines": {
+        "node": ">= 0.4"
+      }
+    },
+    "node_modules/es-set-tostringtag": {
+      "version": "2.1.0",
+      "resolved": "https://registry.npmjs.org/es-set-tostringtag/-/es-set-tostringtag-2.1.0.tgz",
+      "integrity": "sha512-j6vWzfrGVfyXxge+O0x5sh6cvxAog0a/4Rdd2K36zCMV5eJ+/+tOAngRO8cODMNWbVRdVlmGZQL2YS3yR8bIUA==",
+      "dependencies": {
+        "es-errors": "^1.3.0",
+        "get-intrinsic": "^1.2.6",
+        "has-tostringtag": "^1.0.2",
+        "hasown": "^2.0.2"
+      },
+      "engines": {
+        "node": ">= 0.4"
+      }
+    },
     "node_modules/esbuild": {
       "version": "0.18.20",
       "resolved": "https://registry.npmjs.org/esbuild/-/esbuild-0.18.20.tgz",
@@ -2537,12 +2603,14 @@
       }
     },
     "node_modules/form-data": {
-      "version": "4.0.0",
-      "resolved": "https://registry.npmjs.org/form-data/-/form-data-4.0.0.tgz",
-      "integrity": "sha512-ETEklSGi5t0QMZuiXoA/Q6vcnxcLQP5vdugSpuAyi6SVGi2clPPp+xgEhuMaHC+zGgn31Kd235W35f7Hykkaww==",
+      "version": "4.0.4",
+      "resolved": "https://registry.npmjs.org/form-data/-/form-data-4.0.4.tgz",
+      "integrity": "sha512-KrGhL9Q4zjj0kiUt5OO4Mr/A/jlI2jDYs5eHBpYHPcBEVSiipAvn2Ko2HnPe20rmcuuvMHNdZFp+4IlGTMF0Ow==",
       "dependencies": {
         "asynckit": "^0.4.0",
         "combined-stream": "^1.0.8",
+        "es-set-tostringtag": "^2.1.0",
+        "hasown": "^2.0.2",
         "mime-types": "^2.1.12"
       },
       "engines": {
@@ -2569,6 +2637,14 @@
         "node": "^8.16.0 || ^10.6.0 || >=11.0.0"
       }
     },
+    "node_modules/function-bind": {
+      "version": "1.1.2",
+      "resolved": "https://registry.npmjs.org/function-bind/-/function-bind-1.1.2.tgz",
+      "integrity": "sha512-7XHNxH7qX9xG5mIwxkhumTox/MIRNcOgDrxWsMt2pAr23WHp6MrRlN7FBSFpCpr+oVO0F744iUgR82nJMfG2SA==",
+      "funding": {
+        "url": "https://github.com/sponsors/ljharb"
+      }
+    },
     "node_modules/functional-red-black-tree": {
       "version": "1.0.1",
       "resolved": "https://registry.npmjs.org/functional-red-black-tree/-/functional-red-black-tree-1.0.1.tgz",
@@ -2584,6 +2660,41 @@
         "node": "*"
       }
     },
+    "node_modules/get-intrinsic": {
+      "version": "1.3.0",
+      "resolved": "https://registry.npmjs.org/get-intrinsic/-/get-intrinsic-1.3.0.tgz",
+      "integrity": "sha512-9fSjSaos/fRIVIp+xSJlE6lfwhES7LNtKaCBIamHsjr2na1BiABJPo0mOjjz8GJDURarmCPGqaiVg5mfjb98CQ==",
+      "dependencies": {
+        "call-bind-apply-helpers": "^1.0.2",
+        "es-define-property": "^1.0.1",
+        "es-errors": "^1.3.0",
+        "es-object-atoms": "^1.1.1",
+        "function-bind": "^1.1.2",
+        "get-proto": "^1.0.1",
+        "gopd": "^1.2.0",
+        "has-symbols": "^1.1.0",
+        "hasown": "^2.0.2",
+        "math-intrinsics": "^1.1.0"
+      },
+      "engines": {
+        "node": ">= 0.4"
+      },
+      "funding": {
+        "url": "https://github.com/sponsors/ljharb"
+      }
+    },
+    "node_modules/get-proto": {
+      "version": "1.0.1",
+      "resolved": "https://registry.npmjs.org/get-proto/-/get-proto-1.0.1.tgz",
+      "integrity": "sha512-sTSfBjoXBp89JvIKIefqw7U2CCebsc74kiY6awiGogKtoSGbgjYE/G/+l9sF3MWFPNc9IcoOC4ODfKHfxFmp0g==",
+      "dependencies": {
+        "dunder-proto": "^1.0.1",
+        "es-object-atoms": "^1.0.0"
+      },
+      "engines": {
+        "node": ">= 0.4"
+      }
+    },
     "node_modules/get-stdin": {
       "version": "8.0.0",
       "resolved": "https://registry.npmjs.org/get-stdin/-/get-stdin-8.0.0.tgz",
@@ -2662,6 +2773,17 @@
         "url": "https://github.com/sponsors/sindresorhus"
       }
     },
+    "node_modules/gopd": {
+      "version": "1.2.0",
+      "resolved": "https://registry.npmjs.org/gopd/-/gopd-1.2.0.tgz",
+      "integrity": "sha512-ZUKRh6/kUFoAiTAtTYPZJ3hw9wNxx+BIBOijnlG9PnrJsCcSjs1wyyD6vJpaYtgnzDrKYRSqf3OO6Rfa93xsRg==",
+      "engines": {
+        "node": ">= 0.4"
+      },
+      "funding": {
+        "url": "https://github.com/sponsors/ljharb"
+      }
+    },
     "node_modules/grapheme-splitter": {
       "version": "1.0.4",
       "resolved": "https://registry.npmjs.org/grapheme-splitter/-/grapheme-splitter-1.0.4.tgz",
@@ -2677,6 +2799,42 @@
         "node": ">=8"
       }
     },
+    "node_modules/has-symbols": {
+      "version": "1.1.0",
+      "resolved": "https://registry.npmjs.org/has-symbols/-/has-symbols-1.1.0.tgz",
+      "integrity": "sha512-1cDNdwJ2Jaohmb3sg4OmKaMBwuC48sYni5HUw2DvsC8LjGTLK9h+eb1X6RyuOHe4hT0ULCW68iomhjUoKUqlPQ==",
+      "engines": {
+        "node": ">= 0.4"
+      },
+      "funding": {
+        "url": "https://github.com/sponsors/ljharb"
+      }
+    },
+    "node_modules/has-tostringtag": {
+      "version": "1.0.2",
+      "resolved": "https://registry.npmjs.org/has-tostringtag/-/has-tostringtag-1.0.2.tgz",
+      "integrity": "sha512-NqADB8VjPFLM2V0VvHUewwwsw0ZWBaIdgo+ieHtK3hasLz4qeCRjYcqfB6AQrBggRKppKF8L52/VqdVsO47Dlw==",
+      "dependencies": {
+        "has-symbols": "^1.0.3"
+      },
+      "engines": {
+        "node": ">= 0.4"
+      },
+      "funding": {
+        "url": "https://github.com/sponsors/ljharb"
+      }
+    },
+    "node_modules/hasown": {
+      "version": "2.0.2",
+      "resolved": "https://registry.npmjs.org/hasown/-/hasown-2.0.2.tgz",
+      "integrity": "sha512-0hJU9SCPvmMzIBdZFqNPXWa6dqh7WdH0cII9y+CyS8rG3nL48Bclra9HmKhVVUHyPWNH5Y7xDwAB7bfgSjkUMQ==",
+      "dependencies": {
+        "function-bind": "^1.1.2"
+      },
+      "engines": {
+        "node": ">= 0.4"
+      }
+    },
     "node_modules/html-encoding-sniffer": {
       "version": "3.0.0",
       "resolved": "https://registry.npmjs.org/html-encoding-sniffer/-/html-encoding-sniffer-3.0.0.tgz",
@@ -3090,6 +3248,14 @@
         "semver": "bin/semver.js"
       }
     },
+    "node_modules/math-intrinsics": {
+      "version": "1.1.0",
+      "resolved": "https://registry.npmjs.org/math-intrinsics/-/math-intrinsics-1.1.0.tgz",
+      "integrity": "sha512-/IXtbwEk5HTPyEwyKX6hGkYXxM9nbj64B+ilVJnC/R6B0pH5G4V3b0pVbL7DBj4tkhBAppbQUlf6F6Xl9LHu1g==",
+      "engines": {
+        "node": ">= 0.4"
+      }
+    },
     "node_modules/md5-hex": {
       "version": "3.0.1",
       "resolved": "https://registry.npmjs.org/md5-hex/-/md5-hex-3.0.1.tgz",
@@ -5667,6 +5833,15 @@
       "integrity": "sha512-b6Ilus+c3RrdDk+JhLKUAQfzzgLEPy6wcXqS7f/xe1EETvsDP6GORG7SFuOs6cID5YkqchW/LXZbX5bc8j7ZcQ==",
       "dev": true
     },
+    "call-bind-apply-helpers": {
+      "version": "1.0.2",
+      "resolved": "https://registry.npmjs.org/call-bind-apply-helpers/-/call-bind-apply-helpers-1.0.2.tgz",
+      "integrity": "sha512-Sp1ablJ0ivDkSzjcaJdxEunN5/XvksFJ2sMBFfq6x0ryhQV/2b/KwFe21cMpmHtPOSij8K99/wSfoEuTObmuMQ==",
+      "requires": {
+        "es-errors": "^1.3.0",
+        "function-bind": "^1.1.2"
+      }
+    },
     "callsites": {
       "version": "3.1.0",
       "resolved": "https://registry.npmjs.org/callsites/-/callsites-3.1.0.tgz",
@@ -5935,6 +6110,16 @@
         "webidl-conversions": "^7.0.0"
       }
     },
+    "dunder-proto": {
+      "version": "1.0.1",
+      "resolved": "https://registry.npmjs.org/dunder-proto/-/dunder-proto-1.0.1.tgz",
+      "integrity": "sha512-KIN/nDJBQRcXw0MLVhZE9iQHmG68qAVIBg9CqmUYjmQIhgij9U5MFvrqkUL5FbtyyzZuOeOt0zdeRe4UY7ct+A==",
+      "requires": {
+        "call-bind-apply-helpers": "^1.0.1",
+        "es-errors": "^1.3.0",
+        "gopd": "^1.2.0"
+      }
+    },
     "element-plus": {
       "version": "2.2.13",
       "resolved": "https://registry.npmjs.org/element-plus/-/element-plus-2.2.13.tgz",
@@ -5963,6 +6148,35 @@
       "integrity": "sha512-o4q/dYJlmyjP2zfnaWDUC6A3BQFmVTX+tZPezK7k0GLSU9QYCauscf5Y+qcEPzKL+EixVouYDgLQK5H9GrLpkg==",
       "dev": true
     },
+    "es-define-property": {
+      "version": "1.0.1",
+      "resolved": "https://registry.npmjs.org/es-define-property/-/es-define-property-1.0.1.tgz",
+      "integrity": "sha512-e3nRfgfUZ4rNGL232gUgX06QNyyez04KdjFrF+LTRoOXmrOgFKDg4BCdsjW8EnT69eqdYGmRpJwiPVYNrCaW3g=="
+    },
+    "es-errors": {
+      "version": "1.3.0",
+      "resolved": "https://registry.npmjs.org/es-errors/-/es-errors-1.3.0.tgz",
+      "integrity": "sha512-Zf5H2Kxt2xjTvbJvP2ZWLEICxA6j+hAmMzIlypy4xcBg1vKVnx89Wy0GbS+kf5cwCVFFzdCFh2XSCFNULS6csw=="
+    },
+    "es-object-atoms": {
+      "version": "1.1.1",
+      "resolved": "https://registry.npmjs.org/es-object-atoms/-/es-object-atoms-1.1.1.tgz",
+      "integrity": "sha512-FGgH2h8zKNim9ljj7dankFPcICIK9Cp5bm+c2gQSYePhpaG5+esrLODihIorn+Pe6FGJzWhXQotPv73jTaldXA==",
+      "requires": {
+        "es-errors": "^1.3.0"
+      }
+    },
+    "es-set-tostringtag": {
+      "version": "2.1.0",
+      "resolved": "https://registry.npmjs.org/es-set-tostringtag/-/es-set-tostringtag-2.1.0.tgz",
+      "integrity": "sha512-j6vWzfrGVfyXxge+O0x5sh6cvxAog0a/4Rdd2K36zCMV5eJ+/+tOAngRO8cODMNWbVRdVlmGZQL2YS3yR8bIUA==",
+      "requires": {
+        "es-errors": "^1.3.0",
+        "get-intrinsic": "^1.2.6",
+        "has-tostringtag": "^1.0.2",
+        "hasown": "^2.0.2"
+      }
+    },
     "esbuild": {
       "version": "0.18.20",
       "resolved": "https://registry.npmjs.org/esbuild/-/esbuild-0.18.20.tgz",
@@ -6368,12 +6582,14 @@
       "integrity": "sha512-wWN62YITEaOpSK584EZXJafH1AGpO8RVgElfkuXbTOrPX4fIfOyEpW/CsiNd8JdYrAoOvafRTOEnvsO++qCqFA=="
     },
     "form-data": {
-      "version": "4.0.0",
-      "resolved": "https://registry.npmjs.org/form-data/-/form-data-4.0.0.tgz",
-      "integrity": "sha512-ETEklSGi5t0QMZuiXoA/Q6vcnxcLQP5vdugSpuAyi6SVGi2clPPp+xgEhuMaHC+zGgn31Kd235W35f7Hykkaww==",
+      "version": "4.0.4",
+      "resolved": "https://registry.npmjs.org/form-data/-/form-data-4.0.4.tgz",
+      "integrity": "sha512-KrGhL9Q4zjj0kiUt5OO4Mr/A/jlI2jDYs5eHBpYHPcBEVSiipAvn2Ko2HnPe20rmcuuvMHNdZFp+4IlGTMF0Ow==",
       "requires": {
         "asynckit": "^0.4.0",
         "combined-stream": "^1.0.8",
+        "es-set-tostringtag": "^2.1.0",
+        "hasown": "^2.0.2",
         "mime-types": "^2.1.12"
       }
     },
@@ -6390,6 +6606,11 @@
       "dev": true,
       "optional": true
     },
+    "function-bind": {
+      "version": "1.1.2",
+      "resolved": "https://registry.npmjs.org/function-bind/-/function-bind-1.1.2.tgz",
+      "integrity": "sha512-7XHNxH7qX9xG5mIwxkhumTox/MIRNcOgDrxWsMt2pAr23WHp6MrRlN7FBSFpCpr+oVO0F744iUgR82nJMfG2SA=="
+    },
     "functional-red-black-tree": {
       "version": "1.0.1",
       "resolved": "https://registry.npmjs.org/functional-red-black-tree/-/functional-red-black-tree-1.0.1.tgz",
@@ -6402,6 +6623,32 @@
       "integrity": "sha512-8vXOvuE167CtIc3OyItco7N/dpRtBbYOsPsXCz7X/PMnlGjYjSGuZJgM1Y7mmew7BKf9BqvLX2tnOVy1BBUsxQ==",
       "dev": true
     },
+    "get-intrinsic": {
+      "version": "1.3.0",
+      "resolved": "https://registry.npmjs.org/get-intrinsic/-/get-intrinsic-1.3.0.tgz",
+      "integrity": "sha512-9fSjSaos/fRIVIp+xSJlE6lfwhES7LNtKaCBIamHsjr2na1BiABJPo0mOjjz8GJDURarmCPGqaiVg5mfjb98CQ==",
+      "requires": {
+        "call-bind-apply-helpers": "^1.0.2",
+        "es-define-property": "^1.0.1",
+        "es-errors": "^1.3.0",
+        "es-object-atoms": "^1.1.1",
+        "function-bind": "^1.1.2",
+        "get-proto": "^1.0.1",
+        "gopd": "^1.2.0",
+        "has-symbols": "^1.1.0",
+        "hasown": "^2.0.2",
+        "math-intrinsics": "^1.1.0"
+      }
+    },
+    "get-proto": {
+      "version": "1.0.1",
+      "resolved": "https://registry.npmjs.org/get-proto/-/get-proto-1.0.1.tgz",
+      "integrity": "sha512-sTSfBjoXBp89JvIKIefqw7U2CCebsc74kiY6awiGogKtoSGbgjYE/G/+l9sF3MWFPNc9IcoOC4ODfKHfxFmp0g==",
+      "requires": {
+        "dunder-proto": "^1.0.1",
+        "es-object-atoms": "^1.0.0"
+      }
+    },
     "get-stdin": {
       "version": "8.0.0",
       "resolved": "https://registry.npmjs.org/get-stdin/-/get-stdin-8.0.0.tgz",
@@ -6453,6 +6700,11 @@
         "slash": "^3.0.0"
       }
     },
+    "gopd": {
+      "version": "1.2.0",
+      "resolved": "https://registry.npmjs.org/gopd/-/gopd-1.2.0.tgz",
+      "integrity": "sha512-ZUKRh6/kUFoAiTAtTYPZJ3hw9wNxx+BIBOijnlG9PnrJsCcSjs1wyyD6vJpaYtgnzDrKYRSqf3OO6Rfa93xsRg=="
+    },
     "grapheme-splitter": {
       "version": "1.0.4",
       "resolved": "https://registry.npmjs.org/grapheme-splitter/-/grapheme-splitter-1.0.4.tgz",
@@ -6465,6 +6717,27 @@
       "integrity": "sha512-EykJT/Q1KjTWctppgIAgfSO0tKVuZUjhgMr17kqTumMl6Afv3EISleU7qZUzoXDFTAHTDC4NOoG/ZxU3EvlMPQ==",
       "dev": true
     },
+    "has-symbols": {
+      "version": "1.1.0",
+      "resolved": "https://registry.npmjs.org/has-symbols/-/has-symbols-1.1.0.tgz",
+      "integrity": "sha512-1cDNdwJ2Jaohmb3sg4OmKaMBwuC48sYni5HUw2DvsC8LjGTLK9h+eb1X6RyuOHe4hT0ULCW68iomhjUoKUqlPQ=="
+    },
+    "has-tostringtag": {
+      "version": "1.0.2",
+      "resolved": "https://registry.npmjs.org/has-tostringtag/-/has-tostringtag-1.0.2.tgz",
+      "integrity": "sha512-NqADB8VjPFLM2V0VvHUewwwsw0ZWBaIdgo+ieHtK3hasLz4qeCRjYcqfB6AQrBggRKppKF8L52/VqdVsO47Dlw==",
+      "requires": {
+        "has-symbols": "^1.0.3"
+      }
+    },
+    "hasown": {
+      "version": "2.0.2",
+      "resolved": "https://registry.npmjs.org/hasown/-/hasown-2.0.2.tgz",
+      "integrity": "sha512-0hJU9SCPvmMzIBdZFqNPXWa6dqh7WdH0cII9y+CyS8rG3nL48Bclra9HmKhVVUHyPWNH5Y7xDwAB7bfgSjkUMQ==",
+      "requires": {
+        "function-bind": "^1.1.2"
+      }
+    },
     "html-encoding-sniffer": {
       "version": "3.0.0",
       "resolved": "https://registry.npmjs.org/html-encoding-sniffer/-/html-encoding-sniffer-3.0.0.tgz",
@@ -6784,6 +7057,11 @@
         }
       }
     },
+    "math-intrinsics": {
+      "version": "1.1.0",
+      "resolved": "https://registry.npmjs.org/math-intrinsics/-/math-intrinsics-1.1.0.tgz",
+      "integrity": "sha512-/IXtbwEk5HTPyEwyKX6hGkYXxM9nbj64B+ilVJnC/R6B0pH5G4V3b0pVbL7DBj4tkhBAppbQUlf6F6Xl9LHu1g=="
+    },
     "md5-hex": {
       "version": "3.0.1",
       "resolved": "https://registry.npmjs.org/md5-hex/-/md5-hex-3.0.1.tgz",

pom.xml

@@ -127,7 +127,7 @@
         <antlr.st4.version>4.3.4</antlr.st4.version>
         <apache.archive.dist>https://archive.apache.org/dist</apache.archive.dist>
         <atlas.version>2.3.0</atlas.version>
-        <byte-buddy.version>1.14.15</byte-buddy.version>
+        <byte-buddy.version>1.17.6</byte-buddy.version>
         <bouncycastle.version>1.78</bouncycastle.version>
         <codahale.metrics.version>4.2.30</codahale.metrics.version>
         <commons-cli.version>1.5.0</commons-cli.version>
@@ -172,7 +172,7 @@
         <jetty.version>9.4.57.v20241219</jetty.version>
         <jline.version>2.14.6</jline.version>
         <junit.version>4.13.2</junit.version>
-        <kafka.version>3.5.2</kafka.version>
+        <kafka.version>3.9.1</kafka.version>
         <kubernetes-client.version>6.13.5</kubernetes-client.version>
         <kyuubi-relocated.version>0.6.0</kyuubi-relocated.version>
         <kyuubi-relocated-zookeeper.artifacts>kyuubi-relocated-zookeeper-34</kyuubi-relocated-zookeeper.artifacts>
@@ -1572,6 +1572,22 @@
                     <groupId>org.apache.maven.plugins</groupId>
                     <artifactId>maven-shade-plugin</artifactId>
                     <version>${maven.plugin.shade.version}</version>
+                    <dependencies>
+                        <!--
+                          TODO: Remove ASM version management once upstream change released
+                          https://github.com/apache/maven-shade-plugin/pull/744
+                          -->
+                        <dependency>
+                            <groupId>org.ow2.asm</groupId>
+                            <artifactId>asm</artifactId>
+                            <version>9.8</version>
+                        </dependency>
+                        <dependency>
+                            <groupId>org.ow2.asm</groupId>
+                            <artifactId>asm-commons</artifactId>
+                            <version>9.8</version>
+                        </dependency>
+                    </dependencies>
                 </plugin>
 
                 <plugin>