[KYUUBI #6402]: engine.share.level=GROUP enable for a list of hadoop …

…groups

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

@@ -2779,6 +2779,17 @@ object KyuubiConf {
       }
       .createWithDefault("hadoop")
 
+  val PREFERRED_GROUP: OptionalConfigEntry[String] =
+    buildConf("kyuubi.session.preferGroup")
+      .doc("The group name for the group engine launch. This will be checked for the" +
+        "presence of the specified group in the user's allowed groups. If present," +
+        "it will take precedence for GROUP SHARE LEVEL execution. If this is not" +
+        "configured, the session will use the first group name from the list of" +
+        "groups as the primary group.")
+      .version("1.9.3")
+      .stringConf
+      .createOptional
+
   val SERVER_NAME: OptionalConfigEntry[String] =
     buildConf("kyuubi.server.name")
       .doc("The name of Kyuubi Server.")

kyuubi-server/src/main/scala/org/apache/kyuubi/engine/EngineRef.scala

@@ -98,7 +98,10 @@ private[kyuubi] class EngineRef(
   }
 
   // user for launching engine
-  private[kyuubi] val appUser: String = if (doAsEnabled) routingUser else Utils.currentUser
+  private[kyuubi] val appUser: String = shareLevel match {
+    case GROUP => if (doAsEnabled) sessionUser else Utils.currentUser
+    case _ => if (doAsEnabled) routingUser else Utils.currentUser
+  }
 
   @VisibleForTesting
   private[kyuubi] val subdomain: String = conf.get(ENGINE_SHARE_LEVEL_SUBDOMAIN) match {

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

@@ -22,15 +22,29 @@ import java.util.{Map => JMap}
 import org.apache.hadoop.security.UserGroupInformation
 
 import org.apache.kyuubi.Logging
+import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.plugin.GroupProvider
 
 /**
  * Hadoop based group provider, see more information at
  * https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-common/GroupsMapping.html
  */
 class HadoopGroupProvider extends GroupProvider with Logging {
-  override def primaryGroup(user: String, sessionConf: JMap[String, String]): String =
-    groups(user, sessionConf).head
+  override def primaryGroup(user: String, sessionConf: JMap[String, String]): String = {
+    val preferredGroup: Option[String] = Option(sessionConf.get(KyuubiConf.PREFERRED_GROUP.key))
+
+    val userGroups: Array[String] = groups(user, sessionConf)
+
+    val primaryGroup = preferredGroup match {
+      case Some(group) if userGroups.contains(group) => group
+      case None => userGroups.headOption.getOrElse {
+          throw new NoSuchElementException("No groups available for the user")
+        }
+      case Some(group) =>
+        throw new IllegalArgumentException(s"User is not part of the preferred group: $group")
+    }
+    primaryGroup
+  }
 
   override def groups(user: String, sessionConf: JMap[String, String]): Array[String] =
     UserGroupInformation.createRemoteUser(user).getGroupNames match {