[KYUUBI #3869] [K8S][IT][BUG] Fix the issue that connect conf is not used in the jdbc connection string

### _Why are the changes needed?_

Fix the issue that connect conf is not used in the Kyuubi On Kubernetes IT's jdbc connection string.

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

- [x] Wait for IT ci

Closes #3869 from zwangsheng/bugfix/kyuubi_on_k8s_it_connect_conf.

Closes #3869

3cecd5f4 [zwangsheng] fix
63025a28 [zwangsheng] fix
924949f4 [zwangsheng] fix
21e93298 [zwangsheng] fix
fc5794ef [zwangsheng] fix
6dca96cd [zwangsheng] test
c2c81bb4 [zwangsheng] test
b8bb820b [zwangsheng] add unit test
dad4c739 [zwangsheng] fix bind
406f1de5 [zwangsheng] proxy for 185
c7d6ee6d [zwangsheng] merge
2290a24f [Binjie Yang] Update master.yml
9ffcb498 [Binjie Yang] Update KyuubiOnKubernetesTestsSuite.scala
714b340d [Binjie Yang] Update KyuubiOnKubernetesTestsSuite.scala
fa7fc542 [Binjie Yang] Update KyuubiOnKubernetesTestsSuite.scala
af4b9881 [zwangsheng] set 777 for /
49f705eb [zwangsheng] set 777 for test
fc66843d [zwangsheng] stop ci
e2ba0bcf [zwangsheng] add test name
7db4eab1 [zwangsheng] fast test cluster
33d490d1 [zwangsheng] add unit test
e2e12f4e [zwangsheng] fast test cluster
e8251011 [zwangsheng] test
b66468f5 [zwangsheng] test
633d99e4 [zwangsheng] change host
40ba5740 [zwangsheng] test
e393f9a5 [zwangsheng] test
532cd7df [zwangsheng] merge
4597572e [zwangsheng] test
b8fc86a1 [Binjie Yang] Update KyuubiOnKubernetesTestsSuite.scala
34be2761 [zwangsheng] TEST
a3c60e45 [zwangsheng] Changes
19e3bc22 [zwangsheng] for fast test
3ad2337f [zwangsheng] try cluster
39df2c40 [zwangsheng] try cluster
ed8f8baa [zwangsheng] fix client
7f711acb [zwangsheng] fix
b034731e [zwangsheng] fix
d646f4ac [zwangsheng] fix
2b9591c4 [zwangsheng] debug
a977d907 [zwangsheng] fix
0c3486fa [zwangsheng] debug
f0a0304b [zwangsheng] Add serviceAccount
eb3424ab [zwangsheng] fix user
cac7e69d [zwangsheng] proxy user
2886520f [zwangsheng] debug
25a677c6 [zwangsheng] debug
9f201d89 [zwangsheng] debug
e533664d [zwangsheng] fix it test
d9bf9173 [zwangsheng] fix it test

Lead-authored-by: zwangsheng <2213335496@qq.com>
Co-authored-by: Binjie Yang <52876270+zwangsheng@users.noreply.github.com>
Signed-off-by: Cheng Pan <chengpan@apache.org>

.github/workflows/master.yml

@@ -355,17 +355,28 @@ jobs:
         run: |
           kubectl get serviceaccount
           kubectl create serviceaccount kyuubi
+          kubectl create clusterrolebinding kyuubi-role --clusterrole=edit --serviceaccount=default:kyuubi
           kubectl get serviceaccount
       - name: start kyuubi
         run: kubectl apply -f integration-tests/kyuubi-kubernetes-it/src/test/resources/kyuubi-server.yaml
       - name: kyuubi pod check
-        run: kubectl get pods
+        run: |
+          kubectl get pods
+          kubectl describe pods kyuubi-test
       - name: integration tests
         run: >-
           ./build/mvn ${MVN_OPT} clean install
           -pl integration-tests/kyuubi-kubernetes-it -am
           -Pkubernetes-it
           -Dtest=none -DwildcardSuites=org.apache.kyuubi.kubernetes.test.deployment,org.apache.kyuubi.kubernetes.test.KubernetesUtilsTest
+      - name: Cat kyuubi server log
+        if: failure()
+        run: kubectl logs kyuubi-test
+      - name: Cat spark driver log
+        if: failure()
+        run: |
+          kubectl get pods
+          kubectl get pods | grep driver | awk -F " " '{print$1}' | xargs -I {} kubectl logs {}
       - name: Upload test logs
         if: failure()
         uses: actions/upload-artifact@v3

integration-tests/kyuubi-kubernetes-it/src/test/scala/org/apache/kyuubi/kubernetes/test/WithKyuubiServerOnKubernetes.scala

@@ -17,6 +17,7 @@
 
 package org.apache.kyuubi.kubernetes.test
 
+import io.fabric8.kubernetes.api.model.Pod
 import io.fabric8.kubernetes.client.DefaultKubernetesClient
 
 import org.apache.kyuubi.KyuubiFunSuite
@@ -25,26 +26,26 @@ trait WithKyuubiServerOnKubernetes extends KyuubiFunSuite {
   protected def connectionConf: Map[String, String] = Map.empty
   private val miniKubernetesClient: DefaultKubernetesClient = MiniKube.getKubernetesClient
 
-  protected def getJdbcUrl: String = {
-    val kyuubiServers =
-      miniKubernetesClient.pods().list().getItems
-    assert(kyuubiServers.size() == 1)
-    val kyuubiServer = kyuubiServers.get(0)
+  lazy val kyuubiPod: Pod = miniKubernetesClient.pods().withName("kyuubi-test").get()
+  lazy val kyuubiServerIp: String = kyuubiPod.getStatus.getPodIP
+  lazy val miniKubeIp: String = MiniKube.getIp
+  lazy val miniKubeApiMaster: String = miniKubernetesClient.getMasterUrl.toString
+
+  protected def getJdbcUrl(connectionConf: Map[String, String]): String = {
     // Kyuubi server state should be running since mvn compile is quite slowly..
-    if (!"running".equalsIgnoreCase(kyuubiServer.getStatus.getPhase)) {
+    if (!"running".equalsIgnoreCase(kyuubiPod.getStatus.getPhase)) {
       val log =
         miniKubernetesClient
           .pods()
-          .withName(kyuubiServer.getMetadata.getName)
+          .withName(kyuubiPod.getMetadata.getName)
           .getLog
       throw new IllegalStateException(
-        s"Kyuubi server pod state error: ${kyuubiServer.getStatus.getPhase}, log:\n$log")
+        s"Kyuubi server pod state error: ${kyuubiPod.getStatus.getPhase}, log:\n$log")
     }
-    val kyuubiServerIp = MiniKube.getIp
+    val kyuubiServerIp = miniKubeIp
     val kyuubiServerPort =
-      kyuubiServer.getSpec.getContainers.get(0).getPorts.get(0).getHostPort
-    s"jdbc:hive2://$kyuubiServerIp:$kyuubiServerPort/;"
+      kyuubiPod.getSpec.getContainers.get(0).getPorts.get(0).getHostPort
+    val connectStr = connectionConf.map(kv => kv._1 + "=" + kv._2).mkString("#", ";", "")
+    s"jdbc:hive2://$kyuubiServerIp:$kyuubiServerPort/;$connectStr"
   }
-
-  def getMiniKubeApiMaster: String = miniKubernetesClient.getMasterUrl.toString
 }

integration-tests/kyuubi-kubernetes-it/src/test/scala/org/apache/kyuubi/kubernetes/test/deployment/KyuubiOnKubernetesTestsSuite.scala

@@ -18,6 +18,8 @@
 package org.apache.kyuubi.kubernetes.test.deployment
 
 import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{FileSystem, Path}
+import org.apache.hadoop.fs.permission.{FsAction, FsPermission}
 import org.apache.hadoop.net.NetUtils
 
 import org.apache.kyuubi.{Utils, WithSimpleDFSService}
@@ -42,23 +44,23 @@ class KyuubiOnKubernetesWithLocalSparkTestsSuite extends WithKyuubiServerOnKuber
     super.connectionConf ++ Map("spark.master" -> "local", "spark.executor.instances" -> "1")
   }
 
-  override protected def jdbcUrl: String = getJdbcUrl
+  override protected def jdbcUrl: String = getJdbcUrl(connectionConf)
+
+  override protected lazy val user: String = "local"
 }
 
-class KyuubiOnKubernetesWithSparkTestsBase extends WithKyuubiServerOnKubernetes
-  with SparkQueryTests {
+class KyuubiOnKubernetesWithSparkTestsBase extends WithKyuubiServerOnKubernetes {
   override protected def connectionConf: Map[String, String] = {
     super.connectionConf ++
       Map(
-        "spark.master" -> s"k8s://$getMiniKubeApiMaster",
+        "spark.master" -> s"k8s://$miniKubeApiMaster",
+        "spark.kubernetes.container.image" -> "apache/spark:3.3.1",
         "spark.executor.memory" -> "512M",
-        "spark.driver.memory" -> "512M",
+        "spark.driver.memory" -> "1024M",
         "spark.kubernetes.driver.request.cores" -> "250m",
         "spark.kubernetes.executor.request.cores" -> "250m",
         "spark.executor.instances" -> "1")
   }
-
-  override protected def jdbcUrl: String = getJdbcUrl
 }
 
 /**
@@ -72,10 +74,17 @@ class KyuubiOnKubernetesWithSparkTestsBase extends WithKyuubiServerOnKubernetes
  *  ------------       -------------------------------------------------      ---------------------
  */
 class KyuubiOnKubernetesWithClientSparkTestsSuite
-  extends KyuubiOnKubernetesWithSparkTestsBase {
+  extends KyuubiOnKubernetesWithSparkTestsBase with SparkQueryTests {
   override protected def connectionConf: Map[String, String] = {
-    super.connectionConf ++ Map("spark.submit.deployMode" -> "client")
+    super.connectionConf ++ Map(
+      "spark.submit.deployMode" -> "client",
+      "spark.driver.host" -> kyuubiServerIp,
+      "kyuubi.frontend.connection.url.use.hostname" -> "false")
   }
+
+  override protected def jdbcUrl: String = getJdbcUrl(connectionConf)
+
+  override protected lazy val user: String = "client"
 }
 
 /**
@@ -89,11 +98,22 @@ class KyuubiOnKubernetesWithClientSparkTestsSuite
  *  ----------       -----------------     -----------------------------      ---------------------
  */
 class KyuubiOnKubernetesWithClusterSparkTestsSuite
-  extends KyuubiOnKubernetesWithSparkTestsBase with WithSimpleDFSService {
+  extends KyuubiOnKubernetesWithSparkTestsBase with WithSimpleDFSService with SparkQueryTests {
   private val localhostAddress = Utils.findLocalInetAddress.getHostAddress
   private val driverTemplate =
     Thread.currentThread().getContextClassLoader.getResource("driver.yml")
 
+  override def beforeAll(): Unit = {
+    super.beforeAll()
+    val fs = FileSystem.get(getHadoopConf)
+    fs.mkdirs(
+      new Path("/spark"),
+      new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL))
+    fs.setPermission(new Path("/"), new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL))
+    fs.setPermission(new Path("/spark"), new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL))
+    fs.copyFromLocalFile(new Path(driverTemplate.getPath), new Path("/spark/driver.yml"))
+  }
+
   override val hadoopConf: Configuration = {
     val hdfsConf: Configuration = new Configuration()
     hdfsConf.set("dfs.namenode.rpc-bind-host", "0.0.0.0")
@@ -103,6 +123,8 @@ class KyuubiOnKubernetesWithClusterSparkTestsSuite
     // spark use 185 as userid in docker
     hdfsConf.set("hadoop.proxyuser.185.groups", "*")
     hdfsConf.set("hadoop.proxyuser.185.hosts", "*")
+    hdfsConf.set("hadoop.proxyuser.kyuubi.groups", "*")
+    hdfsConf.set("hadoop.proxyuser.kyuubi.hosts", "*")
     hdfsConf
   }
 
@@ -112,9 +134,14 @@ class KyuubiOnKubernetesWithClusterSparkTestsSuite
         "spark.submit.deployMode" -> "cluster",
         "spark.kubernetes.file.upload.path" -> s"hdfs://$localhostAddress:$getDFSPort/spark",
         "spark.hadoop.dfs.client.use.datanode.hostname" -> "true",
-        "spark.kubernetes.authenticate.driver.serviceAccountName" -> "spark",
-        "spark.kubernetes.driver.podTemplateFile" -> driverTemplate.getPath,
-        ZK_CLIENT_PORT_ADDRESS.key -> localhostAddress,
-        FRONTEND_THRIFT_BINARY_BIND_HOST.key -> localhostAddress)
+        "spark.kubernetes.authenticate.driver.serviceAccountName" -> "kyuubi",
+        "spark.kubernetes.driver.podTemplateFile" ->
+          s"hdfs://$localhostAddress:$getDFSPort/spark/driver.yml",
+        "spark.kyuubi.frontend.thrift.binary.bind.host" -> miniKubeIp,
+        "spark.kyuubi.ha.addresses" -> s"$kyuubiServerIp:2181",
+        ZK_CLIENT_PORT_ADDRESS.key -> kyuubiServerIp,
+        FRONTEND_THRIFT_BINARY_BIND_HOST.key -> kyuubiServerIp)
   }
+
+  override protected def jdbcUrl: String = getJdbcUrl(connectionConf)
 }