[abhiramaab]

Fixes #26959

Motivation

In message dispatch rate limiting (applyDispatchRateLimitsToReadLimits), the rate limiter treated entry read limits as raw message permit counts without accounting for batched messages.

When consumers receive batched messages, a single entry can contain multiple messages, represented by avgMessagesPerEntry. Treating each available message permit as one entry allowed the broker to dispatch more messages than the configured rate limit before throttling kicked in. For example, if each entry contains 10 batched messages and the limit allows 50 messages, reading 50 entries dispatches 500 messages.

Modifications

• Updated applyDispatchRateLimitsToReadLimits in AbstractBaseDispatcher to scale available message permits by avgMessagesPerEntry (retrieved via getConsumers()) when dispatchThrottlingOnBatchMessageEnabled is disabled.
• Preserved forward progress for positive sub-batch budgets by allowing at least one entry when availablePermitsOnMsg > 0 using Math.max(1, availablePermitsOnMsg / avg). Existing post-dispatch accounting charges the actual number of messages delivered.
• Added unit tests in AbstractBaseDispatcherTest to verify entry limit calculations for unbatched messages, batched messages, sub-batch forward progress, and exhausted budget scenarios.

Verifying This Change

• Make sure that the change passes the CI checks.

Added the unit test org.apache.pulsar.broker.service.AbstractBaseDispatcherTest.testApplyRateLimitsToMessagesAndBytesToReadWithBatchedMessages to verify rate limit entry calculations for:

• Unbatched messages (avgMessagesPerEntry = 1).
• Batched messages (avgMessagesPerEntry = 10).

Does This Pull Request Potentially Affect One of the Following Parts?

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[codelipenghui]

The average-messages-per-entry approach looks like a good, low-cost way to reduce dispatch bursts without restructuring the dispatch path. I would suggest keeping the change localized, with the following adjustments:

1. Decouple rate-limit conversion from preciseDispatcherFlowControl. Consumer flow control and dispatch rate limiting are separate constraints. Currently, both callers pass an average of 1 when precise flow control is disabled (the default), leaving the original message/entry mismatch unchanged in that configuration.
2. Match the unit used for quota accounting. When dispatchThrottlingOnBatchMessageEnabled is enabled, acquirePermitsForDeliveredMessages charges entries rather than individual messages. In that mode, the available permits should cap entries directly. Divide by the historical average only when quota accounting counts individual messages.
3. Preserve forward progress for positive, sub-batch budgets. Integer division returns zero when the remaining message permits are smaller than the estimated batch size, taking the reschedule-without-reading path. A practical approach is to allow at least one entry when permits are positive, then charge the actual dispatched count through the existing accounting path. Exhausted permits should still defer the read. This avoids making progress depend on accumulating enough permits for a historical estimate.

In short: derive an entry budget in the same unit as quota accounting, independently of the consumer flow-control setting, and retain a one-entry progress allowance for positive budgets. This reuses the existing average, read scheduling, and delivery accounting with minimal additional work.

The tradeoff should be explicit: a historical average reduces expected overshoot but cannot guarantee a strict message-rate cap, especially at startup or when batch sizes change. Allowing one entry can also exceed the remaining message budget because an entry may contain an indivisible batch. Strict admission control would require a broader design rather than just this read-size estimate.

This suggestion is based on source inspection of head 49ed5f922c9de5a5a9aaf536aa03676b8fc61062; I have not run builds or tests.

[abhiramaab]

Thanks @codelipenghui! In the latest push (b839ffc), I’ve kept the changes strictly within AbstractBaseDispatcher and accounted for dispatchThrottlingOnBatchMessageEnabled (Points 1 and 2). I’ll add the 1-entry progress allowance for positive sub-batch budgets (Point 3) shortly.

[abhiramaab]

Hi @codelipenghui @lhotari, thanks for the review and guidance!

I have updated the PR to address all the review points:

1. Self-contained and decoupled from consumer flow control: Reverted all changes to PersistentDispatcherMultipleConsumers and PersistentDispatcherSingleActiveConsumer. The logic is now entirely contained within AbstractBaseDispatcher, resolving avgMessagesPerEntry via getConsumers().

2. Quota unit alignment: Division by avgMessagesPerEntry is performed only when dispatchThrottlingOnBatchMessageEnabled is disabled (the default). When batch-level throttling is enabled, available permits directly cap the number of entries without division.

3. Forward progress for sub-batch budgets: When availablePermitsOnMsg > 0, at least one entry is permitted using Math.max(1, availablePermitsOnMsg / avg). The existing post-dispatch accounting (acquirePermitsForDeliveredMessages) then charges the actual number of messages delivered, ensuring reads continue to make progress. When permits are exhausted, reads are deferred as usual.

Added unit test coverage in AbstractBaseDispatcherTest for unbatched messages, batched messages, sub-batch forward progress, and exhausted budget scenarios.

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[codelipenghui]

Reviewed head 99f6428f7253286d45479db5ed21e7eb6f84d260.

The previous Failover standby-estimate issue is resolved: all three cases in the retained reproducer now pass. The relevant dispatcher and consumer tests also pass (33 test cases in total, including that reproducer).

One P2 regression remains in the new cold-start fallback: a Reader using keyHashRange can remain limited to one entry per read throughout an excluded initial backlog, because range filtering removes entries before the metadata-based estimate is initialized. Please address the inline comment and add coverage for this path.

A focused local test using real consumer statistics, range selection, and dispatcher read/continuation logic reproduces read requests [1, 1, 1, 1] with the average still zero. Storage callbacks and transport were mocked; no end-to-end throughput benchmark was run.

[codelipenghui]

P2 — Bootstrap the estimate for hash-range-skipped entries too

The one-entry cold-start fallback can persist throughout the initial backlog of a Reader using keyHashRange. In the single-active read-completion path, out-of-range entries are released and removed before filterEntriesForConsumer runs. The new metadata-based initialization therefore never sees those entries, and the empty Consumer.sendMessages call does not initialize the average either.

A focused local test completing three out-of-range reads, each containing a ten-message entry, observes read requests [1, 1, 1, 1] and observedAverage=0, despite 50 available message permits and sufficient consumer credit. A long excluded prefix consequently requires one read operation per entry until a matching message is encountered, rather than only one initial probe.

Please initialize the unknown estimate from metadata before range filtering, or use a dispatcher-level observation covering all read entries. Keep initialization restricted to unknown statistics, and add a regression test with an out-of-range initial backlog.