[improve][broker] Make dispatch rate limiter more precise

[poorbarcode]

Motivation

The Dispatch rate limiter serves two purposes:

1. Limit your consumption speed
2. When there are many backlogs, bk can be protected to avoid excessive I/O

But the current design has the following problems:

available permits indicates how many messages the client needs, and then uses the rate limiter to determine how many entries to read. If batch production is not enabled, it is working. When batch production is enabled, the rate limiter counts the number of entries as the number of messages, resulting in excessive entries to be read.

In PR #6719, preciseDispatcherFlowControl was added to optimize the above problem: If enabled preciseDispatcherFlowControl, we call availablePermits / avgMessagesPerEntry to calculate how many entries to read, but this did not solve the problem because that:

• When you compute avgMessagesPerEntry, randomly take a consumer from the subscription, and get attribute stat.avgMessagesPerEntry, it is possible to null the property. We should look for the consumer that meets condition stat.avgMessagesPerEntry != null.
• The rate limiter still counts the number of entries as the number of messages.

Modifications

• First, calculate how many messages to read by rate limiter, and then calculate how many entries to read by avgMessagesPerEntry
• When calculating the "avgMessagesPerEntry," look up the attribute avgMessagesPerEntry from the consumers under subscription, if all consumers avgMessagesPerEntry is zero, look up under the topic.
• Abstracting the method calculateToRead of Single consumer dispatcher and Multi consumer dispatcher into the parent class AbstractDispatcher, this will save codes.

Follow-up PR

After this PR, there are still such scenarios: failing to get avgMessagesPerEntry resulting in reading excessive messages:

1. Create a new topic and a new subscription, since no data has been consumed yet, the consumer does not know how many messages per entry.
2. Use the created topic, only one subscription, and use the failover mode for consumption. Every time a new consumer becomes an active consumer, can not get avgMessagesPerEntry.

The PR below will fix it.

• https://github.com/apache/pulsar/pull/18581

Documentation

• [ ] doc
• [ ] doc-required
• [x] doc-not-needed
• [ ] doc-complete

Matching PR in forked repository

PR in forked repository:

• https://github.com/poorbarcode/pulsar/pull/42

[rdhabalia]

first of all adding precise terminology in dispatch path is not correct and really misguiding. also have you considered performance impact due to this change? we should not take any performance impact because of this requirement.

[poorbarcode]

Hi @rdhabalia

also have you considered performance impact due to this change?

This patch has almost no performance penalty, except that the first time a consumer is created it retrieves a property of another consumer directly from memory

[github-actions[bot]]

The pr had no activity for 30 days, mark with Stale label.

[lhotari]

please rebase