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/11/19 08:51:11 UTC
(pulsar) branch master updated: [fix][broker] Fix creation of replicated subscriptions for partitioned topics (#24997)
This is an automated email from the ASF dual-hosted git repository.
technoboy 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 4566edddd62 [fix][broker] Fix creation of replicated subscriptions for partitioned topics (#24997)
4566edddd62 is described below
commit 4566edddd62567decf031e8006edb025bfc60341
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Nov 19 10:51:05 2025 +0200
[fix][broker] Fix creation of replicated subscriptions for partitioned topics (#24997)
---
.../broker/admin/impl/PersistentTopicsBase.java | 2 +-
.../apache/pulsar/broker/admin/AdminApi2Test.java | 21 +++++++++++++++++++++
2 files changed, 22 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index 5a365b13995..85387ccc267 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -2220,7 +2220,7 @@ public class PersistentTopicsBase extends AdminResource {
try {
pulsar().getAdminClient().topics()
.createSubscriptionAsync(topicNamePartition.toString(),
- subscriptionName, targetMessageId, false, properties)
+ subscriptionName, targetMessageId, replicated, properties)
.handle((r, ex) -> {
if (ex != null) {
// fail the operation on unknown exception or
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
index f889f1375a2..b358c672c6c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
@@ -2543,6 +2543,27 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
}
}
+ @Test(dataProvider = "trueFalse")
+ public void testCreateReplicatedSubscriptionForPartitionedTopic(boolean replicated) throws Exception {
+ final String topic = newUniqueName("persistent://" + defaultNamespace + "/topic");
+ admin.topics().createPartitionedTopic(topic, 10);
+ admin.topics().createSubscription(topic, "sub", MessageId.earliest, replicated);
+ for (int i = 0; i < 10; i++) {
+ String individualPartition = TopicName.get(topic).getPartition(i).toString();
+ TopicStats stats = admin.topics().getStats(individualPartition);
+ assertEquals(stats.getSubscriptions().get("sub").isReplicated(), replicated);
+ }
+ }
+
+ @Test(dataProvider = "trueFalse")
+ public void testCreateReplicatedSubscriptionForNonPartitionedTopic(boolean replicated) throws Exception {
+ final String topic = newUniqueName("persistent://" + defaultNamespace + "/topic");
+ admin.topics().createNonPartitionedTopic(topic);
+ admin.topics().createSubscription(topic, "sub", MessageId.earliest, replicated);
+ TopicStats stats = admin.topics().getStats(topic);
+ assertEquals(stats.getSubscriptions().get("sub").isReplicated(), replicated);
+ }
+
@Test
public void testMaxSubscriptionsPerTopic() throws Exception {
restartClusterAfterTest();