[poorbarcode]

Motivation

The broker currently installs PulsarFlowControlHandler between the length-field decoder and ServerCnx. When a publish limit disables auto-read while handling one frame, the decoder can still emit the remaining complete frames from the same input. The flow-control handler queues those frames until reading resumes.

That queue improves per-message pacing, but it does not undo the socket read or release the received data. A queued frame retains its ByteBuf, and a retained slice can keep a much larger underlying allocation alive. Moving a publish from the business pipeline into this queue therefore does not reduce the memory occupied by that input allocation. The queued request has also not reached the pending-publish counters in ServerCnx, so those counters do not include the queued work.

This becomes particularly problematic when backpressure accounts for allocations retained by in-progress requests, as discussed in #27110. For example, frames A and B share an input allocation; handling A triggers topic throttling, so B is queued. Allocation-based memory throttling then also pauses the connection. Even after A completes and the topic throttle clears, B retains the allocation while the memory throttle prevents B from reaching the code that would process and release it. Removing the intermediate frame queue eliminates this particular circular dependency. This PR is based directly on master and does not include the memory-accounting changes from #27110.

The intended backpressure boundary is subsequent automatic socket reads. Complete frames decoded from data already read should reach ServerCnx, where normal processing can finish and release their references. Existing throttle reasons still disable auto-read, and reading resumes only after all active reasons clear. The decoder still retains incomplete frames and preserves Netty's explicit read paths for making decoding progress.

This deliberately changes the pacing restored by #26013. That PR addressed a real loss of per-message responsiveness after a Netty upgrade; this change chooses to remove the decoded-frame queue rather than preserve that responsiveness. Publish-rate limits remain reactive: messages and bytes are charged to the token buckets, and overconsumption creates debt that delays subsequent reads. A short burst may complete above the configured rate and be followed by a longer pause. Pending-request, pending-byte, and outbound-buffer backpressure also allow already-read complete frames to finish entering the business pipeline.

The overshoot is determined by buffered input and concurrency, not by a fixed number of seconds or a guaranteed 2 MiB allocation. The removed queue did limit downstream work: allowing more frames through can create additional ManagedLedger/BookKeeper operations, callback objects, and response data even when the input allocation is unchanged. This PR does not claim a measured reduction in peak total memory, a throughput improvement, or a hard memory bound. It makes the ownership and backpressure tradeoff explicit and removes a queue that retains input without advancing its processing.

Modifications

Remove the decoded-frame flow-control queue from the broker's inbound pipeline. Let complete frames already decoded reach ServerCnx while existing throttles pause subsequent automatic socket reads. This allows those frames to make progress toward releasing their buffers and removes the queue's interaction with allocation-lifetime backpressure.

• Remove PulsarFlowControlHandler from PulsarChannelInitializer, and delete the unused implementation and its queue-specific tests.
• Document the read-batch overshoot in the initializer, throttle tracker, and publish-rate/pending-publish configuration descriptions.
• Update PersistentTopicE2ETest to use the existing newServerCnx extension point instead of inserting a handler relative to the removed flowController name.
• Add pipeline tests covering every current throttle reason, two frames sharing a pooled allocation, partial-frame completion and close, and actual socket reads remaining paused until all throttle reasons clear. The tests use the production initializer and a mock business handler so the input batch and retained-buffer lifecycle are deterministic.
• Preserve the original average-rate and adjacent-window assertions in PublishRateLimiterOverconsumingTest, with five independent producers, five shared-subscription consumers, and six-message batches. Use 16 KiB message bodies and a test-only fixed 16 KiB receive buffer to bound the number of complete messages delivered by one socket read; production receive-buffer settings are unchanged.
• Add a separate PublishRateLimiterTokenDebtTest using five independent producer connections, 600-message batches, and a controlled clock. It verifies that completed bursts are fully charged, later publishes remain blocked while debt remains, and scheduled unthrottling resumes publishing after replenishment; all 3,005 published values are consumed without duplicates.

Validation:

• The seven parameterized current-read cases reproduced the old queueing behavior before removal: only one of two complete frames reached ServerCnx. All seven pass after removal.
• ./gradlew quickCheck passed, including main/test Checkstyle, Spotless license headers, and RAT.
• Selected pipeline and broker throttling tests passed with retries disabled: PulsarChannelInitializerTest, PublishRateLimiterTest, PublishRateLimiterDisableTest, TopicPublishRateThrottleTest, MessagePublishThrottlingTest, TopicPublishThrottlingInitTest, ResourceGroupRateLimiterTest, and PersistentTopicE2ETest.testBrokerConnectionStats (explicitly including its otherwise-excluded flaky group).
• The final PublishRateLimiterOverconsumingTest and PublishRateLimiterTokenDebtTest passed with retries disabled, together with quickCheck.
• A temporary negative-control copy of the bounded-read rate test failed its original rate assertion when publishing limits were disabled: all 3,000 messages arrived in the first measured window. This confirms that the larger payload and smaller receive buffer do not themselves make the test pass without throttling.

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

The removed handler is a public implementation class in pulsar-common, not a client or broker extension API; code directly importing that internal utility would need adjustment. Publish pacing and the short-term overshoot of existing backpressure thresholds change for all broker connections. Configuration defaults, metric definitions, and the wire protocol are unchanged.

Best practice

• Treat publish-rate and pending-publish thresholds as backpressure triggers, with headroom for data already read and asynchronous downstream work. They are not strict instantaneous ceilings.
• Account for small-message density, concurrent connections, producer batching, and slow BookKeeper writes when sizing that headroom. Extra downstream objects can matter even when the retained input capacity stays the same.
• Evaluate sustained rate control together with token-debt recovery; a short publish burst alone does not establish that throttling is disabled. Deployments relying on the previous per-message pacing must account for the larger bursts and pauses introduced by this change.
• Track retained input allocations separately from logical pending-publish payload where available. Removing this handler alone does not add that accounting or solve retention by caches and other buffer owners.

[lhotari]

When a publish limit disables auto-read while handling one frame, the decoder can still emit the remaining complete frames from the same input. The flow-control handler queues those frames until reading resumes.

Why is this a problem in practice? Please file an issue that describes the concrete problem, ideally with a reproducer.

That queue improves per-message pacing, but it does not undo the socket read or release the received data. A queued frame retains its ByteBuf, and a retained slice can keep a much larger underlying allocation alive. [...]

If retained slices pinning larger allocations are a real problem, it can already be solved without changing the pipeline. Override extractFrame in io.netty.handler.codec.LengthFieldBasedFrameDecoder so it copies the frame instead of returning a retained slice. That has a cost: extra CPU for the copy and possibly more memory fragmentation, so it would need to be measured too.

I think the simpler approach is to estimate the allocation overhead of the current design and size the direct memory limits with that overhead included. We already do this for a container's maximum RSS: you estimate the overhead and leave enough headroom so the limit isn't exceeded.

When you look at RSS, neither Netty's allocator nor the OS allocator has deterministic overhead. There is always some fragmentation, and the OS allocator holds on to memory so it doesn't have to allocate and free it constantly. On the Netty side, the newer adaptive allocator reduces fragmentation, but it has to be paired with a suitable malloc implementation. Alpine's default malloc doesn't work well with it, so on Alpine you need mimalloc, which the Pulsar 5.x Dockerfile now uses (#26791).

On backpressure more broadly: Pulsar lacks proper producer backpressure, and fixing that would be more effective than adjusting memory limits and calculations. The current approach of backpressuring connections by setting auto-read to false is flawed and causes many problems. I've collected the details in this WIP document (it's currently a pile of notes for myself).

This change also deliberately reverts the per-message pacing restored by #26013, so it could have a significant behavioral impact for users who rely on publish rate limiting. It needs a careful performance analysis before it's applied. That could be done with the tests/performance scenarios on Linux x86_64/amd64 using the actual Docker image, since the malloc implementation affects the results.

This PR changes throttling behavior and is labeled for the 4.0.x and 4.2.x maintenance releases. A behavior change like this needs a PIP and shouldn't be cherry-picked to maintenance branches before the PIP is discussed and approved.

[Denovo1998]

One potential issue with removing this handler is that a single socket read may now deliver multiple over-limit SEND frames on the same connection.

PublishRateLimiterImpl queues an unthrottle operation for each exceeded request, but the BrokerPublishRate in ServerCnxThrottleTracker is non-reentrant—repeated throttle marks are dropped. If a prior unthrottle callback executes after a newer burst has already throttled the connection again, it could inadvertently clear the newer throttle state. Multiple callbacks also introduce unnecessary work during small-message bursts.

Could we add a regression test to cover this interleaving scenario and consider coalescing or versioning the unthrottle callbacks? The new token-debt test batches 600 messages into a single SEND frame per connection, so it does not exercise this case.

[codelipenghui]

Reviewed head: 191e2cff46f23f3a91445ba2cf3082a08d2b0728.

Requesting changes for the P2 broker-throttle race already raised in the inline discussion. Queue removal is a reasonable direction, but broker throttle activations and releases need to remain correctly paired.

Removing the queue allows multiple over-budget SEND frames on one connection to each enqueue an unthrottle callback. BrokerPublishRate is non-reentrant, so repeated marks collapse into one flag, while the corresponding releases remain separately queued. The following interleaving can then occur:

1. Tokens replenish, and the first old callback resumes the connection.
2. A fresh socket read delivers another publish, consumes the available tokens and throttles the connection again.
3. A delayed callback from the previous burst clears that newer throttle.

Confirmed by a local controlled regression: with a 10-token/s limit, an initial read batch containing two frames charged 20 tokens each. After replenishment, the scheduler was paused before submitting its second old resume callback. The first callback allowed a fresh frame through a real socket, producing a token balance of -10. Releasing the delayed callback then left autoRead=true, failing the assertion that the new throttle must remain active. The equivalent reentrant TopicPublishRate control passed.

This is not the intentional overshoot from already-read frames: it prematurely enables subsequent reads while new token debt remains.

Please pair broker activations with their queued releases—for example, using reentrant counting as for topic throttling—and add coverage for multiple over-budget frames on the same connection followed by delayed old callbacks. Restoring the frame queue is not required to fix this.

Validation: 23 repository test cases and one counter-based control passed with retries disabled; the local broker regression failed as described. The reproduction uses the production initializer, decoder, limiter and throttle tracker, with a mocked business handler and controlled scheduler handoff. It is not a production-load reproduction. The original-revision baseline and a production fix were not executed.