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:59 UTC
(pulsar) branch branch-3.3 updated (20ac5394a44 -> edaf2550981)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 20ac5394a44 [fix] Prevent IllegalStateException: Field 'message' is not set (#24472)
new 95883cfe066 [fix][test]fix flaky test BrokerServiceAutoTopicCreationTest.testDynamicConfigurationTopicAutoCreationPartitioned (#24505)
new edaf2550981 [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.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit edaf2550981772855bcd213d3d994af19dd40ea2
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 17373e8502b..e2264e91364 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
@@ -1082,8 +1082,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());
}
@@ -1720,7 +1719,7 @@ public class BrokerService implements Closeable {
}
CompletableFuture<Void> maxTopicsCheck = createIfMissing
- ? checkMaxTopicsPerNamespace(topicName, 1)
+ ? checkMaxTopicsPerNamespace(topicName)
: CompletableFuture.completedFuture(null);
CompletableFuture<Void> isTopicAlreadyMigrated = checkTopicAlreadyMigrated(topicName);
@@ -3343,7 +3342,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();
@@ -3750,10 +3749,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 7213abe4557..e378bbf0175 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
@@ -1685,6 +1685,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.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 95883cfe06693b29d501b757502a0f31ed8412d6
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);