[fix][broker] Fix UnsupportedOperationException while setting subscription level dispatch rate policy (apache#24048)

(cherry picked from commit 9f38a5c)
(cherry picked from commit 98c205b)

Author: dao-jun

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java

@@ -86,9 +86,11 @@
 import org.apache.pulsar.common.policies.data.PublishRate;
 import org.apache.pulsar.common.policies.data.RetentionPolicies;
 import org.apache.pulsar.common.policies.data.SubscribeRate;
+import org.apache.pulsar.common.policies.data.SubscriptionPolicies;
 import org.apache.pulsar.common.policies.data.TenantInfoImpl;
 import org.apache.pulsar.common.policies.data.TopicPolicies;
 import org.apache.pulsar.common.policies.data.TopicStats;
+import org.apache.pulsar.common.policies.data.impl.DispatchRateImpl;
 import org.assertj.core.api.Assertions;
 import org.awaitility.Awaitility;
 import org.mockito.Mockito;
@@ -3215,7 +3217,7 @@ public void testDelayedDeliveryPolicy() throws Exception {
 
         admin.topics().delete(topic, true);
     }
-    
+
     @Test
     public void testUpdateRetentionWithPartialFailure() throws Exception {
         String tpName = BrokerTestUtil.newUniqueName("persistent://" + myNamespace + "/tp");
@@ -3259,4 +3261,40 @@ public void testUpdateRetentionWithPartialFailure() throws Exception {
         admin.namespaces().removeRetention(myNamespace);
         admin.topics().delete(tpName, false);
     }
+
+    @Test
+    public void testTopicPoliciesGetSubscriptionPolicies() throws Exception {
+        TopicPolicies topicPolicies = TopicPolicies.builder()
+                .maxProducerPerTopic(10).subscriptionPolicies(null).build();
+        Assert.assertNotNull(topicPolicies.getSubscriptionPolicies());
+        Assert.assertEquals(topicPolicies.getMaxProducerPerTopic(), 10);
+        Assert.assertTrue(topicPolicies.getSubscriptionPolicies().isEmpty());
+        topicPolicies.getSubscriptionPolicies().computeIfAbsent("sub", k ->
+                new SubscriptionPolicies()).setDispatchRate(new DispatchRateImpl());
+        Assert.assertEquals(topicPolicies.getSubscriptionPolicies().get("sub").getDispatchRate()
+                .getDispatchThrottlingRateInByte(), 0);
+    }
+
+    @Test
+    public void testSetSubRateWithNoSub() throws Exception {
+        String topic = "persistent://" + myNamespace + "/testSetSubRateWithNoSub";
+        admin.topics().createNonPartitionedTopic(topic);
+        admin.topicPolicies().setSubscriptionDispatchRate(topic, DispatchRate.builder()
+                .dispatchThrottlingRateInMsg(10)
+                .dispatchThrottlingRateInByte(10)
+                .ratePeriodInSecond(10)
+                .build());
+    }
+
+    @Test
+    public void testSetSubRateWithSub() throws Exception {
+        String topic = "persistent://" + myNamespace + "/testSetSubRateWithSub";
+        admin.topics().createNonPartitionedTopic(topic);
+        admin.topics().createSubscription(topic, "sub1", MessageId.earliest);
+        admin.topicPolicies().setSubscriptionDispatchRate(topic, "sub1", DispatchRate.builder()
+                .dispatchThrottlingRateInMsg(10)
+                .dispatchThrottlingRateInByte(10)
+                .ratePeriodInSecond(10)
+                .build());
+    }
 }

pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java

@@ -19,12 +19,13 @@
 package org.apache.pulsar.common.policies.data;
 
 import com.google.common.collect.Sets;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
@@ -42,6 +43,7 @@
 @Builder
 @NoArgsConstructor
 @AllArgsConstructor
+@SuppressFBWarnings(value = "RCN_REDUNDANT_NULLCHECK_OF_NONNULL_VALUE")
 public class TopicPolicies {
 
     @Builder.Default
@@ -193,6 +195,9 @@ public Set<String> getReplicationClustersSet() {
     }
 
     public Map<String, SubscriptionPolicies> getSubscriptionPolicies() {
-        return subscriptionPolicies == null ? Collections.emptyMap() : subscriptionPolicies;
+        if (subscriptionPolicies == null) {
+            subscriptionPolicies = new ConcurrentHashMap<>();
+        }
+        return subscriptionPolicies;
     }
 }