[KYUUBI #7171] Fix empty list engine result when etcd is used as the service registry

### Why are the changes needed?

In etcd, keys are stored in a flat namespace where / is just part of the key name. For example, /kyuubi_version_USER_SPARK_SQL/test/default is treated as a complete single key.

To retrieve all entries under a similar path, a prefix query (--prefix) is required.

In ZooKeeper, data is organized in a hierarchical tree structure where / represents parent-child relationships.

### How was this patch tested?

added a UT to verify code.
### Was this patch authored or co-authored using generative AI tooling?

No

Closes #7171 from zhaohehuhu/dev-0813.

Closes #7171

48de708 [zhaohehuhu] fix
8f83a46 [zhaohehuhu] refactor
2d9b322 [zhaohehuhu] refactor
85207c4 [zhaohehuhu] refactor
3961489 [zhaohehuhu] reformat
15cc07b [zhaohehuhu] Fix empty list engine result when etcd is used as the service registry

Lead-authored-by: zhaohehuhu <[email protected]>
Co-authored-by: zhaohehuhu <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>

Author: zhaohehuhu

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

@@ -62,9 +62,14 @@ trait DiscoveryClient extends Logging {
   def pathExists(path: String): Boolean
 
   /**
-   * Check if the path non exists.
-   */
-  def pathNonExists(path: String): Boolean
+   * Checks whether the given path and all its child paths (prefix matches) do not exist.
+   * The isPrefix is set to true by default for Etcd to retrieve all entries under the given path.
+   * For other discovery service, it can be set false by default to only check the exact path.
+   * @param path The path to check
+   * @param isPrefix whether to check all paths with the given path as prefix
+   * @return true if the path and all its sub-paths are non-existent; false otherwise
+   */
+  def pathNonExists(path: String, isPrefix: Boolean = false): Boolean
 
   /**
    * Delete a path.

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

@@ -147,8 +147,10 @@ class EtcdDiscoveryClient(conf: KyuubiConf) extends DiscoveryClient {
     !pathNonExists(path)
   }
 
-  override def pathNonExists(path: String): Boolean = {
-    kvClient.get(ByteSequence.from(path.getBytes())).get().getKvs.isEmpty
+  override def pathNonExists(path: String, isPrefix: Boolean = true): Boolean = {
+    kvClient.get(
+      ByteSequence.from(path.getBytes()),
+      GetOption.newBuilder().isPrefix(isPrefix).build()).get().getKvs.isEmpty
   }
 
   override def delete(path: String, deleteChildren: Boolean = false): Unit = {
@@ -311,7 +313,7 @@ class EtcdDiscoveryClient(conf: KyuubiConf) extends DiscoveryClient {
   override def getAndIncrement(path: String, delta: Int = 1): Int = {
     val lockPath = s"${path}_tmp_for_lock"
     tryWithLock(lockPath, 60 * 1000) {
-      if (pathNonExists(path)) {
+      if (pathNonExists(path, isPrefix = false)) {
         create(path, "PERSISTENT")
         setData(path, String.valueOf(0).getBytes)
       }

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

@@ -86,7 +86,7 @@ class ZookeeperDiscoveryClient(conf: KyuubiConf) extends DiscoveryClient {
     zkClient.checkExists().forPath(path) != null
   }
 
-  override def pathNonExists(path: String): Boolean = {
+  override def pathNonExists(path: String, isPrefix: Boolean): Boolean = {
     zkClient.checkExists().forPath(path) == null
   }
 

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))
+
+      // get
+      assert(new String(discoveryClient.getData(path), StandardCharsets.UTF_8) == path)
+
+      // delete
+      discoveryClient.delete(path)
+      assert(!discoveryClient.pathExists(path))
+    }
+  }
 }