fix(stream): release FetchResults if the subsequent fetch fails

Signed-off-by: Ning Yu <[email protected]>

Author: Chillax-0v0

Diff for: core/src/main/scala/kafka/log/streamaspect/ElasticLogFileRecords.java

@@ -159,6 +159,7 @@ private CompletableFuture<LinkedList<FetchResult>> fetch0(FetchContext context,
                         } else {
                             LOGGER.error("Invalid record batch, last offset {} is less than next offset {}",
                                     recordBatchWithContext.lastOffset(), nextFetchOffset);
+                            rst.free();
                             throw new IllegalStateException();
                         }
                         readSize += recordBatchWithContext.rawPayload().remaining();
@@ -168,6 +169,10 @@ private CompletableFuture<LinkedList<FetchResult>> fetch0(FetchContext context,
                                 // add to first since we need to reverse the order.
                                 rstList.addFirst(rst);
                                 return rstList;
+                            }).whenComplete((r, e) -> {
+                                if (e != null) {
+                                    rst.free();
+                                }
                             });
                 });
     }