Fix empty list engine result when etcd is used as the service registry

Author: zhaohehuhu

kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/DiscoveryClient.scala

@@ -66,6 +66,7 @@ trait DiscoveryClient extends Logging {
    */
   def pathNonExists(path: String): Boolean
 
+  def pathNonExists(path: String, isPrefix: Boolean): Boolean
   /**
    * Delete a path.
    * @param path the path to be deleted

kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/etcd/EtcdDiscoveryClient.scala

@@ -151,6 +151,12 @@ class EtcdDiscoveryClient(conf: KyuubiConf) extends DiscoveryClient {
     kvClient.get(ByteSequence.from(path.getBytes())).get().getKvs.isEmpty
   }
 
+  override def pathNonExists(path: String, isPrefix: Boolean): Boolean = {
+    kvClient.get(
+      ByteSequence.from(path.getBytes()),
+      GetOption.newBuilder().isPrefix(isPrefix).build()).get().getKvs.isEmpty
+  }
+
   override def delete(path: String, deleteChildren: Boolean = false): Unit = {
     kvClient.delete(
       ByteSequence.from(path.getBytes()),

kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/zookeeper/ZookeeperDiscoveryClient.scala

@@ -90,6 +90,10 @@ class ZookeeperDiscoveryClient(conf: KyuubiConf) extends DiscoveryClient {
     zkClient.checkExists().forPath(path) == null
   }
 
+  override def pathNonExists(path: String, isPrefix: Boolean): Boolean = {
+    zkClient.checkExists().forPath(path) == null
+  }
+
   override def delete(path: String, deleteChildren: Boolean = false): Unit = {
     if (deleteChildren) {
       zkClient.delete().deletingChildrenIfNeeded().forPath(path)

kyuubi-ha/src/test/scala/org/apache/kyuubi/ha/client/etcd/EtcdDiscoveryClientSuite.scala

@@ -92,4 +92,22 @@ class EtcdDiscoveryClientSuite extends DiscoveryClientTests {
       assert(!discoveryClient.pathExists(path))
     }
   }
+
+  test("etcd test: set, get with path prefix and delete") {
+    withDiscoveryClient(conf) { discoveryClient =>
+      val path = "/kyuubi_version_USER_SPARK_SQL/test/default"
+      val pathPrefix = "/kyuubi_version_USER_SPARK_SQL/test"
+      // set
+      discoveryClient.create(path, "PERSISTENT")
+      assert(discoveryClient.pathExists(path))
+      assert(!discoveryClient.pathNonExists(pathPrefix, isPrefix = true))
+
+      // get
+      assert(new String(discoveryClient.getData(path), StandardCharsets.UTF_8) == path)
+
+      // delete
+      discoveryClient.delete(path)
+      assert(!discoveryClient.pathExists(path))
+    }
+  }
 }

kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/AdminResource.scala

@@ -349,7 +349,7 @@ private[v1] class AdminResource extends ApiRequestContext with Logging {
         case Some(_) =>
           info(s"Listing engine nodes under $engineSpace")
           engineNodes ++= discoveryClient.getServiceNodesInfo(engineSpace)
-        case None if discoveryClient.pathNonExists(engineSpace) =>
+        case None if discoveryClient.pathNonExists(engineSpace, isPrefix = true) =>
           warn(s"Path $engineSpace does not exist. user: $userName, engine type: $engineType, " +
             s"share level: $shareLevel, subdomain: $subdomain")
         case None =>