You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/08/28 06:37:32 UTC
(pulsar) branch branch-3.3 updated (406450dfc35 -> 6e15ca194ed)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 406450dfc35 Fix checkstyle
new d439d945710 [fix][broker] Invalid regex in PulsarLedgerManager causes zk data notification to be ignored (#23977)
new 6e15ca194ed [fix][meta] Use `getChildrenFromStore` to read children data to avoid lost data (#24665)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../java/org/apache/pulsar/metadata/api/MetadataStore.java | 14 ++++++++++++++
.../bookkeeper/AbstractHierarchicalLedgerManager.java | 8 +++++---
.../bookkeeper/LegacyHierarchicalLedgerRangeIterator.java | 12 ++++++++----
.../bookkeeper/LongHierarchicalLedgerRangeIterator.java | 2 +-
.../pulsar/metadata/bookkeeper/PulsarLedgerManager.java | 2 +-
.../apache/pulsar/metadata/impl/AbstractMetadataStore.java | 2 --
.../pulsar/metadata/impl/FaultInjectionMetadataStore.java | 10 ++++++++++
.../apache/pulsar/metadata/impl/RocksdbMetadataStore.java | 2 +-
.../impl/batching/AbstractBatchedMetadataStore.java | 2 +-
.../pulsar/metadata/impl/oxia/OxiaMetadataStore.java | 2 +-
.../pulsar/metadata/impl/MetadataStoreFactoryImplTest.java | 2 +-
11 files changed, 43 insertions(+), 15 deletions(-)
(pulsar) 01/02: [fix][broker] Invalid regex in PulsarLedgerManager causes zk data notification to be ignored (#23977)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d439d9457100cee314451f80774006ef71182596
Author: ken <16...@qq.com>
AuthorDate: Wed Aug 27 17:51:26 2025 +0800
[fix][broker] Invalid regex in PulsarLedgerManager causes zk data notification to be ignored (#23977)
Co-authored-by: fanjianye <fa...@bigo.sg>
(cherry picked from commit a53279837fae8b1c8bd16bec966cbc39823ed210)
---
.../java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerManager.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerManager.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerManager.java
index b003c656353..c426538aee4 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerManager.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerManager.java
@@ -292,7 +292,7 @@ public class PulsarLedgerManager implements LedgerManager {
}
}
- private static final Pattern ledgerPathRegex = Pattern.compile("/L[0-9]+$");
+ private static final Pattern ledgerPathRegex = Pattern.compile(".*/L[0-9]+$");
private void handleDataNotification(Notification n) {
if (!n.getPath().startsWith(ledgerRootPath)
(pulsar) 02/02: [fix][meta] Use `getChildrenFromStore` to read children data to avoid lost data (#24665)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 6e15ca194edd0facb8bae877cb5682a29050d660
Author: Jiwei Guo <te...@apache.org>
AuthorDate: Thu Aug 28 10:01:11 2025 +0800
[fix][meta] Use `getChildrenFromStore` to read children data to avoid lost data (#24665)
(cherry picked from commit 90a70db6fa81f887f65f60194fef3b36438433cb)
---
.../java/org/apache/pulsar/metadata/api/MetadataStore.java | 14 ++++++++++++++
.../bookkeeper/AbstractHierarchicalLedgerManager.java | 8 +++++---
.../bookkeeper/LegacyHierarchicalLedgerRangeIterator.java | 12 ++++++++----
.../bookkeeper/LongHierarchicalLedgerRangeIterator.java | 2 +-
.../apache/pulsar/metadata/impl/AbstractMetadataStore.java | 2 --
.../pulsar/metadata/impl/FaultInjectionMetadataStore.java | 10 ++++++++++
.../apache/pulsar/metadata/impl/RocksdbMetadataStore.java | 2 +-
.../impl/batching/AbstractBatchedMetadataStore.java | 2 +-
.../pulsar/metadata/impl/oxia/OxiaMetadataStore.java | 2 +-
.../pulsar/metadata/impl/MetadataStoreFactoryImplTest.java | 2 +-
10 files changed, 42 insertions(+), 14 deletions(-)
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataStore.java
index 89b0e7a6fe1..be261f047aa 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataStore.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataStore.java
@@ -77,6 +77,20 @@ public interface MetadataStore extends AutoCloseable {
*/
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.
*
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/AbstractHierarchicalLedgerManager.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/AbstractHierarchicalLedgerManager.java
index 4db7f4798c3..bd1c09ddf0f 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/AbstractHierarchicalLedgerManager.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/AbstractHierarchicalLedgerManager.java
@@ -68,7 +68,8 @@ abstract class AbstractHierarchicalLedgerManager {
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 @@ abstract class AbstractHierarchicalLedgerManager {
* 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 @@ abstract class AbstractHierarchicalLedgerManager {
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);
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/LegacyHierarchicalLedgerRangeIterator.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/LegacyHierarchicalLedgerRangeIterator.java
index 37e6dc836f2..cd9533843cc 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/LegacyHierarchicalLedgerRangeIterator.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/LegacyHierarchicalLedgerRangeIterator.java
@@ -69,7 +69,7 @@ public class LegacyHierarchicalLedgerRangeIterator implements LedgerManager.Ledg
* 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 @@ public class LegacyHierarchicalLedgerRangeIterator implements LedgerManager.Ledg
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 @@ public class LegacyHierarchicalLedgerRangeIterator implements LedgerManager.Ledg
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 @@ public class LegacyHierarchicalLedgerRangeIterator implements LedgerManager.Ledg
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) {
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/LongHierarchicalLedgerRangeIterator.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/LongHierarchicalLedgerRangeIterator.java
index 3b32916e6e7..92d3dcbe8f2 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/LongHierarchicalLedgerRangeIterator.java
+++ b/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()) {
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java
index ccbb2a11a90..bb9489fd293 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java
@@ -84,8 +84,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) {
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/FaultInjectionMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/FaultInjectionMetadataStore.java
index 360b8c91d0b..475e3a910a3 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/FaultInjectionMetadataStore.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/FaultInjectionMetadataStore.java
@@ -94,6 +94,16 @@ public class FaultInjectionMetadataStore implements MetadataStoreExtended {
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);
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/RocksdbMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/RocksdbMetadataStore.java
index 06f7b260536..10d9c399169 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/RocksdbMetadataStore.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/RocksdbMetadataStore.java
@@ -404,7 +404,7 @@ public class RocksdbMetadataStore extends AbstractMetadataStore {
}
@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);
}
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/batching/AbstractBatchedMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/batching/AbstractBatchedMetadataStore.java
index 5b45530d2e2..15c8b0b133e 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/batching/AbstractBatchedMetadataStore.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/batching/AbstractBatchedMetadataStore.java
@@ -134,7 +134,7 @@ public abstract class AbstractBatchedMetadataStore extends AbstractMetadataStore
}
@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();
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/oxia/OxiaMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/oxia/OxiaMetadataStore.java
index d2b8906b040..77a5eac88fd 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/oxia/OxiaMetadataStore.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/oxia/OxiaMetadataStore.java
@@ -132,7 +132,7 @@ public class OxiaMetadataStore extends AbstractMetadataStore {
}
@Override
- protected CompletableFuture<List<String>> getChildrenFromStore(String path) {
+ public CompletableFuture<List<String>> getChildrenFromStore(String path) {
var pathWithSlash = path + "/";
return client
diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/impl/MetadataStoreFactoryImplTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/impl/MetadataStoreFactoryImplTest.java
index 34e860aa578..55e4dc73fd3 100644
--- a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/impl/MetadataStoreFactoryImplTest.java
+++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/impl/MetadataStoreFactoryImplTest.java
@@ -98,7 +98,7 @@ public class MetadataStoreFactoryImplTest {
}
@Override
- protected CompletableFuture<List<String>> getChildrenFromStore(String path) {
+ public CompletableFuture<List<String>> getChildrenFromStore(String path) {
return null;
}