[Bug] KvBatchScanner leaks response buffers even after normal close

[fxbing]

Search before asking

• I searched in the issues and found nothing similar.

Fluss version

main (development)

Please describe the bug 🐞

Scanning a primary-key table without a limit through TableScan.createBatchScanner() leaks the network buffers held by ScanKvResponse, even when all results are consumed and the scanner is closed normally.

Four paths reproduce the leak:

1. An empty terminal response.
2. A response containing a business error.
3. Closing the scanner after an in-flight response has completed but before it is consumed.
4. A successful response to the close-scanner request.

In deterministic reproductions using serialized and parsed responses, the underlying ByteBuf.refCnt() remains at 1 instead of reaching 0.

ScanKvResponse is lazily parsed, so the RPC layer transfers buffer ownership to its consumer, including for responses without records. KvBatchScanner currently does not release these buffers.

Solution

Release consumed responses in finally, release unconsumed responses through a completion callback when closing, and release close-scanner responses.

The pending future should remain uncancelled so that a late response can still reach the cleanup callback.

Are you willing to submit a PR?

• I'm willing to submit a PR!