[client] LogRecordReadContext.close() can leak Arrow memory after a failed close

[EchoLee5]

Search before asking

• I searched in the issues and found nothing similar.

Fluss version

main (development), follow-up to #4345 / #4350

Please describe the bug 🐞

LogRecordReadContext.close() in #4350 always clears vectorSchemaRootMap and nulls unshadedBufferAllocator in a finally block, even when IOUtils.closeAll() failed.

Two consequences:

1. Failed close is not retryable. The previous implementation only nulled unshadedBufferAllocator after a successful close. If close throws because callers still hold Arrow batches (pollRecordBatch()), a later close() no longer has a reference to the unshaded allocator and cannot release it. The same applies to vector schema roots that failed to close.

2. Unshaded allocator is closed outside unshadedArrowResourceLock. The close path snapshots the allocator under the lock, then closes it after releasing the lock. getOrCreateUnshadedBufferAllocator() can still observe the old reference and hand it out while close() is closing it. LogRecordReadContext is annotated @ThreadSafe.

synchronized (unshadedArrowResourceLock) {
    if (unshadedBufferAllocator != null) {
        final AutoCloseable unshaded = unshadedBufferAllocator;
        closeables.add(() -> unshaded.close());
    }
}
try {
    IOUtils.closeAll(closeables); // closes unshaded without holding the lock
} finally {
    vectorSchemaRootMap.clear();
    synchronized (unshadedArrowResourceLock) {
        unshadedBufferAllocator = null; // dropped even if close failed
    }
}

Expected

• Keep allocator/root references when close fails, so a later close() can still release leftover native memory.
• Close the unshaded allocator while holding unshadedArrowResourceLock, matching getOrCreateUnshadedBufferAllocator().

Are you willing to submit a PR?

• I am willing to submit a PR!

[EchoLee5]

Closing: this is a bug introduced by unmerged #4350, not a standalone defect. It should be fixed on that PR instead of a new issue.