[pulsar] branch master updated: fix topicPublishRateLimiter not effective after restart broker (#6893)

[zh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

zhaijia 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 1aaa148  fix topicPublishRateLimiter not effective after restart broker (#6893)
1aaa148 is described below

commit 1aaa148125a851536326d33f59434e3aae456096
Author: liudezhi <33...@users.noreply.github.com>
AuthorDate: Wed May 13 13:07:44 2020 +0800

    fix topicPublishRateLimiter not effective after restart broker (#6893)
    
    Master Issue: #6892
    ## Motivation
    
    when config set-publish-rate on namespaces，then can limit publish rate, but when restart broker the limit has expired.
    
    ## Modifications
    
    modify get the acquisition policy sync way, any namespaces will save a policy on zk.
    ```java
            try {
                policies = brokerService.pulsar().getConfigurationCache().policiesCache()
                        .get(AdminResource.path(POLICIES, TopicName.get(topic).getNamespace()))
                        .orElseGet(() -> new Policies());
            } catch (Exception e) {
                log.warn("[{}] Error getting policies {} and publish throttling will be disabled", topic, e.getMessage());
            }
    ```
---
 .../main/java/org/apache/pulsar/broker/service/AbstractTopic.java   | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
index 1454467..414ebc6 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
@@ -97,10 +97,8 @@ public abstract class AbstractTopic implements Topic {
         Policies policies = null;
         try {
             policies = brokerService.pulsar().getConfigurationCache().policiesCache()
-                    .getDataIfPresent(AdminResource.path(POLICIES, TopicName.get(topic).getNamespace()));
-            if (policies == null) {
-                policies = new Policies();
-            }
+                    .get(AdminResource.path(POLICIES, TopicName.get(topic).getNamespace()))
+                    .orElseGet(() -> new Policies());
         } catch (Exception e) {
             log.warn("[{}] Error getting policies {} and publish throttling will be disabled", topic, e.getMessage());
         }