Flink job is deployed to YARN failed with java.lang.OutOfMemoryError: Direct buffer memory

### Search before asking

- [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar.

### Fluss version

0.8.0 (latest release)

### Please describe the bug 🐞

Caused by: java.lang.OutOfMemoryError: Direct buffer memory. The direct out-of-memory error has occurred. This can mean two things: either job(s) require(s) a larger size of JVM direct memory or there is a direct memory leak. The direct memory can be allocated by user code or some of its dependencies. In this case 'taskmanager.memory.task.off-heap.size' configuration option should be increased. Flink framework and its dependencies also consume the direct memory, mostly for network communication. The most of network memory is managed by Flink and should not result in out-of-memory error. In certain special cases, in particular for jobs with high parallelism, the framework may require more direct memory which is not managed by Flink. In this case 'taskmanager.memory.framework.off-heap.size' configuration option should be increased. If the error persists then there is probably a direct memory leak in user code or some of its dependencies which has to be investigated and fixed. The task executor has to be shutdown...
at java.nio.Bits.reserveMemory(Bits.java:178) ~[?:?]
at java.nio.DirectByteBuffer.(DirectByteBuffer.java:121) ~[?:?]
at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:332) ~[?:?]
at org.apache.fluss.shaded.netty4.io.netty.buffer.PoolArena$DirectArena.allocateDirect(PoolArena.java:710) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.shaded.netty4.io.netty.buffer.PoolArena$DirectArena.newChunk(PoolArena.java:685) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.shaded.netty4.io.netty.buffer.PoolArena.allocateNormal(PoolArena.java:212) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.shaded.netty4.io.netty.buffer.PoolArena.tcacheAllocateSmall(PoolArena.java:177) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.shaded.netty4.io.netty.buffer.PoolArena.allocate(PoolArena.java:134) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.shaded.netty4.io.netty.buffer.PoolArena.allocate(PoolArena.java:126) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.shaded.netty4.io.netty.buffer.PooledByteBufAllocatorL$InnerAllocator.newDirectBufferL(PooledByteBufAllocatorL.java:178) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.shaded.netty4.io.netty.buffer.PooledByteBufAllocatorL$InnerAllocator.directBuffer(PooledByteBufAllocatorL.java:211) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.shaded.netty4.io.netty.buffer.PooledByteBufAllocatorL.allocate(PooledByteBufAllocatorL.java:58) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.shaded.arrow.org.apache.arrow.memory.NettyAllocationManager.(NettyAllocationManager.java:77) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.shaded.arrow.org.apache.arrow.memory.NettyAllocationManager.(NettyAllocationManager.java:84) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.shaded.arrow.org.apache.arrow.memory.NettyAllocationManager$1.create(NettyAllocationManager.java:34) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.shaded.arrow.org.apache.arrow.memory.BaseAllocator.newAllocationManager(BaseAllocator.java:355) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.shaded.arrow.org.apache.arrow.memory.BaseAllocator.newAllocationManager(BaseAllocator.java:350) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.shaded.arrow.org.apache.arrow.memory.BaseAllocator.bufferWithoutReservation(BaseAllocator.java:338) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.shaded.arrow.org.apache.arrow.memory.BaseAllocator.buffer(BaseAllocator.java:316) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.shaded.arrow.org.apache.arrow.memory.RootAllocator.buffer(RootAllocator.java:29) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.shaded.arrow.org.apache.arrow.memory.BaseAllocator.buffer(BaseAllocator.java:280) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.shaded.arrow.org.apache.arrow.memory.RootAllocator.buffer(RootAllocator.java:29) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.shaded.arrow.org.apache.arrow.vector.ipc.message.MessageSerializer.readMessageBody(MessageSerializer.java:726) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.shaded.arrow.org.apache.arrow.vector.ipc.message.MessageSerializer.deserializeRecordBatch(MessageSerializer.java:363) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.utils.ArrowUtils.createArrowReader(ArrowUtils.java:160) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.record.DefaultLogRecordBatch.columnRecordIterator(DefaultLogRecordBatch.java:309) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.record.DefaultLogRecordBatch.records(DefaultLogRecordBatch.java:222) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.client.table.scanner.log.CompletedFetch.nextFetchedRecord(CompletedFetch.java:220) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.client.table.scanner.log.CompletedFetch.fetchRecords(CompletedFetch.java:170) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.client.table.scanner.log.LogFetchCollector.fetchRecords(LogFetchCollector.java:160) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.client.table.scanner.log.LogFetchCollector.collectFetch(LogFetchCollector.java:118) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.client.table.scanner.log.LogFetcher.collectFetch(LogFetcher.java:158) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.client.table.scanner.log.LogScannerImpl.pollForFetches(LogScannerImpl.java:231) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.client.table.scanner.log.LogScannerImpl.poll(LogScannerImpl.java:141) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.fluss.flink.source.reader.FlinkSourceSplitReader.fetch(FlinkSourceSplitReader.java:172) ~[afluss-flink-1.18-0.8.0-incubating.jar:0.8.0-incubating]
at org.apache.flink.connector.base.source.reader.fetcher.FetchTask.run(FetchTask.java:58) ~[flink-connector-files-1.18.1.jar:1.18.1]
at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:165) ~[flink-connector-files-1.18.1.jar:1.18.1]
at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:117) ~[flink-connector-files-1.18.1.jar:1.18.1]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) ~[?:?]
at java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[?:?]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[?:?]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[?:?]
... 1 more

### Solution

_No response_

### Are you willing to submit a PR?

- [x] I'm willing to submit a PR!

Start at ArrowUtils.createArrowReader and trace the direct-memory allocation through DefaultLogRecordBatch.columnRecordIterator and FlinkSourceSplitReader.fetch. Reproduce the Flink job on YARN using Fluss 0.8.0, then determine whether the failure is caused by insufficient direct memory or a leak. Done means the job no longer fails with Direct buffer memory exhaustion.

Written by the indexing model from the issue text.