Use dispatchRateLimiterLock to update dispatchRateLimiter.
### Motivation
The object lock may change when execute at line-382, and cause the lock to become useless.
So use `dispatchRateLimiterLock` to synchronize.
@@ -187,6 +187,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
public boolean msgChunkPublished;
private Optional<DispatchRateLimiter> dispatchRateLimiter = Optional.empty();
+ private final Object dispatchRateLimiterLock = new Object();
private Optional<SubscribeRateLimiter> subscribeRateLimiter = Optional.empty();
private final long backloggedCursorThresholdEntries;
public static final int MESSAGE_RATE_BACKOFF_MS = 1000;
@@ -375,7 +376,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
private void initializeRateLimiterIfNeeded() {
- synchronized (dispatchRateLimiter) {
+ synchronized (dispatchRateLimiterLock) {
// dispatch rate limiter for topic
if (!dispatchRateLimiter.isPresent()
&& DispatchRateLimiter.isDispatchRateEnabled(topicPolicies.getDispatchRate().get())) {
@@ -3066,7 +3067,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
private void initializeTopicDispatchRateLimiterIfNeeded(TopicPolicies policies) {
- synchronized (dispatchRateLimiter) {
+ synchronized (dispatchRateLimiterLock) {
if (!dispatchRateLimiter.isPresent() && policies.getDispatchRate() != null) {
this.dispatchRateLimiter = Optional.of(new DispatchRateLimiter(this, Type.TOPIC));
}
