[fix][client] Return flow-control permit when client seek by messageID

[programmerahul]

Fixes #26727

Motivation

When a consumer seeks to (or is created with) a specific startMessageId, the broker re-dispatches the boundary message, which the client filters out in messageReceived() via the isSameEntry(msgId) && isPriorEntryIndex(...) block. That block released the payload and returned without calling increaseAvailablePermits(), so the dropped message's outstanding flow-control permit was never repaid (it is not delivered, so messageProcessed() never runs for it).

Each seek-to-messageId therefore leaks one permit. It is masked at normal receiver queue sizes (seek re-establishes flow control), but with receiverQueueSize=1 the single leaked permit exhausts the whole budget and the consumer stalls immediately after the seek -- the message after the seek target is never delivered. Not chunking-specific; applies to plain messages too.

Modifications

return the permit in the drop block. Adds a test (plain messages, receiverQueueSize=1) that stalls without the fix and passes with it.

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change added tests and can be verified as follows:

• *Added test-case so that no permit leak while seek operation *

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• 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

[lhotari]

Thanks for tracking down this subtle flow-control leak and for adding a regression test with receiverQueueSize=1 that makes it visible.

I checked the premise and it holds: after a seek the consumer reconnects, ConsumerImpl.java:1071 resets the client permits and ConsumerImpl.java:975 sends a fresh full Flow, and only then does the broker re-dispatch the boundary entry. That entry consumes one broker-side permit and is dropped at ConsumerImpl.java:1541 before messageProcessed() can repay it, so the reconnect reset does not cover it. I also ran the new test with the added increaseAvailablePermits(cnx) removed and it fails (receive returns null); with the fix it passes. I found no double-count with chunked messages (non-last chunks are credited in processMessageChunk, the last chunk is repaid by the new line), zero-queue consumers, the payload-processor path or MultiTopicsConsumerImpl, and partially dropped batches are already refunded through the skipped-messages accounting. Two small suggestions inline.

[lhotari]

[QUALITY] refund numMessages rather than a fixed 1 so an undecryptable batch is repaid exactly

This block is also reached for an undecryptable batch (isMessageUndecryptable with the CONSUME crypto failure action), where the broker charged numMessages permits for the entry but only one is repaid here. Using increaseAvailablePermits(cnx, numMessages) (as the duplicate-ack drop at ConsumerImpl.java:1479 does) repays exactly what was consumed, and is identical for the plain and chunked cases where numMessages is 1. Not a regression, just a cheap way to close the remaining gap.

[lhotari]

[QUALITY] the seek test is not chunking-specific and could live with the other seek tests

The test uses plain, non-chunked messages and the PR description says the bug is not chunking-specific, so it would be easier to find in a seek-focused class (for example the existing seek tests in SimpleProducerConsumerTest). The Javadoc that quotes the production code block will also go stale as ConsumerImpl evolves; a short description of the symptom would age better.

It would also be worth adding a chunked variant (a chunked message as the seek boundary with receiverQueueSize=1), since the code comment in ConsumerImpl claims that case is covered.