[FLINK-33981][runtime] Fix not closing DirectoryStream after listing …

…local state files
javabrett · Feb 5, 2024 · 360c1a0 · 360c1a0
1 parent de2d175
commit 360c1a0
Showing 1 changed file with 6 additions and 3 deletions.
diff --git a/...ime/src/main/java/org/apache/flink/runtime/state/TaskExecutorLocalStateStoresManager.java b/...ime/src/main/java/org/apache/flink/runtime/state/TaskExecutorLocalStateStoresManager.java
@@ -46,6 +46,7 @@
 import java.util.Set;
 import java.util.concurrent.Executor;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 /**
  * This class holds the all {@link TaskLocalStateStoreImpl} objects for a task executor (manager).
@@ -293,9 +294,11 @@ private Collection<AllocationID> findStoredAllocations() {
     @Nonnull
     static Collection<Path> listAllocationDirectoriesIn(File localStateRootDirectory)
             throws IOException {
-        return Files.list(localStateRootDirectory.toPath())
-                .filter(path -> path.getFileName().toString().startsWith(ALLOCATION_DIR_PREFIX))
-                .collect(Collectors.toList());
+        try (Stream<Path> fileListStream = Files.list(localStateRootDirectory.toPath())) {
+            return fileListStream
+                    .filter(path -> path.getFileName().toString().startsWith(ALLOCATION_DIR_PREFIX))
+                    .collect(Collectors.toList());
+        }
     }
 
     public void shutdown() {