[lake/tiering] Fix lake writer resource leak in TieringSplitReader close chain

[beryllw]

Purpose

Linked issue: close #3719

TieringSplitReader never closes in-flight lake writers on failure. The leaked Arrow buffers make the log scanner close fail with "Memory was leaked by query", masking the real exception during failover.

Brief change log

• completeLakeWriter(): always close the writer even when complete() throws.
• close(): close all in-flight lake writers (before the log scanner, since their Arrow buffers are allocated from it) and the snapshot split reader.
• finishCurrentTable(): close residual writers before switching tables.
• Introduce IOUtils.closeAll() (aligned with Flink) to close a group of resources and throw the first failure with the rest suppressed.

https://github.com/apache/flink/blob/c643a2953ba44b3b316ba52983932329dc0162e4/flink-core/src/main/java/org/apache/flink/util/IOUtils.java#L245-L271

Tests

API and Format

Documentation

[beryllw]

@luoyuxia cc

[Copilot]

Pull request overview

Fixes a resource-leak/masked-exception issue in Fluss’ Flink tiering source by ensuring in-flight lake writers are reliably closed (including on failure paths) before closing the log scanner/allocator chain, and by adding a shared utility to close multiple resources while preserving the first failure and suppressing the rest.

Changes:

• Make TieringSplitReader always close lake writers when complete() fails, when closing the reader, and defensively when switching/finishing tables.
• Introduce IOUtils.closeAll() to close a collection of AutoCloseables while throwing the first exception and suppressing subsequent ones.
• Add/extend unit tests to validate lake writers are closed on normal completion, on complete() failure, and on split-reader close with writers still in-flight.

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated no comments.

Show a summary per file

File	Description
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java	Ensures lake writers are closed on failure and in close/finish paths; closes writers before log scanner to avoid Arrow allocator leak errors masking root causes.
fluss-common/src/main/java/org/apache/fluss/utils/IOUtils.java	Adds closeAll() helper to close groups of resources and propagate the first close failure with others suppressed.
fluss-common/src/test/java/org/apache/fluss/utils/IOUtilsTest.java	Adds coverage for IOUtils.closeAll() behavior (null tolerance, suppression semantics, and closing all elements).
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/TestingLakeTieringFactory.java	Enhances the testing factory to track created writers, inject complete() failures, and expose closed state for assertions.
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/TieringSplitReaderTest.java	Adds assertions/tests verifying writers are closed on normal completion, on complete() failure, and during reader close while writers are still in-flight.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.