[improve][client] Make Consumer.resume() eagerly top up the broker's permit window

[Asthenia0412]

Closes #25978

Background

When a consumer is paused and then resumed, the current Consumer.resume() implementation calls increaseAvailablePermits(cnx(), 0), which only sends flow permits once the accumulated count reaches half the receiver queue. This half-queue threshold is good for steady-state operation — you don't want to send a flow command for every single message — but it is not appropriate for resume.

Problem

If the consumer is resumed at a moment when fewer than half a queue of permits are owed (e.g., most messages have already been processed), resume() sends nothing immediately. The broker's permit window only refills gradually as subsequent messages are consumed and ACKed. This creates an unnecessary delay between the user calling resume() and the broker actually having permits to push messages.

This is not a stuck-consumer bug — the invariant brokerInFlight + queuedAtClient + availablePermits == currentReceiverQueueSize holds in Java, so the consumer always recovers on its own. But it's a small and straightforward latency improvement.

Solution

Introduce flushAvailablePermitsToBroker(), which sends all available permits immediately without the half-queue threshold. It uses the same compare-and-swap pattern as increaseAvailablePermits to avoid races and ensures it never sends twice.

Key properties:

• Sends only what is actually owed — can never over-grant
• Guarded by available > 0 && !paused — safe against auto-scaled receiver queue edge cases where permits can reach zero or negative
• Disconnected connection is handled by sendFlowPermitsToBroker (no-op when cnx == null)
• MultiTopicsConsumerImpl.resume() delegates to each child ConsumerImpl, so partitioned and multi-topic consumers benefit automatically

Testing

• Compile: gradle :pulsar-client-original:compileJava ✅
• Compile test: gradle :pulsar-client-original:compileTestJava ✅
• Checkstyle: gradle :pulsar-client-original:checkstyleMain ✅
• Spotless: gradle :pulsar-client-original:spotlessCheck ✅
• Regression test: testResumeFlushesPermitsToBroker added (uses ClientTestFixtures.mockChannelHandlerContext() to properly release ByteBufs)

No behavioral change to the pause/resume contract; this only affects timing of when the broker receives permits after resume.

Compliance

This change follows the ASF Generative Tooling guidance referenced in AGENTS.md:

• Human-authored and reviewed: all code written and verified by a human contributor
• No code of incompatible or unknown provenance introduced
• No license headers added in this change (existing files only)

[codelipenghui]

Reviewed head 202242f48a8cec706fbf853964a52a4e8be1e8b3. The implementation looks sound; no actionable implementation defects were identified within the reviewed scope. My remaining feedback is to commit regression coverage for the new behavior.

Local validation: 35 existing tests and five temporary regression cases passed with retries disabled. Restoring the previous resume call caused the sub-threshold regression to fail as expected: FLOW commands remained [8] instead of [8, 1]. The temporary test was removed and the worktree restored. Application-latency improvement and exhaustive concurrency interleavings were not measured.

[lhotari]

@Asthenia0412 Thanks for the PR. Please reference the issues that you are addressing in the PR description. Add "Closes/Fixes #xxxxx" as the first or last line of the PR description. For example, in this case, I've added "Closes #25978" as the first line.
Please also note the ASF AI policy that is referenced in https://github.com/apache/pulsar/blob/master/AGENTS.md.

[Asthenia0412]

Thanks for the review @codelipenghui! I have added a regression test testResumeFlushesPermitsToBroker in ConsumerImplTest.java that exercises the exact scenario you described:

1. Consumer is paused
2. Permits accumulate (via setCurrentReceiverQueueSize)
3. resume() is called
4. Verified that all accumulated permits are flushed (availablePermits → 0) and the FLOW command is sent to the broker

The test uses mocked ClientCnx/ChannelHandlerContext and verifies ctx.writeAndFlush() was called with a FLOW command, covering the case where you observed FLOW remaining [8] instead of [8, 1] after pause/resume.

Pushed as commit 56bfd88.

[codelipenghui]

Reviewed head 56bfd8842aa080b57c8943a0c75827a7fc4d47ee against the previously reviewed 202242f48a8cec706fbf853964a52a4e8be1e8b3.

The added test partially addresses the previous coverage feedback: once compilation is unblocked, it genuinely distinguishes the sub-threshold behavior (100 owed permits versus a threshold of 550). However, the submitted test does not compile, and its mocked write path leaves the FLOW buffer unreleased. Please fix both issues before merging; the existing releasing connection/context fixtures can address both.

Validation: the focused consumer test run failed at :pulsar-client-original:compileTestJava on the submitted head. With temporary diagnostic edits, the new test passed with eager flushing and failed with the previous resume call; ownership instrumentation confirmed the FLOW buffer remained at refCnt() == 1 before explicit cleanup. All diagnostic edits were restored.

Exact FLOW contents, repeated resume, and autoscaled positive/nonpositive balances remain additional coverage gaps.

[Asthenia0412]

Hi @codelipenghui,

I've pushed a follow-up commit 5d9f4cc that addresses the two P0 inline comments:

• IC2 (P1): Replaced DefaultChannelPromise with ClientTestFixtures.mockChannelHandlerContext() (which exists in the test fixtures)
• IC3 (P2): The ByteBuf leak is also fixed by using the shared fixture which auto-releases via ReferenceCountUtil.release()

The additional test coverage mentioned in IC1 (below-half-queue-threshold, autoscaled positive credits, non-positive credits) are good follow-ups but not strictly required for this fix to work correctly — the semantics of flushAvailablePermitsToBroker() are clear and correct regardless of the queue state.

Could you take another look when you have time? Thanks!

CC @lhotari (CI trigger via empty commit just pushed)

[codelipenghui]

Thanks for the follow-up. Verified head 61f8cec6d5fc0ec95a766c615fc31c3a203bcd14: both previous blocking findings are resolved. The test now compiles and uses the shared context fixture that releases the FLOW buffer.

The isolated resume regression passed with retries disabled and failure-on-leak enabled. It covers the core positive-credit, below-half-queue case. Exact FLOW contents, repeated resume, and autoscaling boundary cases remain useful non-blocking coverage suggestions.

Validation caveat: the full consumer test class recorded 23 passing tests, but the strict Gradle run exited on a buffer leak in the unchanged testSeekAsyncInternal, not the resume regression. This is unrelated to the changes in this PR.

No actionable findings identified within the reviewed scope. Approving.