[Fix][broker]Fix topic dispatch rate limiter not init on broker-level

[AnonHxy]

Motivation

• Fix dispatch rate limiter doesn't be initialized on broker-level:
• • If DispatchRateLimiter is disabled at first, then we update only broker-level dispatch rate dynamicly, it will not take affect because topic.getDispatchRateLimiter().isPresent() is false, line:2377
    

    pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

    Lines 2370 to 2379 in 8914b84

    	private void updateTopicMessageDispatchRate() {
    	this.pulsar().getExecutor().execute(() -> {
    	// update message-rate for each topic
    	forEachTopic(topic -> {
    	if (topic instanceof AbstractTopic) {
    	((AbstractTopic) topic).updateBrokerDispatchRate();
    	}
    	if (topic.getDispatchRateLimiter().isPresent()) {
    	topic.getDispatchRateLimiter().get().updateDispatchRate();
    	}

Modifications

• Create a method updateDispatchRateLimiter() to initialize and update the dispatch rate limiter

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Added UT TopicDispatchRateLimiterTest

Documentation

Check the box below or label this PR directly.

Need to update docs?

• doc-not-needed

[AnonHxy]

/pulsarbot run-failure-checks

[AnonHxy]

/pulsarbot run-failure-checks

[AnonHxy]

@Technoboy- @codelipenghui PTAL~, thanks

[codelipenghui]

@AnonHxy Could you please help open a PR for branch-2.10 since I'm not able to cherry-pick it directly.

[AnonHxy]

@AnonHxy Could you please help open a PR for branch-2.10 since I'm not able to cherry-pick it directly.

OK, I am working on this. It is better to merge this cherry-pick first #16778. Then it will be easier to resolve the conflict @codelipenghui

[codelipenghui]

@AnonHxy Done