You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/07/25 17:27:51 UTC
(pulsar) branch branch-3.0 updated (9291b715e1f -> 3bd27943848)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 9291b715e1f [fix] Prevent IllegalStateException: Field 'message' is not set (#24472)
new 49eba28a9cb [fix][test]fix flaky test BrokerServiceAutoTopicCreationTest.testDynamicConfigurationTopicAutoCreationPartitioned (#24505)
new 3bd27943848 [fix][broker] Fix maxTopicsPerNamespace might report a false failure (#24560)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../pulsar/broker/service/BrokerService.java | 25 +++++++++++++++++-----
.../apache/pulsar/broker/admin/NamespacesTest.java | 13 +++++++++++
.../BrokerServiceAutoTopicCreationTest.java | 7 ++++++
3 files changed, 40 insertions(+), 5 deletions(-)
(pulsar) 02/02: [fix][broker] Fix maxTopicsPerNamespace might report a false failure (#24560)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 3bd279438484642f3a3ce7bd1ab936e57294e1f1
Author: Yunze Xu <xy...@163.com>
AuthorDate: Fri Jul 25 21:05:57 2025 +0800
[fix][broker] Fix maxTopicsPerNamespace might report a false failure (#24560)
(cherry picked from commit 1c5c765d00ab98fc250cb4a1062e4dc95f83cfb5)
---
.../pulsar/broker/service/BrokerService.java | 25 +++++++++++++++++-----
.../apache/pulsar/broker/admin/NamespacesTest.java | 13 +++++++++++
2 files changed, 33 insertions(+), 5 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index 685226779c3..2c11aa0fd61 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -1102,8 +1102,7 @@ public class BrokerService implements Closeable {
return FutureUtil.failedFuture(new NotAllowedException(
"Broker is unable to load persistent topic"));
}
- return pulsar.getPulsarResources().getTopicResources().persistentTopicExists(topicName)
- .thenCompose(exists -> {
+ return checkNonPartitionedTopicExists(topicName).thenCompose(exists -> {
if (!exists && !createIfMissing) {
return CompletableFuture.completedFuture(Optional.empty());
}
@@ -1751,7 +1750,7 @@ public class BrokerService implements Closeable {
}
CompletableFuture<Void> maxTopicsCheck = createIfMissing
- ? checkMaxTopicsPerNamespace(topicName, 1)
+ ? checkMaxTopicsPerNamespace(topicName)
: CompletableFuture.completedFuture(null);
maxTopicsCheck.thenCompose(__ ->
@@ -3360,7 +3359,7 @@ public class BrokerService implements Closeable {
PartitionedTopicMetadata configMetadata = new PartitionedTopicMetadata(defaultNumPartitions);
- return checkMaxTopicsPerNamespace(topicName, defaultNumPartitions)
+ return checkMaxTopicsPerNamespace(topicName, defaultNumPartitions, true)
.thenCompose(__ -> {
PartitionedTopicResources partitionResources = pulsar.getPulsarResources().getNamespaceResources()
.getPartitionedTopicResources();
@@ -3764,10 +3763,26 @@ public class BrokerService implements Closeable {
});
}
- private CompletableFuture<Void> checkMaxTopicsPerNamespace(TopicName topicName, int numPartitions) {
+ private CompletableFuture<Boolean> checkNonPartitionedTopicExists(TopicName topicName) {
+ return pulsar.getPulsarResources().getTopicResources().persistentTopicExists(topicName);
+ }
+
+ private CompletableFuture<Void> checkMaxTopicsPerNamespace(TopicName topicName) {
if (isSystemTopic(topicName)) {
return CompletableFuture.completedFuture(null);
}
+ return checkNonPartitionedTopicExists(topicName).thenCompose(exists -> exists
+ ? CompletableFuture.completedFuture(null)
+ : checkMaxTopicsPerNamespace(topicName, 1, false));
+ }
+
+ private CompletableFuture<Void> checkMaxTopicsPerNamespace(TopicName topicName, int numPartitions,
+ boolean validateSystemTopic) {
+ if (validateSystemTopic) {
+ if (isSystemTopic(topicName)) {
+ return CompletableFuture.completedFuture(null);
+ }
+ }
return pulsar.getPulsarResources().getNamespaceResources()
.getPoliciesAsync(topicName.getNamespaceObject())
.thenCompose(optPolicies -> {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
index 53b170d52e6..ff1cf035ec6 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
@@ -1682,6 +1682,19 @@ public class NamespacesTest extends MockedPulsarServiceBaseTest {
assertEquals(e.getHttpError(), "Exceed maximum number of topics in namespace.");
}
+ // Unload the topic to remove this non-partitioned topic from `BrokerService#topics`, so when creating the
+ // consumer, it will trigger the topic loading.
+ admin.topics().unload(topic + "4");
+ final var consumers = new ArrayList<Consumer<byte[]>>();
+ for (int i = 1; i <= 4; i++) {
+ consumers.add(pulsarClient.newConsumer().topic(topic + i).subscriptionName("sub").subscribe());
+ }
+ final var topics = admin.topics().getList(namespace);
+ assertEquals(topics.size(), 10);
+ for (final var consumer : consumers) {
+ consumer.close();
+ }
+
// remove namespace policy limit, use broker configuration instead.
admin.namespaces().removeMaxTopicsPerNamespace(namespace);
admin.topics().createPartitionedTopic(topic + "6", 4);
(pulsar) 01/02: [fix][test]fix flaky test BrokerServiceAutoTopicCreationTest.testDynamicConfigurationTopicAutoCreationPartitioned (#24505)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 49eba28a9cb2f7186e1ddf8062c5004fbc7d090b
Author: Ruimin MA <ma...@gmail.com>
AuthorDate: Fri Jul 11 16:54:47 2025 +0800
[fix][test]fix flaky test BrokerServiceAutoTopicCreationTest.testDynamicConfigurationTopicAutoCreationPartitioned (#24505)
(cherry picked from commit e4de9c08f58d9d58fe80b3be2f0894eef7211ef8)
---
.../pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java | 7 +++++++
1 file changed, 7 insertions(+)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java
index ea5365bcf4b..eb0b5b17e65 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java
@@ -482,6 +482,13 @@ public class BrokerServiceAutoTopicCreationTest extends BrokerTestBase{
admin.brokers().updateDynamicConfiguration("maxNumPartitionsPerPartitionedTopic", "6");
admin.brokers().updateDynamicConfiguration("allowAutoTopicCreationType", "partitioned");
admin.brokers().updateDynamicConfiguration("defaultNumPartitions", "4");
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> {
+ assertTrue(pulsar.getConfiguration().isAllowAutoTopicCreation());
+ assertEquals(pulsar.getConfiguration().getMaxNumPartitionsPerPartitionedTopic(), 6);
+ assertEquals(pulsar.getConfiguration().getAllowAutoTopicCreationType(), TopicType.PARTITIONED);
+ assertEquals(pulsar.getConfiguration().getDefaultNumPartitions(), 4);
+
+ });
Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create();
List<String> topics = admin.topics().getList(namespaceName);
List<String> partitionedTopicList = admin.topics().getPartitionedTopicList(namespaceName);