[AMORO-4147] [Improvement]: Resource leak in CombinedDeleteFilter and MixedDeleteFilter position delete iterators. (#4148)

Author: slfan1989

amoro-format-iceberg/src/main/java/org/apache/amoro/io/reader/CombinedDeleteFilter.java

@@ -397,15 +397,18 @@ private Predicate<StructForDelete<T>> applyPosDeletes() {
     if (positionMap == null) {
       positionMap = new HashMap<>();
       List<CloseableIterable<Record>> deletes = Lists.transform(posDeletes, this::openPosDeletes);
-      CloseableIterator<Record> iterator = CloseableIterable.concat(deletes).iterator();
-      while (iterator.hasNext()) {
-        Record deleteRecord = iterator.next();
-        String path = FILENAME_ACCESSOR.get(deleteRecord).toString();
-        if (positionPathSets != null && !positionPathSets.contains(path)) {
-          continue;
+      try (CloseableIterator<Record> iterator = CloseableIterable.concat(deletes).iterator()) {
+        while (iterator.hasNext()) {
+          Record deleteRecord = iterator.next();
+          String path = FILENAME_ACCESSOR.get(deleteRecord).toString();
+          if (positionPathSets != null && !positionPathSets.contains(path)) {
+            continue;
+          }
+          Roaring64Bitmap posBitMap = positionMap.computeIfAbsent(path, k -> new Roaring64Bitmap());
+          posBitMap.add((Long) POSITION_ACCESSOR.get(deleteRecord));
         }
-        Roaring64Bitmap posBitMap = positionMap.computeIfAbsent(path, k -> new Roaring64Bitmap());
-        posBitMap.add((Long) POSITION_ACCESSOR.get(deleteRecord));
+      } catch (IOException e) {
+        throw new RuntimeException("Failed to close position delete iterator", e);
       }
     }
 

amoro-format-iceberg/src/main/java/org/apache/amoro/io/reader/MixedDeleteFilter.java

@@ -369,19 +369,22 @@ private Predicate<T> applyPosDeletes() {
     if (positionMap == null) {
       positionMap = new HashMap<>();
       List<CloseableIterable<Record>> deletes = Lists.transform(posDeletes, this::openPosDeletes);
-      CloseableIterator<Record> iterator = CloseableIterable.concat(deletes).iterator();
-      while (iterator.hasNext()) {
-        Record deleteRecord = iterator.next();
-        String path = FILENAME_ACCESSOR.get(deleteRecord).toString();
-        if (!pathSets.contains(path)) {
-          continue;
-        }
-        Set<Long> posSet = positionMap.get(path);
-        if (posSet == null) {
-          posSet = new HashSet<>();
-          positionMap.put(path, posSet);
+      try (CloseableIterator<Record> iterator = CloseableIterable.concat(deletes).iterator()) {
+        while (iterator.hasNext()) {
+          Record deleteRecord = iterator.next();
+          String path = FILENAME_ACCESSOR.get(deleteRecord).toString();
+          if (!pathSets.contains(path)) {
+            continue;
+          }
+          Set<Long> posSet = positionMap.get(path);
+          if (posSet == null) {
+            posSet = new HashSet<>();
+            positionMap.put(path, posSet);
+          }
+          posSet.add((Long) POSITION_ACCESSOR.get(deleteRecord));
         }
-        posSet.add((Long) POSITION_ACCESSOR.get(deleteRecord));
+      } catch (IOException e) {
+        throw new RuntimeException("Failed to close position delete iterator", e);
       }
     }