Fix reader resource leakage (#51)

* Fix resource leakage withing streams
exacaster · Jun 13, 2024 · d8f2e72 · d8f2e72
1 parent 6612a57
commit d8f2e72
Showing 1 changed file with 7 additions and 1 deletion.
diff --git a/src/main/java/com/exacaster/deltafetch/search/parquet/ParquetLookupReader.java b/src/main/java/com/exacaster/deltafetch/search/parquet/ParquetLookupReader.java
@@ -35,7 +35,13 @@ public ParquetLookupReader(Configuration conf, String path) {
     public Stream<Map<String, Object>> find(List<ColumnValueFilter> filters, int limit) {
         LOG.debug("Reading: {} with filters {}", path, filters);
         try (var reader = prepareReader(filters)) {
-            return Streams.stream(Iterators.limit(new ParquetIterator(reader), limit));
+            return Streams.stream(Iterators.limit(new ParquetIterator<>(reader), limit)).onClose(() -> {
+                try {
+                    reader.close();
+                } catch (IOException e) {
+                    LOG.error("Failed to close ParquetReader", e);
+                }
+            });
         } catch (IOException e) {
             throw new IllegalStateException("Failed building reader", e);
         }