diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/engine/deploy/yarn/ApplicationMaster.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/engine/deploy/yarn/ApplicationMaster.scala
index 1993009adce..c46dbfa029b 100644
--- a/kyuubi-common/src/main/scala/org/apache/kyuubi/engine/deploy/yarn/ApplicationMaster.scala
+++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/engine/deploy/yarn/ApplicationMaster.scala
@@ -32,7 +32,7 @@ import org.apache.hadoop.yarn.security.AMRMTokenIdentifier
 import org.apache.kyuubi.{KyuubiException, Logging, Utils}
 import org.apache.kyuubi.config.{KyuubiConf, KyuubiReservedKeys}
 import org.apache.kyuubi.service.Serverable
-import org.apache.kyuubi.util.KyuubiHadoopUtils
+import org.apache.kyuubi.util.{IPStackUtils, KyuubiHadoopUtils}
 import org.apache.kyuubi.util.command.CommandLineUtils.confKeyValues
 import org.apache.kyuubi.util.reflect.{DynFields, DynMethods}
 
@@ -184,8 +184,8 @@ object ApplicationMaster extends Logging {
   }
 
   private def resolveHostAndPort(connectionUrl: String): (String, Int) = {
-    val strings = connectionUrl.split(":")
-    (strings(0), strings(1).toInt)
+    val hostPort = IPStackUtils.getHostAndPort(connectionUrl)
+    (hostPort.getHostname, hostPort.getPort)
   }
 
   private def cleanupStagingDir(): Unit = {
diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/service/TFrontendService.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/service/TFrontendService.scala
index 801236f66b2..bbeb2a788a0 100644
--- a/kyuubi-common/src/main/scala/org/apache/kyuubi/service/TFrontendService.scala
+++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/service/TFrontendService.scala
@@ -36,7 +36,7 @@ import org.apache.kyuubi.shaded.hive.service.rpc.thrift._
 import org.apache.kyuubi.shaded.thrift.protocol.TProtocol
 import org.apache.kyuubi.shaded.thrift.server.{ServerContext, TServerEventHandler}
 import org.apache.kyuubi.shaded.thrift.transport.TTransport
-import org.apache.kyuubi.util.{JavaUtils, KyuubiHadoopUtils, NamedThreadFactory}
+import org.apache.kyuubi.util.{IPStackUtils, JavaUtils, KyuubiHadoopUtils, NamedThreadFactory}
 
 /**
  * Apache Thrift based hive-service-rpc base class
@@ -120,7 +120,7 @@ abstract class TFrontendService(name: String)
       case (None, None) => serverAddr.getHostAddress
     }
 
-    host + ":" + actualPort
+    IPStackUtils.concatHostPort(host, actualPort)
   }
 
   protected def getProxyUser(
diff --git a/kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/DiscoveryClient.scala b/kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/DiscoveryClient.scala
index 6a114584d8e..51b57b20459 100644
--- a/kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/DiscoveryClient.scala
+++ b/kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/DiscoveryClient.scala
@@ -19,6 +19,7 @@ package org.apache.kyuubi.ha.client
 
 import org.apache.kyuubi.Logging
 import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.util.IPStackUtils
 
 /**
  * A collection of apis that discovery client need implement.
@@ -201,8 +202,8 @@ object DiscoveryClient {
         maybeInfos("hive.server2.thrift.bind.host"),
         maybeInfos("hive.server2.thrift.port").toInt)
     } else {
-      val strings = instance.split(":")
-      (strings(0), strings(1).toInt)
+      val hostPort = IPStackUtils.getHostAndPort(instance)
+      (hostPort.getHostname, hostPort.getPort)
     }
   }
 }
diff --git a/kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/ServiceNodeInfo.scala b/kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/ServiceNodeInfo.scala
index caca95d72bb..00104b8679a 100644
--- a/kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/ServiceNodeInfo.scala
+++ b/kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/ServiceNodeInfo.scala
@@ -17,6 +17,8 @@
 
 package org.apache.kyuubi.ha.client
 
+import org.apache.kyuubi.util.IPStackUtils
+
 case class ServiceNodeInfo(
     namespace: String,
     nodeName: String,
@@ -25,5 +27,5 @@ case class ServiceNodeInfo(
     version: Option[String],
     engineRefId: Option[String],
     attributes: Map[String, String] = Map.empty) {
-  def instance: String = s"$host:$port"
+  def instance: String = IPStackUtils.concatHostPort(host, port)
 }
diff --git a/kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/zookeeper/ZookeeperDiscoveryClient.scala b/kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/zookeeper/ZookeeperDiscoveryClient.scala
index bead620536a..a559258eecb 100644
--- a/kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/zookeeper/ZookeeperDiscoveryClient.scala
+++ b/kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/zookeeper/ZookeeperDiscoveryClient.scala
@@ -44,6 +44,7 @@ import org.apache.kyuubi.shaded.curator.utils.ZKPaths
 import org.apache.kyuubi.shaded.zookeeper.{CreateMode, KeeperException, WatchedEvent, Watcher}
 import org.apache.kyuubi.shaded.zookeeper.CreateMode.PERSISTENT
 import org.apache.kyuubi.shaded.zookeeper.KeeperException.NodeExistsException
+import org.apache.kyuubi.util.IPStackUtils
 import org.apache.kyuubi.util.ThreadUtils
 
 class ZookeeperDiscoveryClient(conf: KyuubiConf) extends DiscoveryClient {
@@ -316,15 +317,15 @@ class ZookeeperDiscoveryClient(conf: KyuubiConf) extends DiscoveryClient {
     if (!instance.contains(":")) {
       return instance
     }
-    val hostPort = instance.split(":", 2)
+    val hostPort = IPStackUtils.getHostAndPort(instance)
     val confsToPublish = collection.mutable.Map[String, String]()
 
     // Hostname
-    confsToPublish += ("hive.server2.thrift.bind.host" -> hostPort(0))
+    confsToPublish += ("hive.server2.thrift.bind.host" -> hostPort.getHostname)
     // Transport mode
     confsToPublish += ("hive.server2.transport.mode" -> "binary")
     // Transport specific confs
-    confsToPublish += ("hive.server2.thrift.port" -> hostPort(1))
+    confsToPublish += ("hive.server2.thrift.port" -> hostPort.getPort.toString)
     confsToPublish += ("hive.server2.thrift.sasl.qop" -> conf.get(KyuubiConf.SASL_QOP))
     // Auth specific confs
     val authenticationMethod = conf.get(KyuubiConf.AUTHENTICATION_METHOD).mkString(",")
diff --git a/kyuubi-ha/src/test/scala/org/apache/kyuubi/ha/client/DiscoveryClientSuite.scala b/kyuubi-ha/src/test/scala/org/apache/kyuubi/ha/client/DiscoveryClientSuite.scala
index 5a1af35f6e3..4863096ef6a 100644
--- a/kyuubi-ha/src/test/scala/org/apache/kyuubi/ha/client/DiscoveryClientSuite.scala
+++ b/kyuubi-ha/src/test/scala/org/apache/kyuubi/ha/client/DiscoveryClientSuite.scala
@@ -19,7 +19,7 @@ package org.apache.kyuubi.ha.client
 
 import org.apache.kyuubi.KyuubiFunSuite
 
-trait DiscoveryClientSuite extends KyuubiFunSuite {
+class DiscoveryClientSuite extends KyuubiFunSuite {
 
   test("parse host and port from instance string") {
     val host = "127.0.0.1"
@@ -36,5 +36,16 @@ trait DiscoveryClientSuite extends KyuubiFunSuite {
     val (host2, port2) = DiscoveryClient.parseInstanceHostPort(instance2)
     assert(host === host2)
     assert(port === port2)
+
+    // IPv6 address with square brackets
+    val ipv6Host = "fc00:172::1"
+    val (host3, port3) = DiscoveryClient.parseInstanceHostPort(s"[$ipv6Host]:$port")
+    assert(ipv6Host === host3)
+    assert(port === port3)
+
+    // IPv6 address without square brackets
+    val (host4, port4) = DiscoveryClient.parseInstanceHostPort(s"$ipv6Host:$port")
+    assert(ipv6Host === host4)
+    assert(port === port4)
   }
 }
diff --git a/kyuubi-hive-beeline/src/main/java/org/apache/hive/beeline/hs2connection/KyuubiConfFileParser.java b/kyuubi-hive-beeline/src/main/java/org/apache/hive/beeline/hs2connection/KyuubiConfFileParser.java
index dd25aeafca8..e1d779f517c 100644
--- a/kyuubi-hive-beeline/src/main/java/org/apache/hive/beeline/hs2connection/KyuubiConfFileParser.java
+++ b/kyuubi-hive-beeline/src/main/java/org/apache/hive/beeline/hs2connection/KyuubiConfFileParser.java
@@ -26,6 +26,7 @@
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.*;
+import org.apache.kyuubi.util.IPStackUtils;
 import org.apache.kyuubi.util.JavaUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -150,7 +151,7 @@ private void addHosts(Properties props) throws KyuubiConfFileParseException {
     }
 
     int portNum = getPortNum(thriftMode);
-    props.setProperty("hosts", host + ":" + portNum);
+    props.setProperty("hosts", IPStackUtils.concatHostPort(host, portNum));
   }
 
   private int getPortNum(THRIFT_MODE thriftMode) {
diff --git a/kyuubi-hive-jdbc/src/main/java/org/apache/kyuubi/jdbc/hive/KyuubiConnection.java b/kyuubi-hive-jdbc/src/main/java/org/apache/kyuubi/jdbc/hive/KyuubiConnection.java
index cfe3fdfa0cd..15b9b35ae49 100644
--- a/kyuubi-hive-jdbc/src/main/java/org/apache/kyuubi/jdbc/hive/KyuubiConnection.java
+++ b/kyuubi-hive-jdbc/src/main/java/org/apache/kyuubi/jdbc/hive/KyuubiConnection.java
@@ -72,6 +72,7 @@
 import org.apache.kyuubi.shaded.thrift.transport.THttpClient;
 import org.apache.kyuubi.shaded.thrift.transport.TTransport;
 import org.apache.kyuubi.shaded.thrift.transport.TTransportException;
+import org.apache.kyuubi.util.IPStackUtils;
 import org.apache.kyuubi.util.SubjectUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -428,7 +429,7 @@ private String getServerHttpUrl(boolean useSsl) {
     } else if (!httpPath.startsWith("/")) {
       httpPath = "/" + httpPath;
     }
-    return schemeName + "://" + host + ":" + port + httpPath;
+    return schemeName + "://" + IPStackUtils.concatHostPort(host, port) + httpPath;
   }
 
   private TTransport createHttpTransport() throws SQLException, TTransportException {
diff --git a/kyuubi-hive-jdbc/src/main/java/org/apache/kyuubi/jdbc/hive/Utils.java b/kyuubi-hive-jdbc/src/main/java/org/apache/kyuubi/jdbc/hive/Utils.java
index 2567795d954..01009606b82 100644
--- a/kyuubi-hive-jdbc/src/main/java/org/apache/kyuubi/jdbc/hive/Utils.java
+++ b/kyuubi-hive-jdbc/src/main/java/org/apache/kyuubi/jdbc/hive/Utils.java
@@ -30,6 +30,7 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.kyuubi.shaded.hive.service.rpc.thrift.TStatus;
 import org.apache.kyuubi.shaded.hive.service.rpc.thrift.TStatusCode;
+import org.apache.kyuubi.util.IPStackUtils;
 import org.apache.kyuubi.util.reflect.DynConstructors;
 import org.apache.kyuubi.util.reflect.DynMethods;
 import org.slf4j.Logger;
@@ -406,7 +407,7 @@ public static JdbcConnectionParams extractURLComponents(String uri, Properties i
         connParams.setPort(port);
       }
       // We check for invalid host, port while configuring connParams with configureConnParams()
-      authorityStr = connParams.getHost() + ":" + connParams.getPort();
+      authorityStr = IPStackUtils.concatHostPort(connParams.getHost(), connParams.getPort());
       LOG.debug("Resolved authority: " + authorityStr);
       uri = uri.replace(dummyAuthorityString, authorityStr);
     }
@@ -419,7 +420,7 @@ public static JdbcConnectionParams extractURLComponents(String uri, Properties i
   static void configureConnParamsFromZooKeeper(JdbcConnectionParams connParams)
       throws ZooKeeperHiveClientException, JdbcUriParseException {
     ZooKeeperHiveClientHelper.configureConnParams(connParams);
-    String authorityStr = connParams.getHost() + ":" + connParams.getPort();
+    String authorityStr = IPStackUtils.concatHostPort(connParams.getHost(), connParams.getPort());
     LOG.debug("Resolved authority: " + authorityStr);
     String jdbcUriString = connParams.getJdbcUriString();
     // Replace ZooKeeper ensemble from the authority component of the JDBC Uri provided by the
@@ -532,8 +533,8 @@ static boolean updateConnParamsFromZooKeeper(JdbcConnectionParams connParams) {
           connParams
               .getJdbcUriString()
               .replace(
-                  oldServerHost + ":" + oldServerPort,
-                  connParams.getHost() + ":" + connParams.getPort()));
+                  IPStackUtils.concatHostPort(oldServerHost, oldServerPort),
+                  IPStackUtils.concatHostPort(connParams.getHost(), connParams.getPort())));
       LOG.info("Selected HiveServer2 instance with uri: " + connParams.getJdbcUriString());
     } catch (ZooKeeperHiveClientException e) {
       LOG.error(e.getMessage());
diff --git a/kyuubi-hive-jdbc/src/main/java/org/apache/kyuubi/jdbc/hive/ZooKeeperHiveClientHelper.java b/kyuubi-hive-jdbc/src/main/java/org/apache/kyuubi/jdbc/hive/ZooKeeperHiveClientHelper.java
index f94bdb431e6..d11f2126b94 100644
--- a/kyuubi-hive-jdbc/src/main/java/org/apache/kyuubi/jdbc/hive/ZooKeeperHiveClientHelper.java
+++ b/kyuubi-hive-jdbc/src/main/java/org/apache/kyuubi/jdbc/hive/ZooKeeperHiveClientHelper.java
@@ -30,6 +30,7 @@
 import org.apache.kyuubi.shaded.curator.framework.CuratorFramework;
 import org.apache.kyuubi.shaded.curator.framework.CuratorFrameworkFactory;
 import org.apache.kyuubi.shaded.curator.retry.ExponentialBackoffRetry;
+import org.apache.kyuubi.util.IPStackUtils;
 
 class ZooKeeperHiveClientHelper {
   // Pattern for key1=value1;key2=value2
@@ -96,13 +97,14 @@ private static void updateParamsWithZKServerNode(
     // it must be the server uri added by an older version HS2
     Matcher matcher = kvPattern.matcher(dataStr);
     if (!matcher.find()) {
-      String[] split = dataStr.split(":");
-      if (split.length != 2) {
+      try {
+        IPStackUtils.HostPort hostPort = IPStackUtils.getHostAndPort(dataStr);
+        connParams.setHost(hostPort.getHostname());
+        connParams.setPort(hostPort.getPort());
+      } catch (IllegalArgumentException e) {
         throw new ZooKeeperHiveClientException(
-            "Unable to read HiveServer2 uri from ZooKeeper: " + dataStr);
+            "Unable to parse HiveServer2 uri from ZooKeeper: " + dataStr, e);
       }
-      connParams.setHost(split[0]);
-      connParams.setPort(Integer.parseInt(split[1]));
     } else {
       applyConfs(dataStr, connParams);
     }
diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/EngineRef.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/EngineRef.scala
index 621ecb6d62e..1628e8405e9 100644
--- a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/EngineRef.scala
+++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/EngineRef.scala
@@ -47,7 +47,7 @@ import org.apache.kyuubi.metrics.MetricsSystem
 import org.apache.kyuubi.operation.log.OperationLog
 import org.apache.kyuubi.plugin.GroupProvider
 import org.apache.kyuubi.service.authentication.{AuthTypes, AuthUtils}
-import org.apache.kyuubi.util.JavaUtils
+import org.apache.kyuubi.util.{IPStackUtils, JavaUtils}
 
 /**
  * The description and functionality of an engine at server side
@@ -464,7 +464,7 @@ private[kyuubi] object EngineRef {
       val host = conf.get(FRONTEND_ADVERTISED_HOST)
         .orElse(conf.get(FRONTEND_THRIFT_BINARY_BIND_HOST))
         .getOrElse(JavaUtils.findLocalInetAddress.getHostAddress)
-      s"jdbc:kyuubi://$host:$port/default"
+      s"jdbc:kyuubi://${IPStackUtils.concatHostPort(host, port)}/default"
     }
   }
 }
diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiRestFrontendService.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiRestFrontendService.scala
index e71ac256c6b..eb3dcdca734 100644
--- a/kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiRestFrontendService.scala
+++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiRestFrontendService.scala
@@ -41,7 +41,7 @@ import org.apache.kyuubi.server.ui.{JettyServer, JettyUtils}
 import org.apache.kyuubi.service.{AbstractFrontendService, Serverable, Service, ServiceUtils}
 import org.apache.kyuubi.service.authentication.{AuthTypes, AuthUtils}
 import org.apache.kyuubi.session.{KyuubiBatchSession, KyuubiSessionManager, SessionHandle}
-import org.apache.kyuubi.util.{JavaUtils, ThreadUtils}
+import org.apache.kyuubi.util.{IPStackUtils, JavaUtils, ThreadUtils}
 import org.apache.kyuubi.util.ThreadUtils.scheduleTolerableRunnableWithFixedDelay
 
 /**
@@ -110,7 +110,7 @@ class KyuubiRestFrontendService(override val serverable: Serverable)
   override def connectionUrl: String = {
     checkInitialized()
     conf.get(FRONTEND_ADVERTISED_HOST) match {
-      case Some(advertisedHost) => s"$advertisedHost:$port"
+      case Some(advertisedHost) => IPStackUtils.concatHostPort(advertisedHost, port)
       case None => server.getServerUri
     }
   }
diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiTrinoFrontendService.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiTrinoFrontendService.scala
index 021b85e0bd0..ca277eb94e5 100644
--- a/kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiTrinoFrontendService.scala
+++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiTrinoFrontendService.scala
@@ -25,7 +25,7 @@ import org.apache.kyuubi.config.KyuubiConf._
 import org.apache.kyuubi.server.trino.api.v1.ApiRootResource
 import org.apache.kyuubi.server.ui.JettyServer
 import org.apache.kyuubi.service.{AbstractFrontendService, Serverable, Service}
-import org.apache.kyuubi.util.JavaUtils
+import org.apache.kyuubi.util.{IPStackUtils, JavaUtils}
 
 /**
  * A frontend service based on RESTful api via HTTP protocol.
@@ -64,7 +64,7 @@ class KyuubiTrinoFrontendService(override val serverable: Serverable)
   override def connectionUrl: String = {
     checkInitialized()
     conf.get(FRONTEND_ADVERTISED_HOST) match {
-      case Some(advertisedHost) => s"$advertisedHost:$port"
+      case Some(advertisedHost) => IPStackUtils.concatHostPort(advertisedHost, port)
       case None => server.getServerUri
     }
   }
diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/server/ui/JettyServer.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/server/ui/JettyServer.scala
index c1bac1275cc..3146bfb4b58 100644
--- a/kyuubi-server/src/main/scala/org/apache/kyuubi/server/ui/JettyServer.scala
+++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/server/ui/JettyServer.scala
@@ -23,6 +23,7 @@ import org.eclipse.jetty.util.component.LifeCycle
 import org.eclipse.jetty.util.thread.{QueuedThreadPool, ScheduledExecutorScheduler}
 
 import org.apache.kyuubi.Logging
+import org.apache.kyuubi.util.IPStackUtils
 import org.apache.kyuubi.util.JavaUtils
 
 private[kyuubi] class JettyServer(
@@ -37,7 +38,7 @@ private[kyuubi] class JettyServer(
       server.addConnector(connector)
       val localPort = connector.getLocalPort
       require(localPort > 0, "Jetty server port should be positive, but got " + localPort)
-      _serverUri = connector.getHost + ":" + localPort
+      _serverUri = IPStackUtils.concatHostPort(connector.getHost, localPort)
     } catch {
       case e: Exception =>
         stop()
@@ -56,7 +57,7 @@ private[kyuubi] class JettyServer(
 
   @volatile private var _serverUri: String = _
   def getServerUri: String = Option(_serverUri).getOrElse {
-    val uri = connector.getHost + ":" + connector.getLocalPort
+    val uri = IPStackUtils.concatHostPort(connector.getHost, connector.getLocalPort)
     warn("Jetty server is not started yet, returning " + uri)
     uri
   }
diff --git a/kyuubi-util/src/main/java/org/apache/kyuubi/util/IPStackUtils.java b/kyuubi-util/src/main/java/org/apache/kyuubi/util/IPStackUtils.java
new file mode 100644
index 00000000000..27bd6848baf
--- /dev/null
+++ b/kyuubi-util/src/main/java/org/apache/kyuubi/util/IPStackUtils.java
@@ -0,0 +1,157 @@
+/*
+ * 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.util;
+
+/**
+ * Utility methods for handling IPv4/IPv6 host:port strings. The logic mirrors Hive's {@code
+ * org.apache.hadoop.hive.common.IPStackUtils}, introduced in HIVE-28782 (Hive 4.1.0).
+ */
+public final class IPStackUtils {
+
+  private IPStackUtils() {}
+
+  /**
+   * Splits a given input string representing a Hostname or an IP address and port into an
+   * `HostPort` object. The input string must be in the format of IPv4/IPv6/[IPv6]/hostname:port.
+   *
+   * @param input The input string containing the Hostname/IP address and port, in the format
+   *     "IPv4:port", "[IPv6]:port", "IPv6:port", or "hostname:port".
+   * @return A {@link HostPort} object containing the parsed IP address and port number.
+   * @throws IllegalArgumentException If the input format is invalid, if the host is null or empty,
+   *     or if the port number is invalid.
+   */
+  public static HostPort getHostAndPort(String input) {
+    String host;
+    int port;
+
+    if (isEmpty(input)) {
+      throw new IllegalArgumentException("Input string is null or empty");
+    }
+
+    // Check if the input contains a colon, which separates the host and port
+    int colonIndex = input.lastIndexOf(':');
+    if (colonIndex == -1) {
+      throw new IllegalArgumentException("Input does not contain a port.");
+    }
+
+    // Extract the host and port parts
+    host = input.substring(0, colonIndex);
+    port = getPort(input.substring(colonIndex + 1));
+
+    // Check if the host is not null or empty
+    validateHostNotEmpty(host);
+
+    // Handle IPv6 addresses enclosed in square brackets (e.g., [IPv6]:port)
+    if (host.startsWith("[") && host.endsWith("]")) {
+      host = host.substring(1, host.length() - 1); // Remove the square brackets
+    }
+
+    return new HostPort(host, port);
+  }
+
+  /**
+   * Returns an integer representation of the port number. Also validates whether the given string
+   * represents a valid port number. A valid port number is an integer between 0 and 65535
+   * inclusive.
+   *
+   * @param portString The string representing the port number.
+   * @return {@code int} the port number.
+   * @throws IllegalArgumentException if the port string is null/empty or the port number is out of
+   *     the valid range (0-65535).
+   * @throws NumberFormatException if the port string is not a valid integer.
+   */
+  public static int getPort(String portString) {
+    if (isEmpty(portString)) {
+      throw new IllegalArgumentException("port is null or empty");
+    }
+
+    int port = Integer.parseInt(portString);
+    validatePort(port);
+    return port;
+  }
+
+  private static void validateHostNotEmpty(String host) {
+    if (isEmpty(host) || host.equals("[]")) {
+      throw new IllegalArgumentException("Host address is null or empty.");
+    }
+  }
+
+  private static void validatePort(int port) {
+    if (port < 0 || port > 65535) {
+      throw new IllegalArgumentException("Port number out of range (0-65535).");
+    }
+  }
+
+  private static boolean isEmpty(String s) {
+    return s == null || s.isEmpty();
+  }
+
+  /**
+   * Concatenates the host and port with a colon. If the host is an IPv6 address, it is enclosed in
+   * square brackets.
+   *
+   * @param host the host
+   * @param port the port
+   * @return the concatenated host and port
+   * @throws IllegalArgumentException if the host is null/empty or the port number is out of the
+   *     valid range (0-65535).
+   */
+  public static String concatHostPort(String host, int port) {
+    validateHostNotEmpty(host);
+    validatePort(port);
+    return formatIPAddressForURL(host) + ":" + port;
+  }
+
+  /**
+   * Prepares an IP address for use in a URL.
+   *
+   * <p>This method ensures that IPv6 addresses are enclosed in square brackets, as required by URL
+   * syntax. IPv4 addresses and hostnames remain unchanged.
+   *
+   * @param ipAddress the IP address or hostname to format
+   * @return the formatted IP address for use in a URL
+   */
+  public static String formatIPAddressForURL(String ipAddress) {
+    if (ipAddress.contains(":") && !ipAddress.startsWith("[") && !ipAddress.endsWith("]")) {
+      // IPv6 address
+      return "[" + ipAddress + "]";
+    } else {
+      // IPv4 address or hostname
+      return ipAddress;
+    }
+  }
+
+  public static class HostPort {
+
+    private final String hostname;
+    private final int port;
+
+    public HostPort(String hostname, int port) {
+      this.hostname = hostname;
+      this.port = port;
+    }
+
+    public String getHostname() {
+      return hostname;
+    }
+
+    public int getPort() {
+      return port;
+    }
+  }
+}
diff --git a/kyuubi-util/src/test/java/org/apache/kyuubi/util/IPStackUtilsTest.java b/kyuubi-util/src/test/java/org/apache/kyuubi/util/IPStackUtilsTest.java
new file mode 100644
index 00000000000..a69e77fde6e
--- /dev/null
+++ b/kyuubi-util/src/test/java/org/apache/kyuubi/util/IPStackUtilsTest.java
@@ -0,0 +1,136 @@
+/*
+ * 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.util;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import org.apache.kyuubi.util.IPStackUtils.HostPort;
+import org.junit.jupiter.api.Test;
+
+public class IPStackUtilsTest {
+
+  // Test cases for concatHostPort method
+
+  @Test
+  public void testConcatHostPort() {
+    assertEquals("192.168.1.1:8080", IPStackUtils.concatHostPort("192.168.1.1", 8080));
+    assertEquals("[2001:db8::1]:8080", IPStackUtils.concatHostPort("2001:db8::1", 8080));
+    assertEquals("[::1]:9090", IPStackUtils.concatHostPort("::1", 9090));
+    assertEquals("example.com:443", IPStackUtils.concatHostPort("example.com", 443));
+  }
+
+  // Test cases for getHostAndPort method
+
+  @Test
+  public void testGetHostAndPortWithIPv4() {
+    HostPort result = IPStackUtils.getHostAndPort("192.168.1.1:8080");
+    assertEquals("192.168.1.1", result.getHostname());
+    assertEquals(8080, result.getPort());
+  }
+
+  @Test
+  public void testGetHostAndPortWithValidIPv6WithSquaredBrackets() {
+    HostPort result = IPStackUtils.getHostAndPort("[2001:0db8::1]:8080");
+    assertEquals("2001:0db8::1", result.getHostname());
+    assertEquals(8080, result.getPort());
+  }
+
+  @Test
+  public void testGetHostAndPortWithValidIPv6WithoutSquaredBrackets() {
+    HostPort result = IPStackUtils.getHostAndPort("2001:0db8::1:8080");
+    assertEquals("2001:0db8::1", result.getHostname());
+    assertEquals(8080, result.getPort());
+  }
+
+  @Test
+  public void testGetHostAndPortWithHostname() {
+    HostPort result = IPStackUtils.getHostAndPort("example.com:80");
+    assertEquals("example.com", result.getHostname());
+    assertEquals(80, result.getPort());
+  }
+
+  @Test
+  public void testGetHostPortWithInvalidAndPort() {
+    // Test case: port number out of range
+    IllegalArgumentException e1 =
+        assertThrows(
+            IllegalArgumentException.class, () -> IPStackUtils.getHostAndPort("192.168.1.1:70000"));
+    assertEquals("Port number out of range (0-65535).", e1.getMessage());
+
+    // Test case: input missing port
+    IllegalArgumentException e2 =
+        assertThrows(
+            IllegalArgumentException.class, () -> IPStackUtils.getHostAndPort("192.168.1.1"));
+    assertEquals("Input does not contain a port.", e2.getMessage());
+
+    // Test case: missing host
+    IllegalArgumentException e3 =
+        assertThrows(IllegalArgumentException.class, () -> IPStackUtils.getHostAndPort(":8080"));
+    assertEquals("Host address is null or empty.", e3.getMessage());
+  }
+
+  // Test cases for getPort method
+
+  @Test
+  public void testGetPort() {
+    assertEquals(8080, IPStackUtils.getPort("8080"));
+    assertEquals(65535, IPStackUtils.getPort("65535"));
+    assertEquals(0, IPStackUtils.getPort("0"));
+  }
+
+  @Test
+  public void testGetPortWithInvalidPort() {
+    // Test case: port number too high
+    IllegalArgumentException e1 =
+        assertThrows(IllegalArgumentException.class, () -> IPStackUtils.getPort("70000"));
+    assertEquals("Port number out of range (0-65535).", e1.getMessage());
+
+    // Test case: negative port number
+    IllegalArgumentException e2 =
+        assertThrows(IllegalArgumentException.class, () -> IPStackUtils.getPort("-1"));
+    assertEquals("Port number out of range (0-65535).", e2.getMessage());
+
+    // Test case: non-numeric port
+    IllegalArgumentException e3 =
+        assertThrows(IllegalArgumentException.class, () -> IPStackUtils.getPort("abc"));
+    assertEquals("For input string: \"abc\"", e3.getMessage());
+  }
+
+  // Test cases for formatIPAddressForURL method
+
+  @Test
+  public void testFormatIPAddressForURLWithIPv4() {
+    assertEquals("192.168.1.1", IPStackUtils.formatIPAddressForURL("192.168.1.1"));
+  }
+
+  @Test
+  public void testFormatIPAddressForURLWithIPv6() {
+    assertEquals("[2001:0db8::1]", IPStackUtils.formatIPAddressForURL("2001:0db8::1"));
+  }
+
+  @Test
+  public void testFormatIPAddressForURLWithHostname() {
+    assertEquals("example.com", IPStackUtils.formatIPAddressForURL("example.com"));
+  }
+
+  @Test
+  public void testFormatIPAddressForURLWithAlreadyBracketedIPv6() {
+    assertEquals("[2001:0db8::1]", IPStackUtils.formatIPAddressForURL("[2001:0db8::1]"));
+  }
+}