Tidy dispatchRateLimiter init
@@ -321,7 +321,8 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
isEncryptionRequired = false;
updatePublishDispatcher();
updateResourceGroupLimiter(optPolicies);
- initializeRateLimiterIfNeeded();
+ initializeDispatchRateLimiterIfNeeded();
+ updateSubscribeRateLimiter();
return;
}
@@ -329,7 +330,9 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
this.updateTopicPolicyByNamespacePolicy(policies);
- initializeRateLimiterIfNeeded();
+ initializeDispatchRateLimiterIfNeeded();
+
+ updateSubscribeRateLimiter();
updatePublishDispatcher();
@@ -375,7 +378,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
}
- private void initializeRateLimiterIfNeeded() {
+ private void initializeDispatchRateLimiterIfNeeded() {
synchronized (dispatchRateLimiterLock) {
// dispatch rate limiter for topic
if (!dispatchRateLimiter.isPresent()
@@ -383,8 +386,6 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
this.dispatchRateLimiter = Optional.of(new DispatchRateLimiter(this, Type.TOPIC));
}
- updateSubscribeRateLimiter();
-
// dispatch rate limiter for each subscription
subscriptions.forEach((name, subscription) -> {
Dispatcher dispatcher = subscription.getDispatcher();
@@ -2398,7 +2399,9 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
schemaValidationEnforced = data.schema_validation_enforced;
- initializeRateLimiterIfNeeded();
+ initializeDispatchRateLimiterIfNeeded();
+
+ updateSubscribeRateLimiter();
updatePublishDispatcher();
