Active Load: Fix scan tsfile do not close the stream resource (apache…

…#13846) (cherry picked from commit 928e6b3)
YC27 · Oct 21, 2024 · a2c43f5 · a2c43f5
1 parent 1d75836
commit a2c43f5
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 12 deletions.
diff --git a/...ode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScanner.java b/...ode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScanner.java
@@ -42,6 +42,7 @@
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Stream;
 
 public class ActiveLoadDirScanner extends ActiveLoadScheduledExecutorService {
 
@@ -99,16 +100,20 @@ private void scan() throws IOException {
 
       final boolean isGeneratedByPipe =
           listeningDir.equals(IOTDB_CONFIG.getLoadActiveListeningPipeDir());
-      FileUtils.streamFiles(new File(listeningDir), true, (String[]) null)
-          .map(
-              file ->
-                  (file.getName().endsWith(RESOURCE) || file.getName().endsWith(MODS))
-                      ? getTsFilePath(file.getAbsolutePath())
-                      : file.getAbsolutePath())
-          .filter(file -> !activeLoadTsFileLoader.isFilePendingOrLoading(file))
-          .filter(this::isTsFileCompleted)
-          .limit(currentAllowedPendingSize)
-          .forEach(file -> activeLoadTsFileLoader.tryTriggerTsFileLoad(file, isGeneratedByPipe));
+      try (final Stream<File> fileStream =
+          FileUtils.streamFiles(new File(listeningDir), true, (String[]) null)) {
+        fileStream
+            .filter(file -> !activeLoadTsFileLoader.isFilePendingOrLoading(file))
+            .filter(File::exists)
+            .map(
+                file ->
+                    (file.getName().endsWith(RESOURCE) || file.getName().endsWith(MODS))
+                        ? getTsFilePath(file.getAbsolutePath())
+                        : file.getAbsolutePath())
+            .filter(this::isTsFileCompleted)
+            .limit(currentAllowedPendingSize)
+            .forEach(file -> activeLoadTsFileLoader.tryTriggerTsFileLoad(file, isGeneratedByPipe));
+      }
     }
   }
 

diff --git a/...e/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoader.java b/...e/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoader.java
@@ -283,8 +283,8 @@ private void removeToFailDir(final String filePath) {
     }
   }
 
-  public boolean isFilePendingOrLoading(final String filePath) {
-    return pendingQueue.isFilePendingOrLoading(filePath);
+  public boolean isFilePendingOrLoading(final File file) {
+    return pendingQueue.isFilePendingOrLoading(file.getAbsolutePath());
   }
 
   // Metrics