diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java index fbe11fbdf0dc1..2c23bee256193 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java @@ -1078,6 +1078,7 @@ public SubscriptionStatsImpl getStats(Boolean getPreciseBacklog, boolean subscri subStats.msgRateExpired = expiryMonitor.getMessageExpiryRate(); subStats.totalMsgExpired = expiryMonitor.getTotalMessageExpired(); subStats.isReplicated = isReplicated(); + subStats.subscriptionProperties = subscriptionProperties; subStats.isDurable = cursor.isDurable(); if (getType() == SubType.Key_Shared && dispatcher instanceof PersistentStickyKeyDispatcherMultipleConsumers) { PersistentStickyKeyDispatcherMultipleConsumers keySharedDispatcher = diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/CreateSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/CreateSubscriptionTest.java index 652646d0377ae..7a537e369d6fd 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/CreateSubscriptionTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/CreateSubscriptionTest.java @@ -22,6 +22,7 @@ import static org.testng.Assert.assertNotNull; import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; + import com.google.common.collect.Lists; import java.nio.charset.StandardCharsets; import java.util.HashMap; @@ -39,6 +40,8 @@ import org.apache.pulsar.client.api.ProducerConsumerBase; import org.apache.pulsar.client.impl.MessageIdImpl; import org.apache.pulsar.common.naming.TopicName; +import org.apache.pulsar.common.policies.data.PartitionedTopicStats; +import org.apache.pulsar.common.policies.data.TopicStats; import org.awaitility.Awaitility; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; @@ -140,6 +143,42 @@ public void createSubscriptionOnPartitionedTopicWithPartialFailure() throws Exce } } + @Test + public void testSubscriptionPropertiesStats() throws Exception { + // test non-partitioned topic + final String topic = "persistent://my-property/my-ns/topic" + UUID.randomUUID(); + admin.topics().createNonPartitionedTopic(topic); + Map map = new HashMap<>(); + map.put("test-topic", "tag1"); + String subName = "my-sub"; + pulsarClient.newConsumer().topic(topic).receiverQueueSize(1) + .subscriptionProperties(map).subscriptionName(subName).subscribe(); + TopicStats stats = admin.topics().getStats(topic); + Map subProperties = stats.getSubscriptions().get(subName).getSubscriptionProperties(); + assertEquals(subProperties, map); + + // test partitioned-topic + final String partitionedTopic = "persistent://my-property/my-ns/topic" + UUID.randomUUID(); + admin.topics().createPartitionedTopic(partitionedTopic, 10); + Map pMap = new HashMap<>(); + pMap.put("topic1", "tag1"); + pMap.put("topic2", "tag2"); + pMap.put("topic3", "tag3"); + String pSubName = "my-sub-1"; + pulsarClient.newConsumer().topic(partitionedTopic).receiverQueueSize(1) + .subscriptionProperties(pMap).subscriptionName(pSubName).subscribe(); + + PartitionedTopicStats pStats = admin.topics().getPartitionedStats(partitionedTopic, false); + Map pSubProperties = pStats.getSubscriptions().get(pSubName) + .getSubscriptionProperties(); + assertEquals(pSubProperties, pMap); + + PartitionedTopicStats pStatsForPerPartition = admin.topics().getPartitionedStats(partitionedTopic, true); + Map pSubPropForPerPartition = pStatsForPerPartition.getSubscriptions().get(pSubName) + .getSubscriptionProperties(); + assertEquals(pSubPropForPerPartition, pMap); + } + @Test public void addSubscriptionPropertiesTest() throws Exception { String topic = "persistent://my-property/my-ns/topic" + UUID.randomUUID(); diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java index 2ce38aafb34d4..26d940938df9e 100644 --- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java +++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java @@ -106,6 +106,9 @@ public interface SubscriptionStats { /** This is for Key_Shared subscription to get the recentJoinedConsumers in the Key_Shared subscription. */ Map getConsumersAfterMarkDeletePosition(); + /** SubscriptionProperties (key/value strings) associated with this subscribe. */ + Map getSubscriptionProperties(); + /** The number of non-contiguous deleted messages ranges. */ int getNonContiguousDeletedMessagesRanges(); diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java index 78781ac32b46b..98959648bb155 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.common.policies.data.stats; +import java.util.HashMap; import lombok.Data; import org.apache.pulsar.common.policies.data.SubscriptionStats; import java.util.ArrayList; @@ -118,9 +119,13 @@ public class SubscriptionStatsImpl implements SubscriptionStats { /** The serialized size of non-contiguous deleted messages ranges. */ public int nonContiguousDeletedMessagesRangesSerializedSize; + /** SubscriptionProperties (key/value strings) associated with this subscribe. */ + public Map subscriptionProperties; + public SubscriptionStatsImpl() { this.consumers = new ArrayList<>(); this.consumersAfterMarkDeletePosition = new LinkedHashMap<>(); + this.subscriptionProperties = new HashMap<>(); } public void reset() { @@ -141,6 +146,7 @@ public void reset() { consumersAfterMarkDeletePosition.clear(); nonContiguousDeletedMessagesRanges = 0; nonContiguousDeletedMessagesRangesSerializedSize = 0; + subscriptionProperties.clear(); } // if the stats are added for the 1st time, we will need to make a copy of these stats and add it to the current @@ -175,6 +181,7 @@ public SubscriptionStatsImpl add(SubscriptionStatsImpl stats) { this.consumersAfterMarkDeletePosition.putAll(stats.consumersAfterMarkDeletePosition); this.nonContiguousDeletedMessagesRanges += stats.nonContiguousDeletedMessagesRanges; this.nonContiguousDeletedMessagesRangesSerializedSize += stats.nonContiguousDeletedMessagesRangesSerializedSize; + this.subscriptionProperties.putAll(stats.subscriptionProperties); return this; } }