[fix][meta] Use getChildrenFromStore to read children data to avoid lost data (apache#24665)

(cherry picked from commit 90a70db)
(cherry picked from commit e4431bc)

Author: Technoboy-

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataStore.java

@@ -77,6 +77,20 @@ default CompletableFuture<Void> sync(String path) {
      */
     CompletableFuture<List<String>> getChildren(String path);
 
+
+    /**
+     * Return all the nodes (lexicographically sorted) that are children to the specific path.
+     *
+     * If the path itself does not exist, it will return an empty list.
+     *
+     * This method is similar to {@link #getChildren(String)}, but it attempts to read directly from
+     *  the underlying store.
+     *
+     * @param path
+     *            the path of the key to get from the store
+     * @return a future to track the async request
+     */
+    CompletableFuture<List<String>> getChildrenFromStore(String path);
     /**
      * Read whether a specific path exists.
      *

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/AbstractHierarchicalLedgerManager.java

@@ -68,7 +68,8 @@ void asyncProcessLevelNodes(
             final AsyncCallback.VoidCallback finalCb, final Object context,
             final int successRc, final int failureRc) {
 
-        store.getChildren(path)
+        store.sync(path)
+                .thenCompose(__ -> store.getChildrenFromStore(path))
                 .thenAccept(levelNodes -> {
                     if (levelNodes.isEmpty()) {
                         finalCb.processResult(successRc, null, context);
@@ -162,7 +163,7 @@ long getLedgerId(String... levelNodes) throws IOException {
      * Process ledgers in a single zk node.
      *
      * <p>
-     * for each ledger found in this zk node, processor#process(ledgerId) will be triggerred
+     * for each ledger found in this zk node, processor#process(ledgerId) will be triggered
      * to process a specific ledger. after all ledgers has been processed, the finalCb will
      * be called with provided context object. The RC passed to finalCb is decided by :
      * <ul>
@@ -188,7 +189,8 @@ protected void asyncProcessLedgersInSingleNode(
             final String path, final BookkeeperInternalCallbacks.Processor<Long> processor,
             final AsyncCallback.VoidCallback finalCb, final Object ctx,
             final int successRc, final int failureRc) {
-        store.getChildren(path)
+        store.sync(path)
+                .thenCompose(__ -> store.getChildrenFromStore(path))
                 .thenAccept(ledgerNodes -> {
                     Set<Long> activeLedgers = HierarchicalLedgerUtils.ledgerListToSet(ledgerNodes,
                             ledgerRootPath, path);

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/LegacyHierarchicalLedgerRangeIterator.java

@@ -69,7 +69,7 @@ public LegacyHierarchicalLedgerRangeIterator(MetadataStore store, String ledgers
      * Iterate next level1 znode.
      *
      * @return false if have visited all level1 nodes
-     * @throws InterruptedException/KeeperException if error occurs reading zookeeper children
+     * @throws InterruptedException/ExecutionException/TimeoutException if error occurs reading zookeeper children
      */
     private boolean nextL1Node() throws ExecutionException, InterruptedException, TimeoutException {
         l2NodesIter = null;
@@ -83,7 +83,9 @@ private boolean nextL1Node() throws ExecutionException, InterruptedException, Ti
             if (!isLedgerParentNode(curL1Nodes)) {
                 continue;
             }
-            List<String> l2Nodes = store.getChildren(ledgersRoot + "/" + curL1Nodes)
+            String path = ledgersRoot + "/" + curL1Nodes;
+            List<String> l2Nodes = store.sync(path)
+                    .thenCompose(__ -> store.getChildrenFromStore(path))
                     .get(BLOCKING_CALL_TIMEOUT, MILLISECONDS);
             l2NodesIter = l2Nodes.iterator();
             if (!l2NodesIter.hasNext()) {
@@ -99,7 +101,8 @@ private synchronized void preload() throws IOException {
             boolean hasMoreElements = false;
             try {
                 if (l1NodesIter == null) {
-                    List<String> l1Nodes = store.getChildren(ledgersRoot)
+                    List<String> l1Nodes = store.sync(ledgersRoot)
+                            .thenCompose(__ -> store.getChildrenFromStore(ledgersRoot))
                             .get(BLOCKING_CALL_TIMEOUT, MILLISECONDS);
                     l1NodesIter = l1Nodes.iterator();
                     hasMoreElements = nextL1Node();
@@ -162,7 +165,8 @@ LedgerManager.LedgerRange getLedgerRangeByLevel(final String level1, final Strin
         String nodePath = nodeBuilder.toString();
         List<String> ledgerNodes = null;
         try {
-            ledgerNodes = store.getChildren(nodePath).get(BLOCKING_CALL_TIMEOUT, MILLISECONDS);
+            ledgerNodes = store.sync(nodePath).thenCompose(__ -> store.getChildrenFromStore(nodePath))
+                    .get(BLOCKING_CALL_TIMEOUT, MILLISECONDS);
         } catch (ExecutionException | TimeoutException e) {
             throw new IOException("Error when get child nodes from zk", e);
         } catch (InterruptedException e) {

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/LongHierarchicalLedgerRangeIterator.java

@@ -59,7 +59,7 @@ class LongHierarchicalLedgerRangeIterator implements LedgerManager.LedgerRangeIt
      */
     List<String> getChildrenAt(String path) throws IOException {
         try {
-            return store.getChildren(path)
+            return store.sync(path).thenCompose(__ -> store.getChildrenFromStore(path))
                     .get(AbstractMetadataDriver.BLOCKING_CALL_TIMEOUT, TimeUnit.MILLISECONDS);
         } catch (ExecutionException | TimeoutException e) {
             if (log.isDebugEnabled()) {

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java

@@ -87,8 +87,6 @@ public abstract class AbstractMetadataStore implements MetadataStoreExtended, Co
 
     protected final AtomicBoolean isClosed = new AtomicBoolean(false);
 
-    protected abstract CompletableFuture<List<String>> getChildrenFromStore(String path);
-
     protected abstract CompletableFuture<Boolean> existsFromStore(String path);
 
     protected AbstractMetadataStore(String metadataStoreName, OpenTelemetry openTelemetry) {

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/FaultInjectionMetadataStore.java

@@ -94,6 +94,16 @@ public CompletableFuture<List<String>> getChildren(String path) {
         return store.getChildren(path);
     }
 
+    @Override
+    public CompletableFuture<List<String>> getChildrenFromStore(String path) {
+        Optional<MetadataStoreException> ex = programmedFailure(OperationType.GET_CHILDREN, path);
+        if (ex.isPresent()) {
+            return FutureUtil.failedFuture(ex.get());
+        }
+
+        return store.getChildrenFromStore(path);
+    }
+
     @Override
     public CompletableFuture<Boolean> exists(String path) {
         Optional<MetadataStoreException> ex = programmedFailure(OperationType.EXISTS, path);

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/RocksdbMetadataStore.java

@@ -404,7 +404,7 @@ public CompletableFuture<Optional<GetResult>> storeGet(String path) {
     }
 
     @Override
-    protected CompletableFuture<List<String>> getChildrenFromStore(String path) {
+    public CompletableFuture<List<String>> getChildrenFromStore(String path) {
         if (log.isDebugEnabled()) {
             log.debug("getChildrenFromStore.path={},instanceId={}", path, instanceId);
         }

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/batching/AbstractBatchedMetadataStore.java

@@ -144,7 +144,7 @@ public final CompletableFuture<Optional<GetResult>> storeGet(String path) {
     }
 
     @Override
-    protected final CompletableFuture<List<String>> getChildrenFromStore(String path) {
+    public final CompletableFuture<List<String>> getChildrenFromStore(String path) {
         OpGetChildren op = new OpGetChildren(path);
         enqueue(readOps, op);
         return op.getFuture();

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/oxia/OxiaMetadataStore.java

@@ -147,7 +147,7 @@ Stat convertStat(String path, Version version) {
     }
 
     @Override
-    protected CompletableFuture<List<String>> getChildrenFromStore(String path) {
+    public CompletableFuture<List<String>> getChildrenFromStore(String path) {
         var pathWithSlash = path + "/";
 
         return client

pulsar-metadata/src/test/java/org/apache/pulsar/metadata/impl/MetadataStoreFactoryImplTest.java

@@ -100,7 +100,7 @@ protected MyMetadataStore() {
         }
 
         @Override
-        protected CompletableFuture<List<String>> getChildrenFromStore(String path) {
+        public CompletableFuture<List<String>> getChildrenFromStore(String path) {
             return null;
         }