Replace serverOnly with audience and immutable on ConfigEntry

Author: LamiumAmplexicaule

AGENTS.md

@@ -130,7 +130,11 @@ Runtime dependency changes may also require `LICENSE-binary` and `NOTICE` update
 
 Configs are `ConfigEntry` instances in `KyuubiConf` or per-engine config classes.
 
-- Server-only entries call `.serverOnly` on the builder; `KyuubiConf` tracks them in `serverOnlyConfEntries` so they are stripped before engine-side propagation.
+- Each config entry has two independent access-control properties:
+  - **`audience`** — which consumers should receive the config. Set via `.audience(SERVER)`, `.audience(SPARK)`, etc. on the builder. Entries without explicit audience use prefix-based inference (`inferAudience`).
+  - **`immutable`** — whether users can override the config at session time. Set via `.immutable` on the builder. Admin-controlled sources (user defaults, session conf advisors) bypass immutable checks.
+- Server-only entries use `.audience(SERVER)` and `.immutable` on the builder.
+- Engine process builders must use `conf.getEngineConf(EngineType.X)` (not `conf.getAll`) to get configs filtered by audience.
 - `version()` on `ConfigEntry` is the release the key first ships in.
 - Namespace matches scope: `kyuubi.operation.*`, `kyuubi.session.*`, `kyuubi.<engine>.*`.
 - Provide a sensible default; if none is safe, leave it undefined and document the requirement.

kyuubi-common/src/main/scala/org/apache/kyuubi/Utils.scala

@@ -325,17 +325,10 @@ object Utils extends Logging {
   def redactCommandLineArgs(conf: KyuubiConf, commands: Iterable[String]): Iterable[String] = {
     conf.get(SERVER_SECRET_REDACTION_PATTERN) match {
       case Some(redactionPattern) =>
-        var nextKV = false
         commands.map {
-          case PATTERN_FOR_KEY_VALUE_ARG(key, value) if nextKV =>
+          case PATTERN_FOR_KEY_VALUE_ARG(key, value) =>
             val (_, newValue) = redact(redactionPattern, Seq((key, value))).head
-            nextKV = false
             genKeyValuePair(key, newValue)
-
-          case cmd if cmd == CONF =>
-            nextKV = true
-            cmd
-
           case cmd =>
             cmd
         }

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

@@ -0,0 +1,73 @@
+/*
+ * 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.config
+
+import org.apache.kyuubi.engine.EngineType
+
+object ConfigAudience extends Enumeration {
+  type ConfigAudience = Value
+
+  val SERVER, SPARK, FLINK, HIVE, TRINO, JDBC, DATA_AGENT, ALL_ENGINES, ANY = Value
+
+  private val ENGINE_VALUES: Set[Value] = Set(SPARK, FLINK, HIVE, TRINO, JDBC, DATA_AGENT)
+
+  def fromEngineType(engineType: EngineType.EngineType): ConfigAudience = engineType match {
+    case EngineType.SPARK_SQL => SPARK
+    case EngineType.FLINK_SQL => FLINK
+    case EngineType.HIVE_SQL => HIVE
+    case EngineType.TRINO => TRINO
+    case EngineType.JDBC => JDBC
+    case EngineType.DATA_AGENT => DATA_AGENT
+    case _ => throw new IllegalArgumentException(s"Unsupported engine type: $engineType")
+  }
+
+  def isAudienceMatch(
+      configAudience: Set[ConfigAudience],
+      target: ConfigAudience): Boolean = {
+    configAudience.contains(ANY) ||
+    (configAudience.contains(ALL_ENGINES) && ENGINE_VALUES.contains(target)) ||
+    configAudience.contains(target)
+  }
+
+  def inferAudience(key: String): Set[ConfigAudience] = {
+    if (key.startsWith("kyuubi.server.") || key.startsWith("kyuubi.frontend.") ||
+      key.startsWith("kyuubi.backend.server.")) {
+      Set(SERVER)
+    } else if (key.startsWith("kyuubi.engine.spark.") ||
+      key.startsWith("kyuubi.session.engine.spark.") || key.startsWith("spark.")) {
+      Set(SPARK)
+    } else if (key.startsWith("kyuubi.engine.flink.") ||
+      key.startsWith("kyuubi.session.engine.flink.") || key.startsWith("flink.")) {
+      Set(FLINK)
+    } else if (key.startsWith("kyuubi.engine.hive.") ||
+      key.startsWith("kyuubi.session.engine.hive.") || key.startsWith("hive.")) {
+      Set(HIVE)
+    } else if (key.startsWith("kyuubi.engine.trino.") ||
+      key.startsWith("kyuubi.session.engine.trino.") || key.startsWith("trino.")) {
+      Set(TRINO)
+    } else if (key.startsWith("kyuubi.engine.jdbc.") ||
+      key.startsWith("kyuubi.session.engine.jdbc.")) {
+      Set(JDBC)
+    } else if (key.startsWith("kyuubi.engine.data.agent.") ||
+      key.startsWith("kyuubi.session.engine.data-agent.")) {
+      Set(DATA_AGENT)
+    } else {
+      Set(ANY)
+    }
+  }
+}

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

@@ -33,16 +33,22 @@ private[kyuubi] case class ConfigBuilder(key: String) {
   private[config] var _onCreate: Option[ConfigEntry[_] => Unit] = None
   private[config] var _type = ""
   private[config] var _internal = false
-  private[config] var _serverOnly = false
+  private[config] var _audience: Option[Set[ConfigAudience.Value]] = None
+  private[config] var _immutable = false
   private[config] var _alternatives = List.empty[String]
 
   def internal: ConfigBuilder = {
     _internal = true
     this
   }
 
-  def serverOnly: ConfigBuilder = {
-    _serverOnly = true
+  def audience(audiences: ConfigAudience.Value*): ConfigBuilder = {
+    _audience = Some(audiences.toSet)
+    this
+  }
+
+  def immutable: ConfigBuilder = {
+    _immutable = true
     this
   }
 
@@ -138,7 +144,8 @@ private[kyuubi] case class ConfigBuilder(key: String) {
         _doc,
         _version,
         _internal,
-        _serverOnly,
+        _audience,
+        _immutable,
         fallback)
     _onCreate.foreach(_(entry))
     entry
@@ -228,12 +235,12 @@ private[kyuubi] case class TypedConfigBuilder[T](
   /** Turns the config entry into a sequence of values of the underlying type. */
   def toSequence(sp: String = ","): TypedConfigBuilder[Seq[T]] = {
     parent._type = "seq"
-    TypedConfigBuilder(parent, strToSeq(_, fromStr, sp), iterableToStr(_, toStr))
+    TypedConfigBuilder(parent, strToSeq(_, fromStr, sp), iterableToStr(_, toStr, sp))
   }
 
   def toSet(sp: String = ",", skipBlank: Boolean = true): TypedConfigBuilder[Set[T]] = {
     parent._type = "set"
-    TypedConfigBuilder(parent, strToSet(_, fromStr, sp, skipBlank), iterableToStr(_, toStr))
+    TypedConfigBuilder(parent, strToSet(_, fromStr, sp, skipBlank), iterableToStr(_, toStr, sp))
   }
 
   def createOptional: OptionalConfigEntry[T] = {
@@ -246,7 +253,8 @@ private[kyuubi] case class TypedConfigBuilder[T](
       parent._version,
       parent._type,
       parent._internal,
-      parent._serverOnly)
+      parent._audience,
+      parent._immutable)
     parent._onCreate.foreach(_(entry))
     entry
   }
@@ -265,7 +273,8 @@ private[kyuubi] case class TypedConfigBuilder[T](
         parent._version,
         parent._type,
         parent._internal,
-        parent._serverOnly)
+        parent._audience,
+        parent._immutable)
       parent._onCreate.foreach(_(entry))
       entry
   }
@@ -281,7 +290,8 @@ private[kyuubi] case class TypedConfigBuilder[T](
       parent._version,
       parent._type,
       parent._internal,
-      parent._serverOnly)
+      parent._audience,
+      parent._immutable)
     parent._onCreate.foreach(_(entry))
     entry
   }

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

@@ -26,7 +26,8 @@ trait ConfigEntry[T] {
   def version: String
   def typ: String
   def internal: Boolean
-  def serverOnly: Boolean
+  def audience: Option[Set[ConfigAudience.Value]]
+  def immutable: Boolean
 
   def defaultValStr: String
   def defaultVal: Option[T]
@@ -53,7 +54,8 @@ class OptionalConfigEntry[T](
     _version: String,
     _type: String,
     _internal: Boolean,
-    _serverOnly: Boolean) extends ConfigEntry[Option[T]] {
+    _audience: Option[Set[ConfigAudience.Value]],
+    _immutable: Boolean) extends ConfigEntry[Option[T]] {
   override def valueConverter: String => Option[T] = {
     s => Option(rawValueConverter(s))
   }
@@ -84,7 +86,9 @@ class OptionalConfigEntry[T](
 
   override def internal: Boolean = _internal
 
-  override def serverOnly: Boolean = _serverOnly
+  override def audience: Option[Set[ConfigAudience.Value]] = _audience
+
+  override def immutable: Boolean = _immutable
 }
 
 class ConfigEntryWithDefault[T](
@@ -97,7 +101,8 @@ class ConfigEntryWithDefault[T](
     _version: String,
     _type: String,
     _internal: Boolean,
-    _serverOnly: Boolean) extends ConfigEntry[T] {
+    _audience: Option[Set[ConfigAudience.Value]],
+    _immutable: Boolean) extends ConfigEntry[T] {
   override def defaultValStr: String = strConverter(_defaultVal)
 
   override def defaultVal: Option[T] = Option(_defaultVal)
@@ -122,7 +127,9 @@ class ConfigEntryWithDefault[T](
 
   override def internal: Boolean = _internal
 
-  override def serverOnly: Boolean = _serverOnly
+  override def audience: Option[Set[ConfigAudience.Value]] = _audience
+
+  override def immutable: Boolean = _immutable
 }
 
 class ConfigEntryWithDefaultString[T](
@@ -135,7 +142,8 @@ class ConfigEntryWithDefaultString[T](
     _version: String,
     _type: String,
     _internal: Boolean,
-    _serverOnly: Boolean) extends ConfigEntry[T] {
+    _audience: Option[Set[ConfigAudience.Value]],
+    _immutable: Boolean) extends ConfigEntry[T] {
   override def defaultValStr: String = _defaultVal
 
   override def defaultVal: Option[T] = Some(valueConverter(_defaultVal))
@@ -161,7 +169,9 @@ class ConfigEntryWithDefaultString[T](
 
   override def internal: Boolean = _internal
 
-  override def serverOnly: Boolean = _serverOnly
+  override def audience: Option[Set[ConfigAudience.Value]] = _audience
+
+  override def immutable: Boolean = _immutable
 }
 
 class ConfigEntryFallback[T](
@@ -170,7 +180,8 @@ class ConfigEntryFallback[T](
     _doc: String,
     _version: String,
     _internal: Boolean,
-    _serverOnly: Boolean,
+    _audience: Option[Set[ConfigAudience.Value]],
+    _immutable: Boolean,
     fallback: ConfigEntry[T]) extends ConfigEntry[T] {
   override def defaultValStr: String = fallback.defaultValStr
 
@@ -182,6 +193,8 @@ class ConfigEntryFallback[T](
 
   override def key: String = _key
 
+  def fallbackKey: String = fallback.key
+
   override def alternatives: List[String] = _alternatives
 
   override def valueConverter: String => T = fallback.valueConverter
@@ -196,7 +209,9 @@ class ConfigEntryFallback[T](
 
   override def internal: Boolean = _internal
 
-  override def serverOnly: Boolean = _serverOnly
+  override def audience: Option[Set[ConfigAudience.Value]] = _audience
+
+  override def immutable: Boolean = _immutable
 }
 
 object ConfigEntry {