Sync topicPublishRateLimiter update
@@ -110,6 +110,7 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener<TopicP
protected volatile Boolean isAllowAutoUpdateSchema;
protected volatile PublishRateLimiter topicPublishRateLimiter;
+ private final Object topicPublishRateLimiterLock = new Object();
protected volatile ResourceGroupPublishLimiter resourceGroupPublishLimiter;
@@ -1192,6 +1193,7 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener<TopicP
* update topic publish dispatcher for this topic.
*/
public void updatePublishDispatcher() {
+ synchronized (topicPublishRateLimiterLock) {
PublishRate publishRate = topicPolicies.getPublishRate().get();
if (publishRate.publishThrottlingRateInByte > 0 || publishRate.publishThrottlingRateInMsg > 0) {
log.info("Enabling publish rate limiting {} ", publishRate);
@@ -1220,6 +1222,7 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener<TopicP
enableProducerReadForPublishRateLimiting();
}
}
+ }
// subscriptionTypesEnabled is dynamic and can be updated online.
public void updateBrokerSubscriptionTypesEnabled() {
