Add subscription properties for SubscriptionStats

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 =

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<String, String> 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<String, String> 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<String, String> 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<String, String> pSubProperties = pStats.getSubscriptions().get(pSubName)
+                .getSubscriptionProperties();
+        assertEquals(pSubProperties, pMap);
+
+        PartitionedTopicStats pStatsForPerPartition = admin.topics().getPartitionedStats(partitionedTopic, true);
+        Map<String, String> 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();

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<String, String> getConsumersAfterMarkDeletePosition();
 
+    /** SubscriptionProperties (key/value strings) associated with this subscribe. */
+    Map<String, String> getSubscriptionProperties();
+
     /** The number of non-contiguous deleted messages ranges. */
     int getNonContiguousDeletedMessagesRanges();
 

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<String, String> 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;
     }
 }