[client] Replace Netty PooledByteBufAllocator with bump-pointer ChunkedAllocationManager for Arrow memory allocation.

Author: loserwang1024

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/LimitBatchScanner.java

@@ -18,6 +18,7 @@
 package org.apache.fluss.client.table.scanner.batch;
 
 import org.apache.fluss.client.metadata.MetadataUpdater;
+import org.apache.fluss.compression.ChunkedAllocationManager;
 import org.apache.fluss.exception.LeaderNotAvailableException;
 import org.apache.fluss.metadata.KvFormat;
 import org.apache.fluss.metadata.SchemaGetter;
@@ -164,7 +165,12 @@ private List<InternalRow> parseLimitScanResponse(LimitScanResponse limitScanResp
             }
         } else {
             LogRecordReadContext readContext =
-                    LogRecordReadContext.createReadContext(tableInfo, false, null, schemaGetter);
+                    LogRecordReadContext.createReadContext(
+                            tableInfo,
+                            false,
+                            null,
+                            schemaGetter,
+                            new ChunkedAllocationManager.ChunkedFactory());
             LogRecords records = MemoryLogRecords.pointToByteBuffer(recordsBuffer);
             for (LogRecordBatch logRecordBatch : records.batches()) {
                 // A batch of log record maybe little more than limit, thus we need slice the

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java

@@ -23,6 +23,7 @@
 import org.apache.fluss.client.metrics.ScannerMetricGroup;
 import org.apache.fluss.client.table.scanner.RemoteFileDownloader;
 import org.apache.fluss.cluster.BucketLocation;
+import org.apache.fluss.compression.ChunkedAllocationManager;
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.ApiException;
@@ -123,10 +124,14 @@ public LogFetcher(
             SchemaGetter schemaGetter) {
         this.tablePath = tableInfo.getTablePath();
         this.isPartitioned = tableInfo.isPartitioned();
+        ChunkedAllocationManager.ChunkedFactory chunkedFactory =
+                new ChunkedAllocationManager.ChunkedFactory();
         this.readContext =
-                LogRecordReadContext.createReadContext(tableInfo, false, projection, schemaGetter);
+                LogRecordReadContext.createReadContext(
+                        tableInfo, false, projection, schemaGetter, chunkedFactory);
         this.remoteReadContext =
-                LogRecordReadContext.createReadContext(tableInfo, true, projection, schemaGetter);
+                LogRecordReadContext.createReadContext(
+                        tableInfo, true, projection, schemaGetter, chunkedFactory);
         this.projection = projection;
         this.logScannerStatus = logScannerStatus;
         this.maxFetchBytes =

fluss-client/src/main/java/org/apache/fluss/client/write/RecordAccumulator.java

@@ -22,6 +22,7 @@
 import org.apache.fluss.client.metrics.WriterMetricGroup;
 import org.apache.fluss.cluster.BucketLocation;
 import org.apache.fluss.cluster.Cluster;
+import org.apache.fluss.compression.ChunkedAllocationManager;
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.FlussRuntimeException;
@@ -36,7 +37,6 @@
 import org.apache.fluss.row.arrow.ArrowWriter;
 import org.apache.fluss.row.arrow.ArrowWriterPool;
 import org.apache.fluss.shaded.arrow.org.apache.arrow.memory.BufferAllocator;
-import org.apache.fluss.shaded.arrow.org.apache.arrow.memory.RootAllocator;
 import org.apache.fluss.utils.CopyOnWriteMap;
 import org.apache.fluss.utils.MathUtils;
 import org.apache.fluss.utils.clock.Clock;
@@ -63,6 +63,7 @@
 
 import static org.apache.fluss.record.LogRecordBatchFormat.NO_BATCH_SEQUENCE;
 import static org.apache.fluss.record.LogRecordBatchFormat.NO_WRITER_ID;
+import static org.apache.fluss.shaded.arrow.org.apache.arrow.memory.AllocatorUtil.createBufferAllocator;
 import static org.apache.fluss.utils.Preconditions.checkNotNull;
 
 /* This file is based on source code of Apache Kafka Project (https://kafka.apache.org/), licensed by the Apache
@@ -134,7 +135,7 @@ public final class RecordAccumulator {
                 Math.max(1, (int) conf.get(ConfigOptions.CLIENT_WRITER_BATCH_SIZE).getBytes());
 
         this.writerBufferPool = LazyMemorySegmentPool.createWriterBufferPool(conf);
-        this.bufferAllocator = new RootAllocator(Long.MAX_VALUE);
+        this.bufferAllocator = createBufferAllocator(new ChunkedAllocationManager.ChunkedFactory());
         this.arrowWriterPool = new ArrowWriterPool(bufferAllocator);
         this.incomplete = new IncompleteBatches();
         this.nodesDrainIndex = new HashMap<>();