You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2023/06/14 05:31:47 UTC
[pulsar] branch branch-3.0 updated: [fix][broker] Fix redirect loop when using ExtensibleLoadManager and list in bundle admin API (#20528)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new 52e57a6e49f [fix][broker] Fix redirect loop when using ExtensibleLoadManager and list in bundle admin API (#20528)
52e57a6e49f is described below
commit 52e57a6e49fb5a1055476913ad6e6bc3f430c14b
Author: Kai Wang <kw...@apache.org>
AuthorDate: Thu Jun 8 13:58:08 2023 +0800
[fix][broker] Fix redirect loop when using ExtensibleLoadManager and list in bundle admin API (#20528)
PIP: https://github.com/apache/pulsar/issues/16691
When using `ExtensibleLoadManager` and list in bundle admin API,
it will redirect forever because `isServiceUnitOwned` method is checking the `ownershipCache` as the ownership storage, however, when using `ExtensibleLoadManager`, it stored the ownership to table view.
* Call `isServiceUnitOwnedAsync ` when using `isServiceUnitOwned `.
* Add unit test to cover this case.
---
.../pulsar/broker/namespace/NamespaceService.java | 18 +---------
.../extensions/ExtensibleLoadManagerImplTest.java | 41 ++++++++++++++++++++--
2 files changed, 40 insertions(+), 19 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
index e2d4ef51537..c5d516cfc94 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
@@ -1102,19 +1102,7 @@ public class NamespaceService implements AutoCloseable {
}
public boolean isServiceUnitOwned(ServiceUnitId suName) throws Exception {
- if (suName instanceof TopicName) {
- return isTopicOwnedAsync((TopicName) suName).get();
- }
-
- if (suName instanceof NamespaceName) {
- return isNamespaceOwned((NamespaceName) suName);
- }
-
- if (suName instanceof NamespaceBundle) {
- return ownershipCache.isNamespaceBundleOwned((NamespaceBundle) suName);
- }
-
- throw new IllegalArgumentException("Invalid class of NamespaceBundle: " + suName.getClass().getName());
+ return isServiceUnitOwnedAsync(suName).get(config.getMetadataStoreOperationTimeoutSeconds(), TimeUnit.SECONDS);
}
public CompletableFuture<Boolean> isServiceUnitOwnedAsync(ServiceUnitId suName) {
@@ -1168,10 +1156,6 @@ public class NamespaceService implements AutoCloseable {
});
}
- private boolean isNamespaceOwned(NamespaceName fqnn) throws Exception {
- return ownershipCache.getOwnedBundle(getFullBundle(fqnn)) != null;
- }
-
private CompletableFuture<Boolean> isNamespaceOwnedAsync(NamespaceName fqnn) {
// TODO: Add unit tests cover it.
if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config)) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
index 498f48d16d2..d6e208d3202 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
@@ -48,10 +48,11 @@ import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
-
import com.google.common.collect.Sets;
import java.util.LinkedHashMap;
import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -63,7 +64,6 @@ import java.net.URL;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
@@ -927,6 +927,43 @@ public class ExtensibleLoadManagerImplTest extends MockedPulsarServiceBaseTest {
}
}
+ @Test(timeOut = 30 * 1000)
+ public void testListTopic() throws Exception {
+ final String namespace = "public/testListTopic";
+ admin.namespaces().createNamespace(namespace, 3);
+
+ final String persistentTopicName = TopicName.get(
+ "persistent", NamespaceName.get(namespace),
+ "get_topics_mode_" + UUID.randomUUID()).toString();
+
+ final String nonPersistentTopicName = TopicName.get(
+ "non-persistent", NamespaceName.get(namespace),
+ "get_topics_mode_" + UUID.randomUUID()).toString();
+ admin.topics().createPartitionedTopic(persistentTopicName, 3);
+ admin.topics().createPartitionedTopic(nonPersistentTopicName, 3);
+ pulsarClient.newProducer().topic(persistentTopicName).create().close();
+ pulsarClient.newProducer().topic(nonPersistentTopicName).create().close();
+
+ BundlesData bundlesData = admin.namespaces().getBundles(namespace);
+ List<String> boundaries = bundlesData.getBoundaries();
+ int topicNum = 0;
+ for (int i = 0; i < boundaries.size() - 1; i++) {
+ String bundle = String.format("%s_%s", boundaries.get(i), boundaries.get(i + 1));
+ List<String> topic = admin.topics().getListInBundle(namespace, bundle);
+ if (topic == null) {
+ continue;
+ }
+ topicNum += topic.size();
+ for (String s : topic) {
+ assertFalse(TopicName.get(s).isPersistent());
+ }
+ }
+ assertEquals(topicNum, 3);
+
+ List<String> list = admin.topics().getList(namespace);
+ assertEquals(list.size(), 6);
+ }
+
private static abstract class MockBrokerFilter implements BrokerFilter {
@Override