diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala
index 0d04e84aa28..b78eeec6fc9 100644
--- a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala
+++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala
@@ -26,7 +26,7 @@ import scala.collection.JavaConverters._
 
 import org.apache.kyuubi.{Logging, Utils}
 import org.apache.kyuubi.engine.ShareLevel
-import org.apache.kyuubi.service.authentication.{AuthTypes, SaslQOP}
+import org.apache.kyuubi.service.authentication.{PlainAuthTypes, SaslQOP}
 
 case class KyuubiConf(loadSysDefault: Boolean = true) extends Logging {
   import KyuubiConf._
@@ -136,7 +136,9 @@ case class KyuubiConf(loadSysDefault: Boolean = true) extends Logging {
     FRONTEND_THRIFT_BINARY_BIND_PORT,
     FRONTEND_REST_BIND_HOST,
     FRONTEND_REST_BIND_PORT,
-    AUTHENTICATION_METHOD,
+    AUTHENTICATION_SASL_ENABLED,
+    AUTHENTICATION_SASL_KERBEROS_ENABLED,
+    AUTHENTICATION_SASL_PLAIN_AUTH_TYPE,
     SERVER_KEYTAB,
     SERVER_PRINCIPAL,
     KINIT_INTERVAL)
@@ -372,6 +374,31 @@ object KyuubiConf {
       .version("1.4.0")
       .fallbackConf(FRONTEND_LOGIN_BACKOFF_SLOT_LENGTH)
 
+  val AUTHENTICATION_SASL_ENABLED: ConfigEntry[Boolean] = buildConf("authentication.sasl.enabled")
+    .doc("Whether enable SASL mechanism for authentication")
+    .version("1.4.0")
+    .booleanConf
+    .createWithDefault(true)
+
+  val AUTHENTICATION_SASL_KERBEROS_ENABLED: ConfigEntry[Boolean] =
+    buildConf("authentication.sasl.kerberos.enabled")
+    .doc("Whether enable GSSAPI/kerberos mechanism for SASL authentication")
+    .version("1.4.0")
+    .booleanConf
+    .createWithDefault(false)
+
+  val AUTHENTICATION_SASL_PLAIN_AUTH_TYPE: OptionalConfigEntry[String] =
+    buildConf("authentication.sasl.plain.auth.type")
+      .doc("Client authentication types for PLAIN mechanism.<ul>" +
+        " <li>NONE: no authentication check.</li>" +
+        " <li>CUSTOM: User-defined authentication.</li>" +
+        " <li>LDAP: Lightweight Directory Access Protocol authentication.</li></ul>")
+      .version("1.4.0")
+      .stringConf
+      .transform(_.toUpperCase(Locale.ROOT))
+      .checkValues(PlainAuthTypes.values.map(_.toString))
+      .createOptional
+
   val AUTHENTICATION_METHOD: ConfigEntry[String] = buildConf("authentication")
     .doc("Client authentication types.<ul>" +
       " <li>NOSASL: raw transport.</li>" +
@@ -382,8 +409,8 @@ object KyuubiConf {
     .version("1.0.0")
     .stringConf
     .transform(_.toUpperCase(Locale.ROOT))
-    .checkValues(AuthTypes.values.map(_.toString))
-    .createWithDefault(AuthTypes.NONE.toString)
+    .checkValues(PlainAuthTypes.values.map(_.toString))
+    .createWithDefault(PlainAuthTypes.NONE.toString)
 
   val AUTHENTICATION_CUSTOM_CLASS: OptionalConfigEntry[String] =
     buildConf("authentication.custom.class")
diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/service/authentication/KyuubiAuthenticationFactory.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/service/authentication/KyuubiAuthenticationFactory.scala
index 736b6e7f269..b8b2e1ad5c9 100644
--- a/kyuubi-common/src/main/scala/org/apache/kyuubi/service/authentication/KyuubiAuthenticationFactory.scala
+++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/service/authentication/KyuubiAuthenticationFactory.scala
@@ -32,14 +32,16 @@ import org.apache.thrift.transport.{TTransportException, TTransportFactory}
 import org.apache.kyuubi.KyuubiSQLException
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf._
-import org.apache.kyuubi.service.authentication.AuthTypes._
+import org.apache.kyuubi.service.authentication.PlainAuthTypes._
 
 class KyuubiAuthenticationFactory(conf: KyuubiConf) {
+  private val saslEnabled: Boolean = conf.get(AUTHENTICATION_SASL_ENABLED)
+  private val kerberosEnabled: Boolean = conf.get(AUTHENTICATION_SASL_KERBEROS_ENABLED)
+  private val plainAuthType: Option[PlainAuthType] =
+    conf.get(AUTHENTICATION_SASL_PLAIN_AUTH_TYPE).map(PlainAuthTypes.withName)
 
-  private val authType: AuthType = AuthTypes.withName(conf.get(AUTHENTICATION_METHOD))
-
-  private val saslServer: Option[HadoopThriftAuthBridgeServer] = authType match {
-    case KERBEROS =>
+  private val hadoopAuthServer: Option[HadoopThriftAuthBridgeServer] = {
+    if (saslEnabled && kerberosEnabled) {
       val secretMgr = KyuubiDelegationTokenManager(conf)
       try {
         secretMgr.startThreads()
@@ -47,7 +49,9 @@ class KyuubiAuthenticationFactory(conf: KyuubiConf) {
         case e: IOException => throw new TTransportException("Failed to start token manager", e)
       }
       Some(new HadoopThriftAuthBridgeServer(secretMgr))
-    case _ => None
+    } else {
+      None
+    }
   }
 
   private def getSaslProperties: java.util.Map[String, String] = {
@@ -59,34 +63,40 @@ class KyuubiAuthenticationFactory(conf: KyuubiConf) {
   }
 
   def getTTransportFactory: TTransportFactory = {
-    saslServer match {
-      case Some(server) =>
-        val serverTransportFactory = try {
-          server.createSaslServerTransportFactory(getSaslProperties)
-        } catch {
-          case e: TTransportException => throw new LoginException(e.getMessage)
-        }
+    if (!saslEnabled || (!kerberosEnabled && plainAuthType.isEmpty)) {
+      new TTransportFactory()
+    } else {
+      val kerberosTransportFactory = hadoopAuthServer match {
+        case Some(server) =>
+          val transportFactory = try {
+            server.createSaslServerTransportFactory(getSaslProperties)
+          } catch {
+            case e: TTransportException => throw new LoginException(e.getMessage)
+          }
+          Some(transportFactory)
+
+        case _ => None
+      }
 
-        server.wrapTransportFactory(serverTransportFactory)
+      val transportFactory = plainAuthType.map { authType =>
+        PlainSASLHelper.getTransportFactory(authType.toString, conf, kerberosTransportFactory)
+      }.orElse(kerberosTransportFactory).orNull
 
-      case _ => authType match {
-        case NOSASL => new TTransportFactory
-        case _ => PlainSASLHelper.getTransportFactory(authType.toString, conf)
-      }
+      hadoopAuthServer.map(_.wrapTransportFactory(transportFactory)).getOrElse(transportFactory)
     }
   }
 
-  def getTProcessorFactory(fe: Iface): TProcessorFactory = saslServer match {
+  def getTProcessorFactory(fe: Iface): TProcessorFactory = hadoopAuthServer match {
     case Some(server) => FEServiceProcessorFactory(server, fe)
     case _ => PlainSASLHelper.getProcessFactory(fe)
   }
 
   def getRemoteUser: Option[String] = {
-    saslServer.map(_.getRemoteUser).orElse(Option(TSetIpAddressProcessor.getUserName))
+    hadoopAuthServer.map(_.getRemoteUser).orElse(Option(TSetIpAddressProcessor.getUserName))
   }
 
   def getIpAddress: Option[String] = {
-    saslServer.map(_.getRemoteAddress).map(_.getHostAddress)
+    hadoopAuthServer.map(_.getRemoteAddress).map(_.getHostAddress)
       .orElse(Option(TSetIpAddressProcessor.getUserIpAddress))
   }
 }
diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/service/authentication/AuthTypes.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/service/authentication/PlainAuthTypes.scala
similarity index 88%
rename from kyuubi-common/src/main/scala/org/apache/kyuubi/service/authentication/AuthTypes.scala
rename to kyuubi-common/src/main/scala/org/apache/kyuubi/service/authentication/PlainAuthTypes.scala
index eacebad31ca..81af956ecd3 100644
--- a/kyuubi-common/src/main/scala/org/apache/kyuubi/service/authentication/AuthTypes.scala
+++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/service/authentication/PlainAuthTypes.scala
@@ -17,8 +17,8 @@
 
 package org.apache.kyuubi.service.authentication
 
-object AuthTypes extends Enumeration {
-  type AuthType = Value
+object PlainAuthTypes extends Enumeration {
+  type PlainAuthType = Value
 
-  val NOSASL, NONE, LDAP, KERBEROS, CUSTOM = Value
+  val NONE, LDAP, CUSTOM = Value
 }
diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/service/authentication/PlainSASLHelper.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/service/authentication/PlainSASLHelper.scala
index 07cc501dcf5..2c597ec2215 100644
--- a/kyuubi-common/src/main/scala/org/apache/kyuubi/service/authentication/PlainSASLHelper.scala
+++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/service/authentication/PlainSASLHelper.scala
@@ -71,8 +71,11 @@ object PlainSASLHelper {
     SQLPlainProcessorFactory(service)
   }
 
-  def getTransportFactory(authTypeStr: String, conf: KyuubiConf): TTransportFactory = {
-    val saslFactory = new TSaslServerTransport.Factory()
+  def getTransportFactory(
+      authTypeStr: String,
+      conf: KyuubiConf,
+      transportFactory: Option[TSaslServerTransport.Factory] = None): TTransportFactory = {
+    val saslFactory = transportFactory.getOrElse(new TSaslServerTransport.Factory())
     try {
       val handler = new PlainServerCallbackHandler(authTypeStr, conf)
       val props = new java.util.HashMap[String, String]
diff --git a/kyuubi-common/src/test/scala/org/apache/kyuubi/service/ThriftFrontendServiceSuite.scala b/kyuubi-common/src/test/scala/org/apache/kyuubi/service/ThriftFrontendServiceSuite.scala
index fe990911890..3871f82d6c8 100644
--- a/kyuubi-common/src/test/scala/org/apache/kyuubi/service/ThriftFrontendServiceSuite.scala
+++ b/kyuubi-common/src/test/scala/org/apache/kyuubi/service/ThriftFrontendServiceSuite.scala
@@ -39,6 +39,7 @@ class ThriftFrontendServiceSuite extends KyuubiFunSuite {
   protected val conf = KyuubiConf()
     .set(KyuubiConf.FRONTEND_THRIFT_BINARY_BIND_PORT, 0)
     .set("kyuubi.test.server.should.fail", "false")
+    .set(KyuubiConf.AUTHENTICATION_SASL_PLAIN_AUTH_TYPE, "NONE")
 
   val user: String = System.getProperty("user.name")
   val sessionConf: util.Map[String, String] = new util.HashMap()
diff --git a/kyuubi-common/src/test/scala/org/apache/kyuubi/service/authentication/KyuubiAuthenticationFactorySuite.scala b/kyuubi-common/src/test/scala/org/apache/kyuubi/service/authentication/KyuubiAuthenticationFactorySuite.scala
index 50aa53638ef..6922190aec1 100644
--- a/kyuubi-common/src/test/scala/org/apache/kyuubi/service/authentication/KyuubiAuthenticationFactorySuite.scala
+++ b/kyuubi-common/src/test/scala/org/apache/kyuubi/service/authentication/KyuubiAuthenticationFactorySuite.scala
@@ -25,7 +25,6 @@ import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.service.authentication.PlainSASLServer.SaslPlainProvider
 import org.apache.kyuubi.util.KyuubiHadoopUtils
 
-
 class KyuubiAuthenticationFactorySuite extends KyuubiFunSuite {
   import KyuubiAuthenticationFactory._
 
@@ -45,7 +44,7 @@ class KyuubiAuthenticationFactorySuite extends KyuubiFunSuite {
   }
 
   test("AuthType NONE") {
-    val kyuubiConf = KyuubiConf()
+    val kyuubiConf = KyuubiConf().set(KyuubiConf.AUTHENTICATION_SASL_PLAIN_AUTH_TYPE, "NONE")
     val auth = new KyuubiAuthenticationFactory(kyuubiConf)
     auth.getTTransportFactory
     assert(Security.getProviders.exists(_.isInstanceOf[SaslPlainProvider]))
@@ -55,14 +54,15 @@ class KyuubiAuthenticationFactorySuite extends KyuubiFunSuite {
   }
 
   test("AuthType Other") {
-    val conf = KyuubiConf().set(KyuubiConf.AUTHENTICATION_METHOD, "INVALID")
+    val conf = KyuubiConf().set(KyuubiConf.AUTHENTICATION_SASL_PLAIN_AUTH_TYPE, "INVALID")
     val e = intercept[IllegalArgumentException](new KyuubiAuthenticationFactory(conf))
-    assert(e.getMessage === "The value of kyuubi.authentication should be one of" +
-      " CUSTOM, KERBEROS, LDAP, NONE, NOSASL, but was INVALID")
+    assert(e.getMessage === "The value of kyuubi.authentication.sasl.plain.auth.type should be" +
+      " one of CUSTOM, LDAP, NONE, but was INVALID")
   }
 
   test("AuthType LDAP") {
-    val conf = KyuubiConf().set(KyuubiConf.AUTHENTICATION_METHOD, "LDAP")
+    val conf = KyuubiConf().set(KyuubiConf.AUTHENTICATION_SASL_ENABLED, true)
+      .set(KyuubiConf.AUTHENTICATION_SASL_PLAIN_AUTH_TYPE, "LDAP")
     val authFactory = new KyuubiAuthenticationFactory(conf)
     authFactory.getTTransportFactory
     assert(Security.getProviders.exists(_.isInstanceOf[SaslPlainProvider]))
@@ -70,7 +70,8 @@ class KyuubiAuthenticationFactorySuite extends KyuubiFunSuite {
 
 
   test("AuthType KERBEROS w/o keytab/principal") {
-    val conf = KyuubiConf().set(KyuubiConf.AUTHENTICATION_METHOD, "KERBEROS")
+    val conf = KyuubiConf().set(KyuubiConf.AUTHENTICATION_SASL_ENABLED, true)
+      .set(KyuubiConf.AUTHENTICATION_SASL_KERBEROS_ENABLED, true)
 
     val factory = new KyuubiAuthenticationFactory(conf)
     val e = intercept[LoginException](factory.getTTransportFactory)
diff --git a/kyuubi-common/src/test/scala/org/apache/kyuubi/session/SessionManagerSuite.scala b/kyuubi-common/src/test/scala/org/apache/kyuubi/session/SessionManagerSuite.scala
index 95432d0562b..37856cfc1f3 100644
--- a/kyuubi-common/src/test/scala/org/apache/kyuubi/session/SessionManagerSuite.scala
+++ b/kyuubi-common/src/test/scala/org/apache/kyuubi/session/SessionManagerSuite.scala
@@ -37,6 +37,7 @@ class SessionManagerSuite extends ThriftFrontendServiceSuite with Eventually {
     .set(KyuubiConf.OPERATION_IDLE_TIMEOUT, Duration.ofSeconds(20).toMillis)
     .set(KyuubiConf.SESSION_CONF_RESTRICT_LIST, Seq("spark.*"))
     .set(KyuubiConf.SESSION_CONF_IGNORE_LIST, Seq("session.engine.*"))
+    .set(KyuubiConf.AUTHENTICATION_SASL_PLAIN_AUTH_TYPE, "NONE")
 
   test("close expired operations") {
     withSessionHandle{ (client, handle) =>
diff --git a/kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/ServiceDiscovery.scala b/kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/ServiceDiscovery.scala
index 06335f1812b..5a7a92c3113 100644
--- a/kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/ServiceDiscovery.scala
+++ b/kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/ServiceDiscovery.scala
@@ -283,9 +283,17 @@ object ServiceDiscovery extends Logging {
     confsToPublish += ("hive.server2.thrift.port" -> hostPort(1))
     confsToPublish += ("hive.server2.thrift.sasl.qop" -> conf.get(KyuubiConf.SASL_QOP))
     // Auth specific confs
-    val authenticationMethod = conf.get(KyuubiConf.AUTHENTICATION_METHOD)
-    confsToPublish += ("hive.server2.authentication" -> authenticationMethod)
-    if (authenticationMethod.equalsIgnoreCase("KERBEROS")) {
+    confsToPublish += ("hive.server2.authentication.sasl.enabled" ->
+      conf.get(KyuubiConf.AUTHENTICATION_SASL_ENABLED))
+    confsToPublish += ("hive.server2.authentication.sasl.kerberos.enabled" ->
+      conf.get(KyuubiConf.AUTHENTICATION_SASL_KERBEROS_ENABLED))
+    conf.get(KyuubiConf.AUTHENTICATION_SASL_PLAIN_AUTH_TYPE).foreach { plainAuthType =>
+      confsToPublish += ("hive.server2.authentication.sasl.plain.auth.type" -> plainAuthType)
+    }
+    confsToPublish += ("hive.server2.authentication.sasl.plain.auth.type" ->
+      conf.get(KyuubiConf.AUTHENTICATION_SASL_PLAIN_AUTH_TYPE))
+    if (conf.get(KyuubiConf.AUTHENTICATION_SASL_ENABLED) &&
+      conf.get(KyuubiConf.AUTHENTICATION_SASL_KERBEROS_ENABLED)) {
       confsToPublish += ("hive.server2.authentication.kerberos.principal" ->
         conf.get(KyuubiConf.SERVER_PRINCIPAL).map(KyuubiHadoopUtils.getServerPrincipal)
           .getOrElse(""))