You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zi...@apache.org on 2025/04/08 09:53:37 UTC
(pulsar) branch master updated: [fix][broker] Directly query single topic existence when the topic is partitioned (#24154)
This is an automated email from the ASF dual-hosted git repository.
zixuan 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 0d6c6f4b9e8 [fix][broker] Directly query single topic existence when the topic is partitioned (#24154)
0d6c6f4b9e8 is described below
commit 0d6c6f4b9e88a4cd10df50e6f1a9422772ca0779
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Tue Apr 8 17:53:29 2025 +0800
[fix][broker] Directly query single topic existence when the topic is partitioned (#24154)
Signed-off-by: Zixuan Liu <no...@gmail.com>
---
.../pulsar/broker/namespace/NamespaceService.java | 16 +--
.../broker/namespace/NamespaceServiceTest.java | 107 +++++++++++++++++----
.../broker/service/BrokerServiceChaosTest.java | 4 +-
...OneWayReplicatorUsingGlobalPartitionedTest.java | 4 +-
.../pulsar/client/api/ConsumerCreationTest.java | 2 +-
.../pulsar/client/api/ProducerCreationTest.java | 2 +-
.../pulsar/client/api/PulsarClientException.java | 4 +-
7 files changed, 107 insertions(+), 32 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 8f5cef1bdff..e6025ccceb1 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
@@ -1416,18 +1416,22 @@ public class NamespaceService implements AutoCloseable {
return pulsar.getBrokerService().fetchPartitionedTopicMetadataAsync(
topic.isPartitioned() ? TopicName.get(topic.getPartitionedTopicName()) : topic)
.thenCompose(metadata -> {
+ // When the topic has metadata:
+ // - The topic name is non-partitioned, which means that the topic exists.
+ // - The topic name is partitioned, please check the specific partition.
if (metadata.partitions > 0) {
if (!topic.isPartitioned()) {
return CompletableFuture.completedFuture(
TopicExistsInfo.newPartitionedTopicExists(metadata.partitions));
- } else {
- if (topic.getPartitionIndex() < metadata.partitions) {
- return CompletableFuture.completedFuture(
- TopicExistsInfo.newNonPartitionedTopicExists());
- }
+ }
+ if (!topic.isPersistent()) {
+ // A non-persistent partitioned topic contains only metadata.
+ // Since no actual partitions are created, there's no need to check under /managed-ledgers.
+ return CompletableFuture.completedFuture(topic.getPartitionIndex() < metadata.partitions
+ ? TopicExistsInfo.newNonPartitionedTopicExists()
+ : TopicExistsInfo.newTopicNotExists());
}
}
- // Direct query the single topic.
return checkNonPartitionedTopicExists(topic).thenApply(
b -> b ? TopicExistsInfo.newNonPartitionedTopicExists() :
TopicExistsInfo.newTopicNotExists());
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java
index 951247bd688..2a5989eb53d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.broker.namespace;
import static org.apache.pulsar.broker.resources.LoadBalanceResources.BUNDLE_DATA_BASE_PATH;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
@@ -86,6 +87,7 @@ import org.apache.pulsar.common.policies.data.LocalPolicies;
import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.policies.data.TenantInfo;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.common.policies.data.TopicType;
import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.apache.pulsar.metadata.api.GetResult;
import org.apache.pulsar.metadata.api.MetadataCache;
@@ -816,23 +818,6 @@ public class NamespaceServiceTest extends BrokerTestBase {
};
}
- @Test(dataProvider = "topicDomain")
- public void testCheckTopicExists(String topicDomain) throws Exception {
- String topic = topicDomain + "://prop/ns-abc/" + UUID.randomUUID();
- admin.topics().createNonPartitionedTopic(topic);
- Awaitility.await().untilAsserted(() -> {
- assertTrue(pulsar.getNamespaceService().checkTopicExists(TopicName.get(topic)).get().isExists());
- });
-
- String partitionedTopic = topicDomain + "://prop/ns-abc/" + UUID.randomUUID();
- admin.topics().createPartitionedTopic(partitionedTopic, 5);
- Awaitility.await().untilAsserted(() -> {
- assertTrue(pulsar.getNamespaceService().checkTopicExists(TopicName.get(partitionedTopic)).get().isExists());
- assertTrue(pulsar.getNamespaceService()
- .checkTopicExists(TopicName.get(partitionedTopic + "-partition-2")).get().isExists());
- });
- }
-
@Test
public void testAllowedClustersAtNamespaceLevelShouldBeIncludedInAllowedClustersAtTenantLevel() throws Exception {
// 1. Setup
@@ -954,6 +939,94 @@ public class NamespaceServiceTest extends BrokerTestBase {
pulsar.getConfiguration().setForceDeleteTenantAllowed(false);
}
+
+ @Test(dataProvider = "topicDomain")
+ public void checkTopicExistsForNonPartitionedTopic(String topicDomain) throws Exception {
+ TopicName topicName = TopicName.get(topicDomain, "prop", "ns-abc", "topic-" + UUID.randomUUID());
+ admin.topics().createNonPartitionedTopic(topicName.toString());
+ CompletableFuture<TopicExistsInfo> result = pulsar.getNamespaceService().checkTopicExists(topicName);
+ assertThat(result)
+ .succeedsWithin(3, TimeUnit.SECONDS)
+ .satisfies(n -> {
+ assertTrue(n.isExists());
+ assertEquals(n.getPartitions(), 0);
+ assertEquals(n.getTopicType(), TopicType.NON_PARTITIONED);
+ n.recycle();
+ });
+ }
+
+ @Test(dataProvider = "topicDomain")
+ public void checkTopicExistsForPartitionedTopic(String topicDomain) throws Exception {
+ TopicName topicName = TopicName.get(topicDomain, "prop", "ns-abc", "topic-" + UUID.randomUUID());
+ admin.topics().createPartitionedTopic(topicName.toString(), 3);
+
+ // Check the topic exists by the partitions.
+ CompletableFuture<TopicExistsInfo> result = pulsar.getNamespaceService().checkTopicExists(topicName);
+ assertThat(result)
+ .succeedsWithin(3, TimeUnit.SECONDS)
+ .satisfies(n -> {
+ assertTrue(n.isExists());
+ assertEquals(n.getPartitions(), 3);
+ assertEquals(n.getTopicType(), TopicType.PARTITIONED);
+ n.recycle();
+ });
+
+ // Check the specific partition.
+ result = pulsar.getNamespaceService().checkTopicExists(topicName.getPartition(2));
+ assertThat(result)
+ .succeedsWithin(3, TimeUnit.SECONDS)
+ .satisfies(n -> {
+ assertTrue(n.isExists());
+ assertEquals(n.getPartitions(), 0);
+ assertEquals(n.getTopicType(), TopicType.NON_PARTITIONED);
+ n.recycle();
+ });
+
+ // Partition index is out of range.
+ result = pulsar.getNamespaceService().checkTopicExists(topicName.getPartition(10));
+ assertThat(result)
+ .succeedsWithin(3, TimeUnit.SECONDS)
+ .satisfies(n -> {
+ assertFalse(n.isExists());
+ assertEquals(n.getPartitions(), 0);
+ assertEquals(n.getTopicType(), TopicType.NON_PARTITIONED);
+ n.recycle();
+ });
+ }
+
+ @Test(dataProvider = "topicDomain")
+ public void checkTopicExistsForNonExistentNonPartitionedTopic(String topicDomain) {
+ TopicName topicName = TopicName.get(topicDomain, "prop", "ns-abc", "topic-" + UUID.randomUUID());
+ CompletableFuture<TopicExistsInfo> result = pulsar.getNamespaceService().checkTopicExists(topicName);
+ assertThat(result)
+ .succeedsWithin(3, TimeUnit.SECONDS)
+ .satisfies(n -> {
+ // when using the pulsar client to check non_persistent topic, always return true, so ignore to
+ // check that.
+ if (topicDomain.equals(TopicDomain.persistent)) {
+ assertFalse(n.isExists());
+ }
+ n.recycle();
+ });
+ }
+
+ @Test(dataProvider = "topicDomain")
+ public void checkTopicExistsForNonExistentPartitionTopic(String topicDomain) {
+ TopicName topicName =
+ TopicName.get(topicDomain, "prop", "ns-abc", "topic-" + UUID.randomUUID() + "-partition-10");
+ CompletableFuture<TopicExistsInfo> result = pulsar.getNamespaceService().checkTopicExists(topicName);
+ assertThat(result)
+ .succeedsWithin(3, TimeUnit.SECONDS)
+ .satisfies(n -> {
+ // when using the pulsar client to check non_persistent topic, always return true, so ignore to
+ // check that.
+ if (topicDomain.equals(TopicDomain.persistent)) {
+ assertFalse(n.isExists());
+ }
+ n.recycle();
+ });
+ }
+
/**
* 1. Manually trigger "LoadReportUpdaterTask"
* 2. Registry another new zk-node-listener "waitForBrokerChangeNotice".
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceChaosTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceChaosTest.java
index 73db8513794..4187364e46f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceChaosTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceChaosTest.java
@@ -36,9 +36,7 @@ import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@Slf4j
-// TODO: This test is in flaky group until CI is fixed.
-// To be addressed as part of https://github.com/apache/pulsar/pull/24154
-@Test(groups = "flaky")
+@Test(groups = "broker")
public class BrokerServiceChaosTest extends CanReconnectZKClientPulsarServiceBaseTest {
@Override
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalPartitionedTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalPartitionedTest.java
index a22067101c3..2a2a1befd16 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalPartitionedTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalPartitionedTest.java
@@ -41,9 +41,7 @@ import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@Slf4j
-// TODO: This test is in flaky group until CI is fixed.
-// To be addressed as part of https://github.com/apache/pulsar/pull/24154
-@Test(groups = "flaky")
+@Test(groups = "broker")
public class OneWayReplicatorUsingGlobalPartitionedTest extends OneWayReplicatorTest {
@Override
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerCreationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerCreationTest.java
index a81dbe02b34..195485739e0 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerCreationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerCreationTest.java
@@ -117,7 +117,7 @@ public class ConsumerCreationTest extends ProducerConsumerBase {
Consumer<byte[]> ignored =
pulsarClient.newConsumer().topic(partitionedTopic).subscriptionName("my-sub").subscribe();
} else {
- assertThrows(PulsarClientException.TopicDoesNotExistException.class, () -> {
+ assertThrows(PulsarClientException.NotFoundException.class, () -> {
@Cleanup
Consumer<byte[]> ignored =
pulsarClient.newConsumer().topic(partitionedTopic).subscriptionName("my-sub").subscribe();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCreationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCreationTest.java
index cd75383a487..48ee112d750 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCreationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCreationTest.java
@@ -258,7 +258,7 @@ public class ProducerCreationTest extends ProducerConsumerBase {
@Cleanup
Producer<byte[]> ignored = pulsarClient.newProducer().topic(partitionedTopic).create();
} else {
- assertThrows(PulsarClientException.TopicDoesNotExistException.class, () -> {
+ assertThrows(PulsarClientException.NotFoundException.class, () -> {
@Cleanup
Producer<byte[]> ignored = pulsarClient.newProducer().topic(partitionedTopic).create();
});
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java
index 32cec950a69..6bcba51defd 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java
@@ -961,7 +961,9 @@ public class PulsarClientException extends IOException {
public static Throwable wrap(Throwable t, String msg) {
msg += "\n" + t.getMessage();
// wrap an exception with new message info
- if (t instanceof TopicDoesNotExistException) {
+ if (t instanceof NotFoundException) {
+ return new NotFoundException(msg);
+ } else if (t instanceof TopicDoesNotExistException) {
return new TopicDoesNotExistException(msg);
} else if (t instanceof TimeoutException) {
return new TimeoutException(msg);