chenhongSZ commented on code in PR #21212:
URL: https://github.com/apache/pulsar/pull/21212#discussion_r1338610291
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java:
##########
@@ -3061,6 +3084,104 @@ public void testGlobalTopicPolicies() throws Exception {
}
+ @Test
+ public void testInitPolicesCacheAndNotifyListeners() throws Exception {
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ // set up policies
+ TopicName topicName = TopicName.get(topic);
+ TopicPolicies localInitPolicy =
TopicPolicies.builder().maxConsumerPerTopic(10).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName,
localInitPolicy).get();
+ TopicPolicies globalInitPolicy =
+
TopicPolicies.builder().maxConsumerPerTopic(20).maxProducerPerTopic(30).isGlobal(true).build();
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(topicName,
globalInitPolicy).get();
+
+ // the policies cache
+ SystemTopicBasedTopicPoliciesService topicPoliciesService
+ = (SystemTopicBasedTopicPoliciesService)
pulsar.getTopicPoliciesService();
+
+ // reload namespace to trigger init polices cache and notify listeners
+ admin.namespaces().unload(myNamespace);
+
assertNull(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)));
+ pulsarClient.newProducer().topic(topic).create().close();
+ Awaitility.await().untilAsserted(
+ () ->
assertEquals(topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace)).isDone()
+ &&
!topicPoliciesService.getPoliciesCacheInit(NamespaceName.get(myNamespace))
+ .isCompletedExceptionally(), true));
+
+ // the final policies take effect in topic
+ HierarchyTopicPolicies hierarchyTopicPolicies =
+
pulsar.getBrokerService().getTopics().get(topic).get().get().getHierarchyTopicPolicies();
+
+
assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxConsumerPerTopic(),
10);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName,
true).getMaxConsumerPerTopic(), 20);
+
assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getGlobalTopicValue(),
20);
+
assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().getLocalTopicValue(),
10);
+ assertEquals(hierarchyTopicPolicies.getMaxConsumerPerTopic().get(),
10);
+
+
assertEquals(topicPoliciesService.getTopicPolicies(topicName).getMaxProducerPerTopic(),
null);
+ assertEquals(topicPoliciesService.getTopicPolicies(topicName,
true).getMaxProducerPerTopic(), 30);
+
assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getGlobalTopicValue(),
30);
+
assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().getLocalTopicValue(),
null);
+ assertEquals(hierarchyTopicPolicies.getMaxProducersPerTopic().get(),
30);
+ }
+
+ @Test
+ public void testInitPolicesCacheAndNotifyListenersAfterCompaction() throws
Exception {
Review Comment:
I think the test case is exactly what you described, please correct me if I
have misunderstood.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
