Use HierarchyTopicPolicies to update topic-level subscribeRate
@@ -3046,7 +3046,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
FutureUtil.waitForAll(consumerCheckFutures).thenRun(() -> {
updatePublishDispatcher();
- initializeTopicSubscribeRateLimiterIfNeeded(Optional.ofNullable(policies));
+ updateSubscribeRateLimiter();
if (this.subscribeRateLimiter.isPresent()) {
subscribeRateLimiter.ifPresent(subscribeRateLimiter ->
subscribeRateLimiter.onSubscribeRateUpdate(getSubscribeRate()));
@@ -3079,23 +3079,6 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
}
- private void initializeTopicSubscribeRateLimiterIfNeeded(Optional<TopicPolicies> policies) {
- if (!policies.isPresent()) {
- return;
- }
- synchronized (subscribeRateLimiter) {
- if (!subscribeRateLimiter.isPresent()
- && policies.get().getSubscribeRate() != null
- && policies.get().getSubscribeRate().subscribeThrottlingRatePerConsumer > 0) {
- this.subscribeRateLimiter = Optional.of(new SubscribeRateLimiter(this));
- } else if (!policies.get().isSubscribeRateSet()
- || policies.get().getSubscribeRate().subscribeThrottlingRatePerConsumer <= 0) {
- subscribeRateLimiter.ifPresent(SubscribeRateLimiter::close);
- this.subscribeRateLimiter = Optional.empty();
- }
- }
- }
-
protected CompletableFuture<Void> initTopicPolicy() {
if (brokerService.pulsar().getConfig().isSystemTopicEnabled()
&& brokerService.pulsar().getConfig().isTopicLevelPoliciesEnabled()) {
