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 2025/09/01 08:09:20 UTC
(pulsar) branch branch-4.0 updated: [fix][broker][branch-4.0]Can not access topic policies if topic partitions have not been created (#24680)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 109ba07040f [fix][broker][branch-4.0]Can not access topic policies if topic partitions have not been created (#24680)
109ba07040f is described below
commit 109ba07040f49f45216706e1af8e3f4e3557c730
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Mon Sep 1 16:09:14 2025 +0800
[fix][broker][branch-4.0]Can not access topic policies if topic partitions have not been created (#24680)
---
.../org/apache/pulsar/broker/admin/AdminResource.java | 16 +++++++++++++---
.../org/apache/pulsar/broker/service/ReplicatorTest.java | 14 ++++++++++++++
.../apache/pulsar/client/api/PartitionCreationTest.java | 15 +++++++++++++++
3 files changed, 42 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
index 0f0b59e8961..684f65bbdd3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
@@ -750,15 +750,25 @@ public abstract class AdminResource extends PulsarWebResource {
return pulsar().getNamespaceService().getListOfTopics(topicName.getNamespaceObject(),
CommandGetTopicsOfNamespace.Mode.ALL)
.thenCompose(topics -> {
- boolean exists = false;
+ boolean persistentResourceCreated = false;
for (String topic : topics) {
if (topicName.getPartitionedTopicName().equals(
TopicName.get(topic).getPartitionedTopicName())) {
- exists = true;
+ persistentResourceCreated = true;
break;
}
}
- return CompletableFuture.completedFuture(exists);
+ if (persistentResourceCreated) {
+ return CompletableFuture.completedFuture(true);
+ }
+ return pulsar().getPulsarResources().getNamespaceResources().getPartitionedTopicResources()
+ .getPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()), false)
+ .thenApply(optMetadata -> {
+ if (optMetadata.isEmpty()) {
+ return false;
+ }
+ return optMetadata.get().partitions > topicName.getPartitionIndex();
+ });
});
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
index 2e1ed43bd8a..f14cba61e60 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
@@ -1538,6 +1538,20 @@ public class ReplicatorTest extends ReplicatorTestBase {
for (int i = 0; i < numPartitions; i++) {
expectedTopicList.add(pt.getPartition(i).toString());
}
+ PersistentTopic partition0 = (PersistentTopic) pulsar1.getBrokerService()
+ .getTopic(TopicName.get(persistentPartitionedTopic).getPartition(0).toString(), false)
+ .join().get();
+ PersistentTopic partition1 = (PersistentTopic) pulsar1.getBrokerService()
+ .getTopic(TopicName.get(persistentPartitionedTopic).getPartition(1).toString(), false)
+ .join().get();
+ PersistentTopic partition2 = (PersistentTopic) pulsar1.getBrokerService()
+ .getTopic(TopicName.get(persistentPartitionedTopic).getPartition(2).toString(), false)
+ .join().get();
+ Awaitility.await().untilAsserted(() -> {
+ assertTrue(partition0.getReplicators().size() >= 2);
+ assertTrue(partition1.getReplicators().size() >= 2);
+ assertTrue(partition2.getReplicators().size() >= 2);
+ });
checkListContainExpectedTopic(admin1, namespace, expectedTopicList);
checkListContainExpectedTopic(admin2, namespace, expectedTopicList);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionCreationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionCreationTest.java
index ddb3e91a730..b3b9c9cec2c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionCreationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionCreationTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.api;
+import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.impl.MultiTopicsConsumerImpl;
import org.apache.pulsar.common.naming.TopicDomain;
@@ -102,6 +103,20 @@ public class PartitionCreationTest extends ProducerConsumerBase {
Assert.assertNotNull(pulsarClient.newConsumer().topic(topic).subscriptionName("sub-1").subscribe());
}
+ @Test
+ public void testGetPoliciesIfPartitionsNotCreated() throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+ int numPartitions = 3;
+ // simulate partitioned topic without partitions
+ pulsar.getPulsarResources().getNamespaceResources().getPartitionedTopicResources()
+ .createPartitionedTopicAsync(TopicName.get(topic),
+ new PartitionedTopicMetadata(numPartitions)).join();
+ // Verify: the command will not get an topic not found error.
+ admin.topics().getReplicationClusters(topic, true);
+ // cleanup.
+ admin.topics().deletePartitionedTopic(topic);
+ }
+
@Test(timeOut = 60000)
public void testCreateConsumerForPartitionedTopicUpdateWhenDisableTopicAutoCreation() throws Exception {
conf.setAllowAutoTopicCreation(false);