topic publish rate limit not take effect

[aloyszhang]

Describe the bug
set-publish-rate does not take effect

To Reproduce

1. set-publish-rate
2. do publish

Expected behavior
producer should not publish messages faster than the publish rate

Screenshots
publish rate limit information

bin/pulsar-admin namespaces get-publish-rate rate/rate 
{
  "publishThrottlingRateInMsg" : 10,
  "publishThrottlingRateInByte" : 10240
}

do publish

bin/pulsar-perf produce -threads 1 -u pulsar://100.76.43.216:6650 -o 10000 -n 1 -b 0 -bm 0 -s 1024 -r 100000 rate/rate/rate

14:43:32.193 [main] INFO  org.apache.pulsar.testclient.PerformanceProducer - Throughput produced:    161.6  msg/s ---      1.4 Mbit/s --- failure      0.0 msg/s --- Latency: mean: 5314.773 ms - med: 5730.591 - 95pct: 8716.543 - 99pct: 8744.191 - 99.9pct: 8744.319 - 99.99pct: 8744.319 - Max: 8744.319
14:43:42.262 [main] INFO  org.apache.pulsar.testclient.PerformanceProducer - Throughput produced:    214.2  msg/s ---      1.8 Mbit/s --- failure      0.0 msg/s --- Latency: mean: 14039.659 ms - med: 13728.319 - 95pct: 18667.391 - 99pct: 18700.927 - 99.9pct: 18701.055 - 99.99pct: 18701.055 - Max: 18701.055

As shown, under the limit of 10 msgs and 10240 bytes, when send message with 1024B bytes, it should not be faster than 10msg/s.
But the tes result is much more than 10msg/s.
Additional context
As my researching , pulsar do topic publish rate limit using two individual thread,

 long topicTickTimeMs = pulsar().getConfiguration().getTopicPublisherThrottlingTickTimeMillis();
        if (topicTickTimeMs > 0) {
            if (this.topicPublishRateLimiterMonitor == null) {
                this.topicPublishRateLimiterMonitor = Executors.newSingleThreadScheduledExecutor(
                        new DefaultThreadFactory("pulsar-topic-publish-rate-limiter-monitor"));
                if (topicTickTimeMs > 0) {
                    topicPublishRateLimiterMonitor.scheduleAtFixedRate(safeRun(() -> checkTopicPublishThrottlingRate()),
                            topicTickTimeMs, topicTickTimeMs, TimeUnit.MILLISECONDS);
                    
                    topicPublishRateLimiterMonitor.scheduleAtFixedRate(safeRun(() -> refreshTopicPublishRate()), 1, 1,
                            TimeUnit.SECONDS);
                }
            }
        } 

one thread schedule task that sums up publish-rate across all cnx on a topic and another schedule task that refreshes rate-limiting bucket.
This means we can only pause publish after send message for topicPublisherThrottlingTickTimeMillis long times, and messages send before topicPublisherThrottlingTickTimeMillis will never trigger rate limit.

In order to get efficient publish rate limit, I think, we should use something like RateLimiter instead of a period task.

[codelipenghui]

@aloyszhang Thanks for the feedback, I think this is an enhancement. I would like to change the tag to enhancement.

[aloyszhang]

@codelipenghui Thanks for your suggestion, LGTM.

[codelipenghui]

@aloyszhang I think we can add options in the broker.conf. So that users can choose the precise rate control or not. If users don't care about the cost of the competition, they can choose the precise rate control.

[aloyszhang]

@codelipenghui OK, I'll add the option.