[fix][broker] Use dispatchRateLimiterLock to update dispatchRateLimiter.

[Technoboy-]

Motivation

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

Lines 377 to 399 in 58c82a7

	private void initializeRateLimiterIfNeeded() {
	synchronized (dispatchRateLimiter) {
	// dispatch rate limiter for topic
	if (!dispatchRateLimiter.isPresent()
	&& DispatchRateLimiter.isDispatchRateEnabled(topicPolicies.getDispatchRate().get())) {
	this.dispatchRateLimiter = Optional.of(new DispatchRateLimiter(this, Type.TOPIC));
	}
	updateSubscribeRateLimiter();
	// dispatch rate limiter for each subscription
	subscriptions.forEach((name, subscription) -> {
	Dispatcher dispatcher = subscription.getDispatcher();
	if (dispatcher != null) {
	dispatcher.initializeDispatchRateLimiterIfNeeded();
	}
	});
	// dispatch rate limiter for each replicator
	replicators.forEach((name, replicator) ->
	replicator.initializeDispatchRateLimiterIfNeeded());
	}
	}

The object lock may change when execute at line-382, and cause the lock to become useless.

So use dispatchRateLimiterLock to synchronize.

Documentation

• no-need-doc
  (Please explain why)