This is an automated email from the ASF dual-hosted git repository.
rdhabalia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new b8e6948f62d [fix][broker] Broker failed to load v1 namespace resources
cache (#20783)
b8e6948f62d is described below
commit b8e6948f62d6ec2ca53b6a85fe2fd07d4dee6853
Author: Rajan Dhabalia <rdhaba...@apache.org>
AuthorDate: Fri Jul 21 15:41:50 2023 -0700
[fix][broker] Broker failed to load v1 namespace resources cache (#20783)
---
.../pulsar/broker/resources/BaseResources.java | 36 ++++++++++++++++++++++
.../broker/resources/NamespaceResources.java | 2 +-
.../pulsar/broker/admin/PersistentTopicsTest.java | 14 +++++++++
3 files changed, 51 insertions(+), 1 deletion(-)
diff --git
a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BaseResources.java
b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BaseResources.java
index 42add4271f6..4011a482075 100644
---
a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BaseResources.java
+++
b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BaseResources.java
@@ -20,11 +20,16 @@ package org.apache.pulsar.broker.resources;
import com.fasterxml.jackson.core.type.TypeReference;
import com.google.common.base.Joiner;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
@@ -78,6 +83,37 @@ public class BaseResources<T> {
return cache.getChildren(path);
}
+ protected CompletableFuture<List<String>> getChildrenRecursiveAsync(String
path) {
+ Set<String> children = ConcurrentHashMap.newKeySet();
+ CompletableFuture<List<String>> result = new CompletableFuture<>();
+ getChildrenRecursiveAsync(path, children, result, new
AtomicInteger(1), path);
+ return result;
+ }
+
+ private void getChildrenRecursiveAsync(String path, Set<String> children,
CompletableFuture<List<String>> result,
+ AtomicInteger totalResults, String parent) {
+ cache.getChildren(path).thenAccept(childList -> {
+ childList = childList != null ? childList :
Collections.emptyList();
+ if (totalResults.decrementAndGet() == 0 && childList.isEmpty()) {
+ result.complete(new ArrayList<>(children));
+ return;
+ }
+ if (childList.isEmpty()) {
+ return;
+ }
+ // remove current node from children if current node is not leaf
+ children.remove(parent);
+ // childPrefix creates a path hierarchy if children has multi
level path
+ String childPrefix = path.equals(parent) ? "" : parent + "/";
+ totalResults.addAndGet(childList.size());
+ for (String child : childList) {
+ children.add(childPrefix + child);
+ String childPath = path + "/" + child;
+ getChildrenRecursiveAsync(childPath, children, result,
totalResults, child);
+ }
+ });
+ }
+
protected Optional<T> get(String path) throws MetadataStoreException {
try {
return getAsync(path).get(operationTimeoutSec, TimeUnit.SECONDS);
diff --git
a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/NamespaceResources.java
b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/NamespaceResources.java
index 48f82596567..e5dd13c32eb 100644
---
a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/NamespaceResources.java
+++
b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/NamespaceResources.java
@@ -62,7 +62,7 @@ public class NamespaceResources extends
BaseResources<Policies> {
}
public CompletableFuture<List<String>> listNamespacesAsync(String tenant) {
- return getChildrenAsync(joinPath(BASE_POLICIES_PATH, tenant));
+ return getChildrenRecursiveAsync(joinPath(BASE_POLICIES_PATH, tenant));
}
public CompletableFuture<Boolean> getPoliciesReadOnlyAsync() {
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
index 83ed63bf0d9..a4f6bd4650f 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
@@ -31,6 +31,7 @@ import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertTrue;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.HashMap;
@@ -1700,4 +1701,17 @@ public class PersistentTopicsTest extends
MockedPulsarServiceBaseTest {
// verify we only call getReplicatedSubscriptionStatusAsync once.
verify(topics, times(1)).getReplicatedSubscriptionStatusAsync(any(),
any());
}
+
+ @Test
+ public void testNamespaceResources() throws Exception {
+ String ns1V1 = "test/" + testNamespace + "v1";
+ String ns1V2 = testNamespace + "v2";
+ admin.namespaces().createNamespace(testTenant+"/"+ns1V1);
+ admin.namespaces().createNamespace(testTenant+"/"+ns1V2);
+
+ List<String> namespaces =
pulsar.getPulsarResources().getNamespaceResources().listNamespacesAsync(testTenant)
+ .get();
+ assertTrue(namespaces.contains(ns1V2));
+ assertTrue(namespaces.contains(ns1V1));
+ }
}
