[KYUUBI #1503] Introduce FlinkThriftBinaryFrontendService to do the further initialization

…

<!--
Thanks for sending a pull request!

Here are some tips for you:
  1. If this is your first time, please read our contributor guidelines: https://kyuubi.readthedocs.io/en/latest/community/contributions.html
  2. If the PR is related to an issue in https://github.com/apache/incubator-kyuubi/issues, add '[KYUUBI #XXXX]' in your PR title, e.g., '[KYUUBI #XXXX] Your PR title ...'.
  3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][KYUUBI #XXXX] Your PR title ...'.
-->

### _Why are the changes needed?_
<!--
Please clarify why the changes are needed. For instance,
  1. If you add a feature, you can talk about the use case of it.
  2. If you fix a bug, you can clarify why it is a bug.
-->

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

Closes #1512 from yanghua/KYUUBI-1503.

Closes #1503

b41fa73 [yanghua] [KYUUBI #1503] Introduce FlinkThriftBinaryFrontendService to do the further initialization

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

Author: yanghua

externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/FlinkThriftBinaryFrontendService.scala

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.engine.flink
+
+import org.apache.kyuubi.ha.client.{EngineServiceDiscovery, ServiceDiscovery}
+import org.apache.kyuubi.service.{Serverable, Service, ThriftBinaryFrontendService}
+
+class FlinkThriftBinaryFrontendService(
+    override val serverable: Serverable)
+  extends ThriftBinaryFrontendService("FlinkThriftBinaryFrontendService") {
+
+  override lazy val discoveryService: Option[Service] = {
+    if (ServiceDiscovery.supportServiceDiscovery(conf)) {
+      Some(new EngineServiceDiscovery(this))
+    } else {
+      None
+    }
+  }
+
+}

externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/SparkThriftBinaryFrontendServiceSuite.scala

@@ -27,6 +27,7 @@ import org.apache.thrift.transport.TSocket
 
 import org.apache.kyuubi.{KyuubiFunSuite, KyuubiSQLException, Utils}
 import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.config.KyuubiConf.{FRONTEND_THRIFT_BINARY_BIND_HOST, FRONTEND_THRIFT_BINARY_BIND_PORT}
 import org.apache.kyuubi.operation.{OperationHandle, OperationType}
 import org.apache.kyuubi.service.ThriftBinaryFrontendService.{FeServiceServerContext, SERVER_VERSION}
 import org.apache.kyuubi.service.authentication.PlainSASLHelper
@@ -117,6 +118,26 @@ class ThriftFrontendServiceSuite extends KyuubiFunSuite {
     assert(errResp.getStatus.getStatusCode === TStatusCode.ERROR_STATUS)
   }
 
+  test("engine connect url use hostname") {
+    val conf = new KyuubiConf()
+      .set(FRONTEND_THRIFT_BINARY_BIND_HOST.key, "localhost")
+      .set(FRONTEND_THRIFT_BINARY_BIND_PORT, 0)
+    val service = new ThriftBinaryFrontendService("DummyThriftBinaryFrontendService") {
+      override val serverable: Serverable = new NoopThriftBinaryFrontendServer
+      override val discoveryService: Option[Service] = None
+    }
+    intercept[IllegalStateException](service.connectionUrl)
+
+    conf.set(KyuubiConf.ENGINE_CONNECTION_URL_USE_HOSTNAME, true)
+    service.initialize(conf)
+    // default use hostname
+    assert(service.connectionUrl.startsWith("localhost"))
+
+    // use ip address
+    conf.set(KyuubiConf.ENGINE_CONNECTION_URL_USE_HOSTNAME, false)
+    assert(service.connectionUrl.startsWith("127.0.0.1"))
+  }
+
   test("open session") {
     withThriftClient { client =>
       val req = new TOpenSessionReq()