[do not merge] [improve] [broker] Make the dispatch rate limit more precise when subscription is created

[poorbarcode]

This PR should merge after #18553.

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.

After PR #18553, 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.

Modifications

• Record avgMessagesPerEntry on the topic, used if avgMessagesPerEntry could not be found in subscriptions.
• If avgMessagesPerEntry is not initialized (it could also be called the first reading), at most one message can be read. Just like the original design: only one message is read when ledger.getStats().getEntrySizeAverage() is not initialized.
  • Why not use mechanism ledger.getStats().getEntrySizeAverage() instead of creating another attribute topic.avgMessagesPerEntry? Because ledger stat is not set immediately after produce messages, need to wait for the scheduled task(default interval 60 seconds) in ManagedLedgerFactoryImpl.
  • See the code below (line-3396):

pulsar/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

Lines 3391 to 3400 in a165bda

	private int applyMaxSizeCap(int maxEntries, long maxSizeBytes) {
	if (maxSizeBytes == NO_MAX_SIZE_LIMIT) {
	return maxEntries;
	}
	double avgEntrySize = ledger.getStats().getEntrySizeAverage();
	if (!Double.isFinite(avgEntrySize)) {
	// We don't have yet any stats on the topic entries. Let's try to use the cursor avg size stats
	avgEntrySize = (double) entriesReadSize / (double) entriesReadCount;
	}

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository:

• [ci] [improve] [broker] Make the dispatch rate limit more precise when subscription is created poorbarcode/pulsar#43

[github-actions]

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

[lhotari]

please rebase