[client] Release KvBatchScanner response buffers

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

@@ -37,6 +37,7 @@
 import org.apache.fluss.rpc.messages.ScanKvRequest;
 import org.apache.fluss.rpc.messages.ScanKvResponse;
 import org.apache.fluss.rpc.protocol.Errors;
+import org.apache.fluss.shaded.netty4.io.netty.buffer.ByteBuf;
 import org.apache.fluss.utils.CloseableIterator;
 import org.apache.fluss.utils.SchemaUtil;
 
@@ -144,25 +145,30 @@ public CloseableIterator<InternalRow> pollBatch(Duration timeout) throws IOExcep
         }
         inFlight = null;
 
-        if (response.hasErrorCode() && response.getErrorCode() != 0) {
-            return handleErrorResponse(response);
-        }
+        try {
+            if (response.hasErrorCode() && response.getErrorCode() != 0) {
+                return handleErrorResponse(response);
+            }
 
-        if (response.hasScannerId()) {
-            scannerId = response.getScannerId();
-        }
+            if (response.hasScannerId()) {
+                scannerId = response.getScannerId();
+            }
 
-        boolean hasMore = response.hasHasMoreResults() && response.isHasMoreResults();
-        if (hasMore) {
-            sendContinuation();
-        } else {
-            drained = true;
-        }
+            boolean hasMore = response.hasHasMoreResults() && response.isHasMoreResults();
+            if (hasMore) {
+                sendContinuation();
+            } else {
+                drained = true;
+            }
 
-        if (!response.hasRecords()) {
-            return drained ? null : CloseableIterator.emptyIterator();
+            if (!response.hasRecords()) {
+                return drained ? null : CloseableIterator.emptyIterator();
+            }
+            return CloseableIterator.wrap(parseRecords(response).iterator());
+        } finally {
+            // getRecords() copies the data, so returned rows no longer reference this buffer.
+            releaseResponse(response);
         }
-        return CloseableIterator.wrap(parseRecords(response).iterator());
     }
 
     @Override
@@ -275,9 +281,10 @@ private void terminate() {
         if (!closed.compareAndSet(false, true)) {
             return;
         }
-        if (inFlight != null) {
-            inFlight.cancel(true);
-            inFlight = null;
+        CompletableFuture<ScanKvResponse> pending = inFlight;
+        inFlight = null;
+        if (pending != null) {
+            pending.whenComplete((response, throwable) -> releaseResponse(response));
         }
         sendBestEffortClose();
     }
@@ -296,6 +303,7 @@ private void sendBestEffortClose() {
                                     .setCloseScanner(true))
                     .whenComplete(
                             (resp, ex) -> {
+                                releaseResponse(resp);
                                 if (ex != null) {
                                     LOG.debug(
                                             "close_scanner RPC failed for scanner of bucket {};"
@@ -309,6 +317,15 @@ private void sendBestEffortClose() {
         }
     }
 
+    private static void releaseResponse(@Nullable ScanKvResponse response) {
+        if (response != null) {
+            ByteBuf parsedByteBuf = response.getParsedByteBuf();
+            if (parsedByteBuf != null) {
+                parsedByteBuf.release();
+            }
+        }
+    }
+
     private List<InternalRow> parseRecords(ScanKvResponse response) {
         DefaultValueRecordBatch valueRecords =
                 DefaultValueRecordBatch.pointToBytes(response.getRecords());