[client] LogFetcher close chain leaks Arrow memory when close fails mid-chain

[beryllw]

Search before asking

• I searched in the issues and found nothing similar.

Fluss version

main (development)

Please describe the bug 🐞

The log scanner close chain leaks off-heap Arrow memory when any intermediate close() fails, because each layer closes resources sequentially and aborts on the first exception.

Three leak paths:

1. LogScannerImpl.pollRecordBatch() — sendFetches() is called after scanRecords is already materialized but without a try-catch. If sendFetches() throws (e.g., OSS QpsLimitExceeded during Paimon write), the polled Arrow batches are neither returned to the caller nor closed.

2. LogFetcher.close() — iterates tableReadContexts in a plain for loop. Any failing TableReadContext.close() skips the remaining contexts and the shared chunkedFactory. Because isClosed is already true, the pooled native chunks are never released.

3. LogRecordReadContext.close() — if bufferAllocator.close() throws due to outstanding Arrow allocations, unshadedBufferAllocator is neither closed nor nulled. Same sequential-skip issue in TableReadContext.close(): readContext.close() failing prevents remoteReadContext.close().

This is a follow-up to #3719, which fixed lake writer leaks in TieringSplitReader but did not address the deeper scanner close chain.

Solution

Use IOUtils.closeAll() at each layer to close all resources regardless of individual failures, and clear references in finally blocks. Wrap sendFetches() in try-catch to release polled batches on write failure.

Are you willing to submit a PR?

• I am willing to submit a PR!