[client] Replace Unsafe with Netty direct memory in ChunkedAllocationManager and expose metrics.

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

@@ -106,7 +106,7 @@ public final class RecordAccumulator {
     /** The arrow buffer allocator to allocate memory for arrow log write batch. */
     private final BufferAllocator bufferAllocator;
 
-    /** The chunked allocation manager factory, stored for explicit native memory release. */
+    /** The chunked allocation manager factory, stored for explicit direct memory release. */
     private final ChunkedAllocationManager.ChunkedFactory chunkedFactory;
 
     /** Coordinates batch memory deallocation with resource destruction. */
@@ -206,6 +206,15 @@ private void registerMetrics(WriterMetricGroup writerMetricGroup) {
         // The number of user threads blocked waiting for buffer memory to enqueue their records
         writerMetricGroup.gauge(
                 MetricNames.WRITER_BUFFER_WAITING_THREADS, writerBufferPool::queued);
+        writerMetricGroup.gauge(
+                MetricNames.WRITER_ACCUMULATOR_HEAP_MEMORY_USED_BYTES,
+                () -> writerBufferPool.totalSize() - writerBufferPool.availableMemory());
+        writerMetricGroup.gauge(
+                MetricNames.WRITER_ACCUMULATOR_ARROW_MEMORY_USED_BYTES,
+                bufferAllocator::getAllocatedMemory);
+        writerMetricGroup.gauge(
+                MetricNames.WRITER_ACCUMULATOR_DIRECT_MEMORY_ALLOCATED_BYTES,
+                chunkedFactory::getDirectMemoryAllocatedBytes);
     }
 
     /** Assigns and appends a record using the layout owned by its write context. */

fluss-client/src/test/java/org/apache/fluss/client/write/RecordAccumulatorTest.java

@@ -32,6 +32,8 @@
 import org.apache.fluss.metadata.TableDescriptor;
 import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.metrics.Gauge;
+import org.apache.fluss.metrics.MetricNames;
 import org.apache.fluss.record.ChangeType;
 import org.apache.fluss.record.DefaultKvRecord;
 import org.apache.fluss.record.IndexedLogRecord;
@@ -148,6 +150,60 @@ public void start() {
     // TODO Add more tests to test lingMs, retryBackoffMs, deliveryTimeoutMs and
     //  nextBatchExpiryTimeMs if we introduced.
 
+    @Test
+    void testAccumulatorMemoryMetrics() throws Exception {
+        int batchSize = 1024;
+        TestingWriterMetricGroup metrics = TestingWriterMetricGroup.newInstance();
+        RecordAccumulator accum =
+                createTestRecordAccumulator(
+                        5000, batchSize, 256, 10L * batchSize, bucketAssigner, metrics);
+
+        try {
+            assertThat(metrics.getMetrics())
+                    .containsKeys(
+                            MetricNames.WRITER_ACCUMULATOR_HEAP_MEMORY_USED_BYTES,
+                            MetricNames.WRITER_ACCUMULATOR_ARROW_MEMORY_USED_BYTES,
+                            MetricNames.WRITER_ACCUMULATOR_DIRECT_MEMORY_ALLOCATED_BYTES);
+
+            Gauge<?> heapMemoryUsed =
+                    (Gauge<?>)
+                            metrics.getMetrics()
+                                    .get(MetricNames.WRITER_ACCUMULATOR_HEAP_MEMORY_USED_BYTES);
+            Gauge<?> arrowMemoryUsed =
+                    (Gauge<?>)
+                            metrics.getMetrics()
+                                    .get(MetricNames.WRITER_ACCUMULATOR_ARROW_MEMORY_USED_BYTES);
+            Gauge<?> directMemoryAllocatedBytes =
+                    (Gauge<?>)
+                            metrics.getMetrics()
+                                    .get(
+                                            MetricNames
+                                                    .WRITER_ACCUMULATOR_DIRECT_MEMORY_ALLOCATED_BYTES);
+
+            assertThat(((Number) heapMemoryUsed.getValue()).longValue()).isZero();
+            assertThat(((Number) arrowMemoryUsed.getValue()).longValue()).isZero();
+            assertThat(((Number) directMemoryAllocatedBytes.getValue()).longValue()).isZero();
+
+            bucketAssigner.setBucketId(0);
+            accum.append(
+                    WriteRecord.forArrowAppend(
+                            DATA1_TABLE_INFO, DATA1_PHYSICAL_TABLE_PATH, row(1, "a"), null),
+                    (bucket, offset, exception) -> {},
+                    cluster);
+
+            long heapMemory = ((Number) heapMemoryUsed.getValue()).longValue();
+            long arrowMemory = ((Number) arrowMemoryUsed.getValue()).longValue();
+            long directMemory = ((Number) directMemoryAllocatedBytes.getValue()).longValue();
+            assertThat(heapMemory).isPositive();
+            assertThat(arrowMemory).isPositive();
+            assertThat(directMemory).isGreaterThanOrEqualTo(arrowMemory);
+        } finally {
+            accum.close();
+            accum.abortAllBatches(new RuntimeException("test cleanup"));
+            accum.destroyResources();
+        }
+    }
+
     @Test
     void testDrainBatches() throws Exception {
         // test case: node1(tb1, tb2), node2(tb3).
@@ -815,6 +871,22 @@ private RecordAccumulator createTestRecordAccumulator(
             int pageSize,
             long totalSize,
             BucketAssigner assigner) {
+        return createTestRecordAccumulator(
+                batchTimeoutMs,
+                batchSize,
+                pageSize,
+                totalSize,
+                assigner,
+                TestingWriterMetricGroup.newInstance());
+    }
+
+    private RecordAccumulator createTestRecordAccumulator(
+            int batchTimeoutMs,
+            int batchSize,
+            int pageSize,
+            long totalSize,
+            BucketAssigner assigner,
+            TestingWriterMetricGroup metrics) {
         conf.set(ConfigOptions.CLIENT_WRITER_BATCH_TIMEOUT, Duration.ofMillis(batchTimeoutMs));
         // TODO client writer buffer maybe removed.
         conf.set(ConfigOptions.CLIENT_WRITER_BUFFER_MEMORY_SIZE, new MemorySize(totalSize));
@@ -832,7 +904,7 @@ private RecordAccumulator createTestRecordAccumulator(
                                 RpcClient.create(conf, TestingClientMetricGroup.newInstance()),
                                 TabletServerGateway.class),
                         null),
-                TestingWriterMetricGroup.newInstance(),
+                metrics,
                 clock,
                 (tableInfo, path) -> assigner);
     }

fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java

@@ -337,6 +337,14 @@ public class MetricNames {
     public static final String WRITER_RECORDS_PER_BATCH = "recordsPerBatch";
     public static final String WRITER_SEND_LATENCY_MS = "sendLatencyMs";
 
+    // for record accumulator memory
+    public static final String WRITER_ACCUMULATOR_HEAP_MEMORY_USED_BYTES =
+            "accumulatorHeapMemoryUsedBytes";
+    public static final String WRITER_ACCUMULATOR_ARROW_MEMORY_USED_BYTES =
+            "accumulatorArrowMemoryUsedBytes";
+    public static final String WRITER_ACCUMULATOR_DIRECT_MEMORY_ALLOCATED_BYTES =
+            "accumulatorDirectMemoryAllocatedBytes";
+
     // for scanner
     public static final String SCANNER_TIME_MS_BETWEEN_POLL = "timeMsBetweenPoll";
     public static final String SCANNER_LAST_POLL_SECONDS_AGO = "lastPollSecondsAgo";

fluss-common/src/main/java/org/apache/fluss/shaded/arrow/org/apache/arrow/memory/ChunkedAllocationManager.java

@@ -18,11 +18,13 @@
 package org.apache.fluss.shaded.arrow.org.apache.arrow.memory;
 
 import org.apache.fluss.annotation.VisibleForTesting;
-import org.apache.fluss.shaded.arrow.org.apache.arrow.memory.util.MemoryUtil;
+import org.apache.fluss.shaded.netty4.io.netty.buffer.ByteBuf;
+import org.apache.fluss.shaded.netty4.io.netty.buffer.UnpooledByteBufAllocator;
 
 import java.util.ArrayDeque;
 import java.util.Deque;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * An {@link AllocationManager} that packs small allocations into large pre-allocated chunks using a
@@ -79,7 +81,8 @@
  * <h3>How it works</h3>
  *
  * <ul>
- *   <li>Allocates large chunks (default 4MB) from native memory via {@code Unsafe}.
+ *   <li>Allocates large chunks (default 4MB) from Netty direct buffers ({@code
+ *       UnpooledByteBufAllocator}).
  *   <li>For each small allocation request, bumps a pointer within the current active chunk.
  *   <li>Only switches to a new chunk when the current one has no room.
  *   <li>Reference-counts each chunk: when all sub-allocations are released, the chunk is recycled
@@ -106,15 +109,15 @@
  * }</pre>
  *
  * <p>For allocations >= chunkSize, a dedicated memory region is allocated directly (no bump
- * pointer), behaving identically to {@code UnsafeAllocationManager}.
+ * pointer), using a dedicated Netty {@code ByteBuf}.
  */
 public class ChunkedAllocationManager extends AllocationManager {
 
     /** 8-byte alignment for all sub-allocations within a chunk. */
     private static final long ALIGNMENT = 8;
 
     /** Default chunk size: 4MB (matches Netty 4.1+ maxOrder=9). */
-    private static final long DEFAULT_CHUNK_SIZE = 4L * 1024 * 1024;
+    private static final int DEFAULT_CHUNK_SIZE = 4 * 1024 * 1024;
 
     /** Default maximum number of empty chunks to keep in the free-list. */
     private static final int DEFAULT_MAX_FREE_CHUNKS = 3;
@@ -126,7 +129,8 @@ public class ChunkedAllocationManager extends AllocationManager {
     private final long offsetInChunk;
 
     // --- Fields for direct allocation (large request, owns its own memory) ---
-    private final long directAddress;
+    private final ByteBuf directByteBuf;
+    private final ChunkedFactory chunkedFactory;
 
     /** Sub-allocation carved from a shared {@link Chunk}. */
     private ChunkedAllocationManager(
@@ -135,16 +139,22 @@ private ChunkedAllocationManager(
         this.chunk = chunk;
         this.offsetInChunk = offset;
         this.allocatedSize = size;
-        this.directAddress = 0;
+        this.directByteBuf = null;
+        this.chunkedFactory = null;
     }
 
     /** Direct allocation for oversized requests (>= chunkSize). Owns its own memory region. */
-    private ChunkedAllocationManager(BufferAllocator accountingAllocator, long address, long size) {
+    private ChunkedAllocationManager(
+            BufferAllocator accountingAllocator,
+            ByteBuf directByteBuf,
+            long size,
+            ChunkedFactory chunkedFactory) {
         super(accountingAllocator);
         this.chunk = null;
         this.offsetInChunk = 0;
         this.allocatedSize = size;
-        this.directAddress = address;
+        this.directByteBuf = directByteBuf;
+        this.chunkedFactory = chunkedFactory;
     }
 
     @Override
@@ -155,17 +165,18 @@ public long getSize() {
     @Override
     protected long memoryAddress() {
         if (chunk != null) {
-            return chunk.address + offsetInChunk;
+            return chunk.directByteBuf.memoryAddress() + offsetInChunk;
         }
-        return directAddress;
+        return directByteBuf.memoryAddress();
     }
 
     @Override
     protected void release0() {
         if (chunk != null) {
             chunk.releaseSubAllocation();
         } else {
-            MemoryUtil.UNSAFE.freeMemory(directAddress);
+            directByteBuf.release();
+            chunkedFactory.decrementDirectMemoryBytes(allocatedSize);
         }
     }
 
@@ -174,12 +185,12 @@ protected void release0() {
     // -------------------------------------------------------------------------
 
     /**
-     * A contiguous native memory region that holds multiple small allocations via bump-pointer.
+     * A contiguous direct memory region that holds multiple small allocations via bump-pointer.
      * Reference-counted: when all sub-allocations are released (count reaches 0), the chunk is
      * recycled back to the factory's free-list.
      */
     static class Chunk {
-        final long address;
+        final ByteBuf directByteBuf;
         final long capacity;
         /** Bump pointer — only accessed under the factory's synchronized lock. */
         long used;
@@ -198,8 +209,8 @@ static class Chunk {
         /** Back-reference to the owning factory for recycling on drain. */
         final ChunkedFactory factory;
 
-        Chunk(long capacity, ChunkedFactory factory) {
-            this.address = MemoryUtil.UNSAFE.allocateMemory(capacity);
+        Chunk(int capacity, ChunkedFactory factory) {
+            this.directByteBuf = UnpooledByteBufAllocator.DEFAULT.directBuffer(capacity);
             this.capacity = capacity;
             this.used = 0;
             this.factory = factory;
@@ -266,9 +277,9 @@ void resetBump() {
             // subAllocCount is already 0 at this point.
         }
 
-        /** Frees the underlying native memory. */
+        /** Deterministically releases the underlying direct memory. */
         void destroy() {
-            MemoryUtil.UNSAFE.freeMemory(address);
+            directByteBuf.release();
         }
     }
 
@@ -288,7 +299,7 @@ void destroy() {
      */
     public static class ChunkedFactory implements AllocationManager.Factory {
 
-        private final long chunkSize;
+        private final int chunkSize;
         private final int maxFreeChunks;
 
         /** The chunk currently receiving bump allocations. May be null initially. */
@@ -297,6 +308,9 @@ public static class ChunkedFactory implements AllocationManager.Factory {
         /** Pool of empty chunks available for reuse. */
         private final Deque<Chunk> freeChunks = new ArrayDeque<>();
 
+        /** Total direct memory, in bytes, currently allocated by this factory. */
+        private final AtomicLong directMemoryAllocatedBytes = new AtomicLong();
+
         /** Set to true when {@link #close()} is called. */
         private boolean closed;
 
@@ -311,18 +325,26 @@ public ChunkedFactory() {
          * @param chunkSize maximum size of each chunk (bytes). Allocations >= this go direct.
          * @param maxFreeChunks maximum number of empty chunks to keep cached for reuse.
          */
-        public ChunkedFactory(long chunkSize, int maxFreeChunks) {
+        public ChunkedFactory(int chunkSize, int maxFreeChunks) {
             this.chunkSize = chunkSize;
             this.maxFreeChunks = maxFreeChunks;
         }
 
         @Override
         public synchronized AllocationManager create(
                 BufferAllocator accountingAllocator, long size) {
+            if (closed) {
+                throw new IllegalStateException("ChunkedFactory has been closed.");
+            }
             if (size > chunkSize) {
-                // Large allocation: give it its own memory region.
-                long address = MemoryUtil.UNSAFE.allocateMemory(size);
-                return new ChunkedAllocationManager(accountingAllocator, address, size);
+                // Large allocation: use Netty direct buffer for deterministic release.
+                if (size > Integer.MAX_VALUE) {
+                    throw new IllegalArgumentException(
+                            "Allocation size " + size + " exceeds maximum " + Integer.MAX_VALUE);
+                }
+                ByteBuf directByteBuf = UnpooledByteBufAllocator.DEFAULT.directBuffer((int) size);
+                directMemoryAllocatedBytes.addAndGet(size);
+                return new ChunkedAllocationManager(accountingAllocator, directByteBuf, size, this);
             }
 
             // Align to 8 bytes for safe direct-memory access.
@@ -352,7 +374,9 @@ private synchronized Chunk obtainChunk() {
                 recycled.resetBump();
                 return recycled;
             }
-            return new Chunk(chunkSize, this);
+            Chunk chunk = new Chunk(chunkSize, this);
+            directMemoryAllocatedBytes.addAndGet(chunkSize);
+            return chunk;
         }
 
         /**
@@ -381,16 +405,16 @@ synchronized void onChunkDrained(Chunk chunk, int expectedGeneration) {
 
             if (closed) {
                 // Factory is closed — no one will ever reuse this chunk. Free it.
-                chunk.destroy();
+                destroyChunk(chunk);
             } else if (chunk == activeChunk) {
                 // Still the active chunk — just reset bump pointer for continued use.
                 chunk.resetBump();
             } else if (freeChunks.size() < maxFreeChunks) {
                 // Not active, pool has room — recycle.
                 freeChunks.offerFirst(chunk);
             } else {
-                // Pool is full — free native memory.
-                chunk.destroy();
+                // Pool is full — free direct memory.
+                destroyChunk(chunk);
             }
         }
 
@@ -403,12 +427,31 @@ public synchronized void close() {
             closed = true;
             while (!freeChunks.isEmpty()) {
                 Chunk poll = freeChunks.poll();
-                poll.destroy();
-            }
-            if (activeChunk != null && activeChunk.subAllocCount.get() == 0) {
-                activeChunk.destroy();
+                destroyChunk(poll);
             }
+            Chunk chunk = activeChunk;
             activeChunk = null;
+            if (chunk != null && chunk.subAllocCount.get() == 0) {
+                // Invalidate a pending onChunkDrained callback that observed this zero count before
+                // close acquired the factory lock.
+                chunk.drainGeneration++;
+                destroyChunk(chunk);
+            }
+        }
+
+        /** Returns the direct memory, in bytes, currently allocated by this factory. */
+        public long getDirectMemoryAllocatedBytes() {
+            return directMemoryAllocatedBytes.get();
+        }
+
+        private void destroyChunk(Chunk chunk) {
+            long cap = chunk.capacity;
+            chunk.destroy();
+            directMemoryAllocatedBytes.addAndGet(-cap);
+        }
+
+        void decrementDirectMemoryBytes(long size) {
+            directMemoryAllocatedBytes.addAndGet(-size);
         }
 
         @VisibleForTesting