[KYUUBI #2240][SUB-TASK] Skip add metadata manager if frontend does not support rest

### _Why are the changes needed?_

metadata manager is only used for batch session for rest frontend.

metadata manager is a heavy class, it will initialize metadata table and some of thread pool.

### _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.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #3134 from ulysses-you/metadata.

Closes #2240

12f82fc [ulysses-you] seq
de72ed7 [ulysses-you] address comment
5d3eded [ulysses-you] fix
7699852 [ulysses-you] try fix
d496a5e [ulysses-you] style
23d22f8 [ulysses-you] fix
8fadb9e [ulysses-you] style
f61039a [ulysses-you] fix
ea690ef [ulysses-you] address comment
15a25ec [ulysses-you] address comment
9ccdd17 [ulysses-you] Skip add metadata manager if frontend does not support rest

Authored-by: ulysses-you <ulyssesyou18@gmail.com>
Signed-off-by: ulysses-you <ulyssesyou@apache.org>

Author: ulysses-you

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

@@ -45,14 +45,24 @@ class KyuubiSessionManager private (name: String) extends SessionManager(name) {
   // this lazy is must be specified since the conf is null when the class initialization
   lazy val sessionConfAdvisor: SessionConfAdvisor = PluginLoader.loadSessionConfAdvisor(conf)
   val applicationManager = new KyuubiApplicationManager()
-  private val metadataManager = new MetadataManager()
+  private lazy val metadataManager: Option[MetadataManager] = {
+    // Currently, the metadata manager is used by the REST frontend which provides batch job APIs,
+    // so we initialize it only when Kyuubi starts with the REST frontend.
+    if (conf.get(FRONTEND_PROTOCOLS).map(FrontendProtocols.withName)
+        .contains(FrontendProtocols.REST)) {
+      Option(new MetadataManager())
+    } else {
+      None
+    }
+  }
 
   private var limiter: Option[SessionLimiter] = None
 
   override def initialize(conf: KyuubiConf): Unit = {
+    this.conf = conf
     addService(applicationManager)
     addService(credentialsManager)
-    addService(metadataManager)
+    metadataManager.foreach(addService)
     initSessionLimiter(conf)
     super.initialize(conf)
   }
@@ -166,23 +176,23 @@ class KyuubiSessionManager private (name: String) extends SessionManager(name) {
   }
 
   def insertMetadata(metadata: Metadata): Unit = {
-    metadataManager.insertMetadata(metadata)
+    metadataManager.foreach(_.insertMetadata(metadata))
   }
 
   def updateMetadata(metadata: Metadata): Unit = {
-    metadataManager.updateMetadata(metadata)
+    metadataManager.foreach(_.updateMetadata(metadata))
   }
 
   def getMetadataRequestsRetryRef(identifier: String): Option[MetadataRequestsRetryRef] = {
-    Option(metadataManager.getMetadataRequestsRetryRef(identifier))
+    Option(metadataManager.map(_.getMetadataRequestsRetryRef(identifier)).orNull)
   }
 
   def deRegisterMetadataRequestsRetryRef(identifier: String): Unit = {
-    metadataManager.deRegisterRequestsRetryRef(identifier)
+    metadataManager.foreach(_.deRegisterRequestsRetryRef(identifier))
   }
 
   def getBatchFromMetadataStore(batchId: String): Batch = {
-    metadataManager.getBatch(batchId)
+    metadataManager.map(_.getBatch(batchId)).orNull
   }
 
   def getBatchesFromMetadataStore(
@@ -193,16 +203,18 @@ class KyuubiSessionManager private (name: String) extends SessionManager(name) {
       endTime: Long,
       from: Int,
       size: Int): Seq[Batch] = {
-    metadataManager.getBatches(batchType, batchUser, batchState, createTime, endTime, from, size)
+    metadataManager.map(
+      _.getBatches(batchType, batchUser, batchState, createTime, endTime, from, size))
+      .getOrElse(Seq.empty)
   }
 
   def getBatchMetadata(batchId: String): Metadata = {
-    metadataManager.getBatchSessionMetadata(batchId)
+    metadataManager.map(_.getBatchSessionMetadata(batchId)).orNull
   }
 
   @VisibleForTesting
   def cleanupMetadata(identifier: String): Unit = {
-    metadataManager.cleanupMetadataById(identifier)
+    metadataManager.foreach(_.cleanupMetadataById(identifier))
   }
 
   override def start(): Unit = synchronized {
@@ -216,7 +228,7 @@ class KyuubiSessionManager private (name: String) extends SessionManager(name) {
 
   def getBatchSessionsToRecover(kyuubiInstance: String): Seq[KyuubiBatchSessionImpl] = {
     Seq(OperationState.PENDING, OperationState.RUNNING).flatMap { stateToRecover =>
-      metadataManager.getBatchesRecoveryMetadata(
+      metadataManager.map(_.getBatchesRecoveryMetadata(
         stateToRecover.toString,
         kyuubiInstance,
         0,
@@ -236,17 +248,17 @@ class KyuubiSessionManager private (name: String) extends SessionManager(name) {
           metadata.requestConf,
           batchRequest,
           Some(metadata))
-      }
+      }).getOrElse(Seq.empty)
     }
   }
 
   def getPeerInstanceClosedBatchSessions(kyuubiInstance: String): Seq[Metadata] = {
     Seq(OperationState.PENDING, OperationState.RUNNING).flatMap { stateToKill =>
-      metadataManager.getPeerInstanceClosedBatchesMetadata(
+      metadataManager.map(_.getPeerInstanceClosedBatchesMetadata(
         stateToKill.toString,
         kyuubiInstance,
         0,
-        Int.MaxValue)
+        Int.MaxValue)).getOrElse(Seq.empty)
     }
   }