[KYUUBI #3946] Config the rest frontend service max worker thread

### _Why are the changes needed?_

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

- [ ] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #3946 from turboFei/config_rest_pool_size.

Closes #3946

30641c1 [fwang12] save

Authored-by: fwang12 <fwang12@ebay.com>
Signed-off-by: fwang12 <fwang12@ebay.com>

docs/deployment/settings.md

@@ -301,6 +301,7 @@ kyuubi.frontend.protocols|THRIFT_BINARY|A comma separated list for all frontend
 kyuubi.frontend.proxy.http.client.ip.header|X-Real-IP|The http header to record the real client ip address. If your server is behind a load balancer or other proxy, the server will see this load balancer or proxy IP address as the client IP address, to get around this common issue, most load balancers or proxies offer the ability to record the real remote IP address in an HTTP header that will be added to the request for other devices to use. Note that, because the header value can be specified to any ip address, so it will not be used for authentication.|string|1.6.0
 kyuubi.frontend.rest.bind.host|<undefined>|Hostname or IP of the machine on which to run the REST frontend service.|string|1.4.0
 kyuubi.frontend.rest.bind.port|10099|Port of the machine on which to run the REST frontend service.|int|1.4.0
+kyuubi.frontend.rest.max.worker.threads|999|Maximum number of threads in the of frontend worker thread pool for the rest frontend service|int|1.6.2
 kyuubi.frontend.ssl.keystore.algorithm|<undefined>|SSL certificate keystore algorithm.|string|1.7.0
 kyuubi.frontend.ssl.keystore.password|<undefined>|SSL certificate keystore password.|string|1.7.0
 kyuubi.frontend.ssl.keystore.path|<undefined>|SSL certificate keystore location.|string|1.7.0

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

@@ -515,6 +515,13 @@ object KyuubiConf {
       .version("1.4.0")
       .fallbackConf(FRONTEND_MAX_WORKER_THREADS)
 
+  val FRONTEND_REST_MAX_WORKER_THREADS: ConfigEntry[Int] =
+    buildConf("kyuubi.frontend.rest.max.worker.threads")
+      .doc("Maximum number of threads in the of frontend worker thread pool for the rest " +
+        "frontend service")
+      .version("1.6.2")
+      .fallbackConf(FRONTEND_MAX_WORKER_THREADS)
+
   val FRONTEND_WORKER_KEEPALIVE_TIME: ConfigEntry[Long] =
     buildConf("kyuubi.frontend.worker.keepalive.time")
       .doc("(deprecated) Keep-alive time (in milliseconds) for an idle worker thread")

kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiRestFrontendService.scala

@@ -30,7 +30,7 @@ import org.eclipse.jetty.servlet.FilterHolder
 
 import org.apache.kyuubi.{KyuubiException, Utils}
 import org.apache.kyuubi.config.KyuubiConf
-import org.apache.kyuubi.config.KyuubiConf.{FRONTEND_REST_BIND_HOST, FRONTEND_REST_BIND_PORT, METADATA_RECOVERY_THREADS}
+import org.apache.kyuubi.config.KyuubiConf.{FRONTEND_REST_BIND_HOST, FRONTEND_REST_BIND_PORT, FRONTEND_REST_MAX_WORKER_THREADS, METADATA_RECOVERY_THREADS}
 import org.apache.kyuubi.server.api.v1.ApiRootResource
 import org.apache.kyuubi.server.http.authentication.{AuthenticationFilter, KyuubiHttpAuthenticationFactory}
 import org.apache.kyuubi.server.ui.JettyServer
@@ -67,7 +67,11 @@ class KyuubiRestFrontendService(override val serverable: Serverable)
 
   override def initialize(conf: KyuubiConf): Unit = synchronized {
     this.conf = conf
-    server = JettyServer(getName, host, conf.get(FRONTEND_REST_BIND_PORT))
+    server = JettyServer(
+      getName,
+      host,
+      conf.get(FRONTEND_REST_BIND_PORT),
+      conf.get(FRONTEND_REST_MAX_WORKER_THREADS))
     super.initialize(conf)
   }
 

kyuubi-server/src/main/scala/org/apache/kyuubi/server/ui/JettyServer.scala

@@ -71,9 +71,8 @@ private[kyuubi] case class JettyServer(
 
 object JettyServer {
 
-  def apply(name: String, host: String, port: Int): JettyServer = {
-    // TODO: Configurable pool size
-    val pool = new QueuedThreadPool()
+  def apply(name: String, host: String, port: Int, poolSize: Int): JettyServer = {
+    val pool = new QueuedThreadPool(poolSize)
     pool.setName(name)
     pool.setDaemon(true)
     val server = new Server(pool)