From 9a5fd8428222f3cfc4675efb84261c3582cdbb7d Mon Sep 17 00:00:00 2001
From: Yordan Pavlov <y.d.pavlov@gmail.com>
Date: Mon, 17 Aug 2026 12:46:00 +0000
Subject: [PATCH] [FLINK-40327][state/forst] Close ReadOptions in
 ForStGeneralMultiGetOperation to fix native memory leak

ForStGeneralMultiGetOperation.process() creates a native ReadOptions per
async read batch and never closes it, leaking native memory proportional
to async read volume. Wrap it in try-with-resources: multiGetAsList is
synchronous within the executor lambda and returns byte[] copies, so
nothing referencing the ReadOptions outlives the batch.
---
 .../flink/state/forst/ForStGeneralMultiGetOperation.java       | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStGeneralMultiGetOperation.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStGeneralMultiGetOperation.java
index a235de7e2087c5..ebf379797b1acb 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStGeneralMultiGetOperation.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStGeneralMultiGetOperation.java
@@ -80,8 +80,7 @@ public CompletableFuture<Void> process() {
         for (List<ForStDBGetRequest<?, ?, ?, ?>> getRequests : splitRequests) {
             executor.execute(
                     () -> {
-                        try {
-                            ReadOptions readOptions = new ReadOptions();
+                        try (ReadOptions readOptions = new ReadOptions()) {
                             readOptions.setReadaheadSize(0);
                             List<byte[]> keys = new ArrayList<>(getRequests.size());
                             List<ColumnFamilyHandle> columnFamilyHandles =