[lake/tiering] Tiering job leaks lake writers on failure and masks the real exception

[beryllw]

Search before asking

• I searched in the issues and found nothing similar.

Fluss version

0.9.0 (latest release)

Please describe the bug 🐞

When a tiering task fails and Flink triggers a failover, the close path throws "Memory was leaked by query", masking the real failure:

2026-07-21 12:04:14,661 INFO  org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Closing Source Reader.
2026-07-21 12:04:14,661 INFO  org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - Shutting down split fetcher 42
2026-07-21 12:04:14,819 INFO  org.apache.fluss.client.table.scanner.log.RemoteLogDownloader$DownloadRemoteLogThread [] - Shutting down
2026-07-21 12:04:14,819 INFO  org.apache.fluss.client.table.scanner.log.RemoteLogDownloader$DownloadRemoteLogThread [] - Stopped
2026-07-21 12:04:14,820 INFO  org.apache.fluss.client.table.scanner.log.RemoteLogDownloader$DownloadRemoteLogThread [] - Shutdown completed
2026-07-21 12:04:14,868 ERROR org.apache.arrow.memory.BaseAllocator                        [] - Memory was leaked by query. Memory leaked: (81920)
Outstanding child allocators : 
  Allocator(system-column-allocator) 0/81920/81920/9223372036854775807 (res/actual/peak/limit)
Allocator(ROOT) 0/81920/4742874/9223372036854775807 (res/actual/peak/limit)

2026-07-21 12:04:14,869 ERROR org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager [] - Received uncaught exception.
java.lang.RuntimeException: Failed to close Arrow buffer allocator. Arrow batches returned by pollRecordBatch() must be closed before closing the scanner.
	at org.apache.fluss.record.LogRecordReadContext.close(LogRecordReadContext.java:401) ~[xxx]
	at org.apache.fluss.client.table.scanner.log.LogFetcher.close(LogFetcher.java:617) ~[xxx]
	at org.apache.fluss.client.table.scanner.log.LogScannerImpl.close(LogScannerImpl.java:375) ~[xxx]
	at org.apache.fluss.flink.tiering.source.TieringSplitReader.close(TieringSplitReader.java:768) ~[xxx]
	at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:124) [xxx]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) [?:?]
	at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
	at java.lang.Thread.run(Thread.java:991) [?:?]
Caused by: java.lang.IllegalStateException: Memory was leaked by query. Memory leaked: (81920)
Outstanding child allocators : 
  Allocator(system-column-allocator) 0/81920/81920/9223372036854775807 (res/actual/peak/limit)
Allocator(ROOT) 0/81920/4742874/9223372036854775807 (res/actual/peak/limit)

	at org.apache.arrow.memory.BaseAllocator.close(BaseAllocator.java:504) ~[blob_p-5c46eb4a000924a0af48b02fdbab6d18c382ee46-3b580ff3b0d21175f35f1083631c63c1:xxx]
	at org.apache.arrow.memory.RootAllocator.close(RootAllocator.java:27) ~[blob_p-5c46eb4a000924a0af48b02fdbab6d18c382ee46-3b580ff3b0d21175f35f1083631c63c1:xxx]
	at org.apache.fluss.record.LogRecordReadContext.close(LogRecordReadContext.java:399) ~[xxx]
	... 9 more
2026-07-21 12:04:14,870 INFO  org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - Split fetcher 42 exited.
2026-07-21 12:04:14,870 INFO  org.apache.fluss.client.token.DefaultSecurityTokenManager    [] - Stopping security token renewal
2026-07-21 12:04:14,871 INFO  org.apache.fluss.client.token.DefaultSecurityTokenManager    [] - Stopped security token renewal
2026-07-21 12:04:16,878 INFO  org.apache.fluss.rpc.netty.client.NettyClient                [] - Netty client was shutdown successfully.

How to reproduce

1. Run a tiering job.
2. Make a lake writer fail in complete() (e.g. a transient storage error), or fail the task while lake writers are still writing.
3. The close path throws "Memory was leaked by query" instead of the real exception.

Root cause

TieringSplitReader never closes in-flight lake writers on failure:

• completeLakeWriter() removes the writer from the tracking map before calling complete(); if complete() throws, the writer is never closed.
• close() only closes the log scanner and table; in-flight lake writers and the snapshot split reader are left unclosed.

The leaked writers hold Arrow buffers allocated from the log scanner, so closing the
scanner afterwards fails.

Impact

Resources (Arrow buffers, lake write buffers, output streams) leak on every restart. The real exception is buried, making troubleshooting harder.

Solution

Always close lake writers: on complete() failure, on split reader close (before the log scanner), and when switching tables.

Are you willing to submit a PR?

• I'm willing to submit a PR!