fix: release FetchResults if failed to fetch

Signed-off-by: Ning Yu <[email protected]>
AutoMQ · Nov 22, 2024 · 5330d7a · 5330d7a
1 parent 9d4d12b
commit 5330d7a
Showing 1 changed file with 6 additions and 0 deletions.
diff --git a/core/src/main/scala/kafka/log/streamaspect/ElasticLogFileRecords.java b/core/src/main/scala/kafka/log/streamaspect/ElasticLogFileRecords.java
@@ -142,6 +142,12 @@ private CompletableFuture<Records> readAll0(FetchContext context, long startOffs
         }
         List<FetchResult> results = new LinkedList<>();
         return fetch0(context, nextFetchOffset, endOffset, maxSize, results)
+                .whenComplete((nil, e) -> {
+                    if (e != null) {
+                        results.forEach(FetchResult::free);
+                        results.clear();
+                    }
+                })
                 .thenApply(nil -> PooledMemoryRecords.of(baseOffset, results, context.readOptions().pooledBuf()));
     }