codelipenghui commented on a change in pull request #12340:
URL: https://github.com/apache/pulsar/pull/12340#discussion_r727615842
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -114,17 +114,11 @@ public AbstractTopic(String topic, BrokerService
brokerService) {
this.inactiveTopicPolicies.setMaxInactiveDurationSeconds(brokerService.pulsar().getConfiguration().getBrokerDeleteInactiveTopicsMaxInactiveDurationSeconds());
this.inactiveTopicPolicies.setInactiveTopicDeleteMode(brokerService.pulsar().getConfiguration().getBrokerDeleteInactiveTopicsMode());
this.lastActive = System.nanoTime();
- Policies policies = null;
- 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());
- }
+ Policies policies =
brokerService.pulsar().getConfigurationCache().policiesCache()
Review comment:
Looks we can remove this line.
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -135,12 +129,10 @@ protected boolean isProducersExceeded() {
}
if (maxProducers == null) {
- Policies policies;
- try {
- policies =
brokerService.pulsar().getConfigurationCache().policiesCache()
- .get(AdminResource.path(POLICIES,
TopicName.get(topic).getNamespace()))
- .orElseGet(() -> new Policies());
- } catch (Exception e) {
+ Policies policies =
brokerService.pulsar().getConfigurationCache().policiesCache()
Review comment:
This will change the behavior if the policies cache have not been init
yet or the cache has been invalidated?
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java
##########
@@ -65,13 +65,11 @@ public BacklogQuota getDefaultQuota() {
}
public BacklogQuota getBacklogQuota(String namespace, String policyPath) {
- try {
- return zkCache.get(policyPath)
- .map(p ->
p.backlog_quota_map.getOrDefault(BacklogQuotaType.destination_storage,
defaultQuota))
- .orElse(defaultQuota);
- } catch (Exception e) {
- log.warn("Failed to read policies data, will apply the default
backlog quota: namespace={}", namespace, e);
- return this.defaultQuota;
+ Policies policies = zkCache.getDataIfPresent(policyPath);
Review comment:
If the cache is invalidated, we might use the broker configuration?
--
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]
