Find local InetAddress (#801)

* Find local InetAddress

* Find local InetAddress

* Find local InetAddress

* Find local InetAddress

kyuubi-common/src/main/scala/org/apache/kyuubi/Utils.scala

@@ -18,6 +18,7 @@
 package org.apache.kyuubi
 
 import java.io.{File, InputStreamReader, IOException}
+import java.net.{Inet4Address, InetAddress, NetworkInterface}
 import java.nio.charset.StandardCharsets
 import java.nio.file.{Files, Path, Paths}
 import java.util.{Properties, UUID}
@@ -175,4 +176,32 @@ private[kyuubi] object Utils extends Logging {
   def addShutdownHook(hook: Runnable, priority: Int): Unit = {
     ShutdownHookManager.get().addShutdownHook(hook, priority)
   }
+
+  /**
+   * This block of code is based on Spark's Utils.findLocalInetAddress()
+   */
+  def findLocalInetAddress: InetAddress = {
+    val address = InetAddress.getLocalHost
+    if (address.isLoopbackAddress) {
+      val activeNetworkIFs = NetworkInterface.getNetworkInterfaces.asScala.toSeq
+      val reOrderedNetworkIFs = if (isWindows) activeNetworkIFs else activeNetworkIFs.reverse
+
+      for (ni <- reOrderedNetworkIFs) {
+        val addresses = ni.getInetAddresses.asScala
+          .filterNot(addr => addr.isLinkLocalAddress || addr.isLoopbackAddress).toSeq
+        if (addresses.nonEmpty) {
+          val addr = addresses.find(_.isInstanceOf[Inet4Address]).getOrElse(addresses.head)
+          // because of Inet6Address.toHostName may add interface at the end if it knows about it
+          val strippedAddress = InetAddress.getByAddress(addr.getAddress)
+          // We've found an address that looks reasonable!
+          warn(s"${address.getHostName} was resolved to a loopback address: " +
+            s"${address.getHostAddress}, using ${strippedAddress.getHostAddress}")
+          return strippedAddress
+        }
+      }
+      warn(s"${address.getHostName} was resolved to a loopback address: ${address.getHostAddress}" +
+        " but we couldn't find any external IP address!")
+    }
+    address
+  }
 }

kyuubi-common/src/main/scala/org/apache/kyuubi/service/FrontendService.scala

@@ -30,6 +30,7 @@ import org.apache.thrift.server.{ServerContext, TServer, TServerEventHandler, TT
 import org.apache.thrift.transport.{TServerSocket, TTransport}
 
 import org.apache.kyuubi.{KyuubiException, KyuubiSQLException, Logging}
+import org.apache.kyuubi.Utils
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.operation.{FetchOrientation, OperationHandle}
 import org.apache.kyuubi.service.authentication.KyuubiAuthenticationFactory
@@ -60,7 +61,7 @@ class FrontendService private (name: String, be: BackendService, oomHook: Runnab
     try {
       hadoopConf = KyuubiHadoopUtils.newHadoopConf(conf)
       val serverHost = conf.get(FRONTEND_BIND_HOST)
-      serverAddr = serverHost.map(InetAddress.getByName).getOrElse(InetAddress.getLocalHost)
+      serverAddr = serverHost.map(InetAddress.getByName).getOrElse(Utils.findLocalInetAddress)
       portNum = conf.get(FRONTEND_BIND_PORT)
       val minThreads = conf.get(FRONTEND_MIN_WORKER_THREADS)
       val maxThreads = conf.get(FRONTEND_MAX_WORKER_THREADS)

kyuubi-common/src/test/scala/org/apache/kyuubi/KyuubiFunSuite.scala

@@ -26,7 +26,6 @@ import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, Outcome}
 import org.scalatest.concurrent.Eventually
 import org.scalatest.funsuite.AnyFunSuite
 
-import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.internal.Tests.IS_TESTING
 
 trait KyuubiFunSuite extends AnyFunSuite
@@ -36,7 +35,6 @@ trait KyuubiFunSuite extends AnyFunSuite
   with ThreadAudit
   with Logging {
   // scalastyle:on
-  System.setProperty(KyuubiConf.FRONTEND_BIND_HOST.key, "127.0.0.1")
   override def beforeAll(): Unit = {
     System.setProperty(IS_TESTING.key, "true")
     doThreadPreAudit()

kyuubi-common/src/test/scala/org/apache/kyuubi/UtilsSuite.scala

@@ -18,6 +18,7 @@
 package org.apache.kyuubi
 
 import java.io.{File, IOException}
+import java.net.InetAddress
 import java.nio.file.Files
 import java.security.PrivilegedExceptionAction
 import java.util.Properties
@@ -123,4 +124,13 @@ class UtilsSuite extends KyuubiFunSuite {
     intercept[IllegalArgumentException](Utils.shortVersion("-" + KYUUBI_VERSION))
     intercept[IllegalArgumentException](Utils.majorMinorVersion("-" + KYUUBI_VERSION))
   }
+
+  test("findLocalInetAddress") {
+    val address = InetAddress.getLocalHost
+    if (!address.isLoopbackAddress) {
+      assert(Utils.findLocalInetAddress === InetAddress.getLocalHost)
+    } else {
+      assert(Utils.findLocalInetAddress !== InetAddress.getLocalHost)
+    }
+  }
 }

kyuubi-common/src/test/scala/org/apache/kyuubi/operation/JDBCTestUtils.scala

@@ -25,7 +25,6 @@ import org.apache.thrift.protocol.TBinaryProtocol
 import org.apache.thrift.transport.TSocket
 
 import org.apache.kyuubi.{KyuubiFunSuite, Utils}
-import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.service.authentication.PlainSASLHelper
 
 trait JDBCTestUtils extends KyuubiFunSuite {
@@ -38,10 +37,7 @@ trait JDBCTestUtils extends KyuubiFunSuite {
   private var _sparkHiveConfs: Map[String, String] = Map.empty
   private var _sparkHiveVars: Map[String, String] = Map.empty
   protected def sessionConfigs: Map[String, String] = _sessionConfs
-  protected def sparkHiveConfigs: Map[String, String] = {
-    // TODO: KYUUBI-504: forbid setting FRONTEND_BIND_HOST by connection string in engine side
-    Map(KyuubiConf.FRONTEND_BIND_HOST.key -> "localhost") ++: _sparkHiveConfs
-  }
+  protected def sparkHiveConfigs: Map[String, String] = _sparkHiveConfs
   protected def sparkHiveVars: Map[String, String] = _sparkHiveVars
 
   def withSessionConf[T](

kyuubi-zookeeper/src/main/scala/org/apache/kyuubi/zookeeper/EmbeddedZookeeper.scala

@@ -18,6 +18,7 @@
 package org.apache.kyuubi.zookeeper
 
 import java.io.File
+import java.net.InetAddress
 
 import scala.collection.JavaConverters._
 
@@ -53,7 +54,8 @@ class EmbeddedZookeeper extends AbstractService("EmbeddedZookeeper") {
         "maxSessionTimeout" -> Integer.valueOf(timeout)
       }).toMap[String, Object].asJava
 
-    val hostname = conf.get(ZK_CLIENT_PORT_ADDRESS).orNull
+    val hostname = conf.get(ZK_CLIENT_PORT_ADDRESS).map(InetAddress.getByName)
+      .getOrElse(Utils.findLocalInetAddress).getCanonicalHostName
     spec = new InstanceSpec(
       dataDirectory,
       clientPort,