[KYUUBI #5566] InternalRestClient respects `kyuubi.engine.security.en…

…abled` to add HTTP auth header

### _Why are the changes needed?_

`kyuubi.engine.security.enabled` aims to control whether enabled security mechanism internal communication, but the current implementation is not symmetrical, the auth generator ignores the conf and always produces the auth header, but the auth header handler is only activated when conf is enabled, that causes authentication failure when `kyuubi.engine.security.enabled=false`(default value)

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

- [x] [Run test](https://kyuubi.readthedocs.io/en/master/contributing/code/testing.html#running-tests) locally before make a pull request

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

No.

Closes #5566 from pan3793/none-auth.

Closes #5566

d42a4c3 [Cheng Pan] Revert "Extract AnonymousAuthenticationHandler from BasicAuthenticationHandler"
b544343 [Cheng Pan] Extract AnonymousAuthenticationHandler from BasicAuthenticationHandler
75c4b7d [Cheng Pan] InternalRestClient respects `kyuubi.engine.security.enabled` to add HTTP auth header

Authored-by: Cheng Pan <chengpan@apache.org>
Signed-off-by: Cheng Pan <chengpan@apache.org>
(cherry picked from commit 5f53073)
Signed-off-by: Cheng Pan <chengpan@apache.org>

kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/BatchesResource.scala

@@ -58,6 +58,8 @@ private[v1] class BatchesResource extends ApiRequestContext with Logging {
     fe.getConf.get(BATCH_INTERNAL_REST_CLIENT_SOCKET_TIMEOUT).toInt
   private lazy val internalConnectTimeout =
     fe.getConf.get(BATCH_INTERNAL_REST_CLIENT_CONNECT_TIMEOUT).toInt
+  private lazy val internalSecurityEnabled =
+    fe.getConf.get(ENGINE_SECURITY_ENABLED)
 
   private def batchV2Enabled(reqConf: Map[String, String]): Boolean = {
     KyuubiServer.kyuubiServer.getConf.get(BATCH_SUBMITTER_ENABLED) &&
@@ -67,7 +69,12 @@ private[v1] class BatchesResource extends ApiRequestContext with Logging {
   private def getInternalRestClient(kyuubiInstance: String): InternalRestClient = {
     internalRestClients.computeIfAbsent(
       kyuubiInstance,
-      k => new InternalRestClient(k, internalSocketTimeout, internalConnectTimeout))
+      kyuubiInstance =>
+        new InternalRestClient(
+          kyuubiInstance,
+          internalSocketTimeout,
+          internalConnectTimeout,
+          internalSecurityEnabled))
   }
 
   private def sessionManager = fe.be.sessionManager.asInstanceOf[KyuubiSessionManager]

kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/InternalRestClient.scala

@@ -33,7 +33,11 @@ import org.apache.kyuubi.service.authentication.InternalSecurityAccessor
  * @param socketTimeout the socket timeout for http client.
  * @param connectTimeout the connect timeout for http client.
  */
-class InternalRestClient(kyuubiInstance: String, socketTimeout: Int, connectTimeout: Int) {
+class InternalRestClient(
+    kyuubiInstance: String,
+    socketTimeout: Int,
+    connectTimeout: Int,
+    securityEnabled: Boolean) {
   require(
     InternalSecurityAccessor.get() != null,
     "Internal secure access across Kyuubi instances is not enabled")
@@ -59,12 +63,14 @@ class InternalRestClient(kyuubiInstance: String, socketTimeout: Int, connectTime
   }
 
   private def initKyuubiRestClient(): KyuubiRestClient = {
-    KyuubiRestClient.builder(s"http://$kyuubiInstance")
+    val builder = KyuubiRestClient.builder(s"http://$kyuubiInstance")
       .apiVersion(KyuubiRestClient.ApiVersion.V1)
       .socketTimeout(socketTimeout)
       .connectionTimeout(connectTimeout)
-      .authHeaderGenerator(InternalRestClient.internalAuthHeaderGenerator)
-      .build()
+    if (securityEnabled) {
+      builder.authHeaderGenerator(InternalRestClient.internalAuthHeaderGenerator)
+    }
+    builder.build()
   }
 
   private def withAuthUser[T](user: String)(f: => T): T = {
@@ -82,7 +88,7 @@ object InternalRestClient {
     override def initialValue(): String = null
   }
 
-  final val internalAuthHeaderGenerator = new AuthHeaderGenerator {
+  final lazy val internalAuthHeaderGenerator = new AuthHeaderGenerator {
     override def generateAuthHeader(): String = {
       val authUser = AUTH_USER.get()
       require(authUser != null, "The auth user shall be not null")