Skip to content

Commit 52e57a6

Browse files
Demogorgon314Technoboy-
authored andcommitted
[fix][broker] Fix redirect loop when using ExtensibleLoadManager and list in bundle admin API (#20528)
PIP: #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.
1 parent 692c182 commit 52e57a6

File tree

2 files changed

+40
-19
lines changed

2 files changed

+40
-19
lines changed

pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java

+1-17
Original file line numberDiff line numberDiff line change
@@ -1102,19 +1102,7 @@ public Set<NamespaceBundle> getOwnedServiceUnits() {
11021102
}
11031103

11041104
public boolean isServiceUnitOwned(ServiceUnitId suName) throws Exception {
1105-
if (suName instanceof TopicName) {
1106-
return isTopicOwnedAsync((TopicName) suName).get();
1107-
}
1108-
1109-
if (suName instanceof NamespaceName) {
1110-
return isNamespaceOwned((NamespaceName) suName);
1111-
}
1112-
1113-
if (suName instanceof NamespaceBundle) {
1114-
return ownershipCache.isNamespaceBundleOwned((NamespaceBundle) suName);
1115-
}
1116-
1117-
throw new IllegalArgumentException("Invalid class of NamespaceBundle: " + suName.getClass().getName());
1105+
return isServiceUnitOwnedAsync(suName).get(config.getMetadataStoreOperationTimeoutSeconds(), TimeUnit.SECONDS);
11181106
}
11191107

11201108
public CompletableFuture<Boolean> isServiceUnitOwnedAsync(ServiceUnitId suName) {
@@ -1168,10 +1156,6 @@ public CompletableFuture<Boolean> isServiceUnitActiveAsync(TopicName topicName)
11681156
});
11691157
}
11701158

1171-
private boolean isNamespaceOwned(NamespaceName fqnn) throws Exception {
1172-
return ownershipCache.getOwnedBundle(getFullBundle(fqnn)) != null;
1173-
}
1174-
11751159
private CompletableFuture<Boolean> isNamespaceOwnedAsync(NamespaceName fqnn) {
11761160
// TODO: Add unit tests cover it.
11771161
if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config)) {

pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java

+39-2
Original file line numberDiff line numberDiff line change
@@ -48,10 +48,11 @@
4848
import static org.testng.Assert.assertNull;
4949
import static org.testng.Assert.assertTrue;
5050
import static org.testng.Assert.fail;
51-
5251
import com.google.common.collect.Sets;
5352
import java.util.LinkedHashMap;
5453
import java.util.Set;
54+
import java.util.UUID;
55+
import java.util.concurrent.CompletableFuture;
5556
import java.util.concurrent.ExecutionException;
5657
import java.util.concurrent.TimeUnit;
5758
import java.util.concurrent.atomic.AtomicInteger;
@@ -63,7 +64,6 @@
6364
import java.util.List;
6465
import java.util.Map;
6566
import java.util.Optional;
66-
import java.util.concurrent.CompletableFuture;
6767
import org.apache.commons.lang3.reflect.FieldUtils;
6868
import org.apache.pulsar.broker.PulsarService;
6969
import org.apache.pulsar.broker.ServiceConfiguration;
@@ -927,6 +927,43 @@ public void testDisableBroker() throws Exception {
927927
}
928928
}
929929

930+
@Test(timeOut = 30 * 1000)
931+
public void testListTopic() throws Exception {
932+
final String namespace = "public/testListTopic";
933+
admin.namespaces().createNamespace(namespace, 3);
934+
935+
final String persistentTopicName = TopicName.get(
936+
"persistent", NamespaceName.get(namespace),
937+
"get_topics_mode_" + UUID.randomUUID()).toString();
938+
939+
final String nonPersistentTopicName = TopicName.get(
940+
"non-persistent", NamespaceName.get(namespace),
941+
"get_topics_mode_" + UUID.randomUUID()).toString();
942+
admin.topics().createPartitionedTopic(persistentTopicName, 3);
943+
admin.topics().createPartitionedTopic(nonPersistentTopicName, 3);
944+
pulsarClient.newProducer().topic(persistentTopicName).create().close();
945+
pulsarClient.newProducer().topic(nonPersistentTopicName).create().close();
946+
947+
BundlesData bundlesData = admin.namespaces().getBundles(namespace);
948+
List<String> boundaries = bundlesData.getBoundaries();
949+
int topicNum = 0;
950+
for (int i = 0; i < boundaries.size() - 1; i++) {
951+
String bundle = String.format("%s_%s", boundaries.get(i), boundaries.get(i + 1));
952+
List<String> topic = admin.topics().getListInBundle(namespace, bundle);
953+
if (topic == null) {
954+
continue;
955+
}
956+
topicNum += topic.size();
957+
for (String s : topic) {
958+
assertFalse(TopicName.get(s).isPersistent());
959+
}
960+
}
961+
assertEquals(topicNum, 3);
962+
963+
List<String> list = admin.topics().getList(namespace);
964+
assertEquals(list.size(), 6);
965+
}
966+
930967
private static abstract class MockBrokerFilter implements BrokerFilter {
931968

932969
@Override

0 commit comments

Comments
 (0)