Pipe: fix file handle leak when processing with extractor.pattern

iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipeConnector.java

@@ -128,9 +128,13 @@ void customize(PipeParameters parameters, PipeConnectorRuntimeConfiguration conf
    * @throws Exception the user can throw errors if necessary
    */
   default void transfer(TsFileInsertionEvent tsFileInsertionEvent) throws Exception {
-    for (final TabletInsertionEvent tabletInsertionEvent :
-        tsFileInsertionEvent.toTabletInsertionEvents()) {
-      transfer(tabletInsertionEvent);
+    try {
+      for (final TabletInsertionEvent tabletInsertionEvent :
+          tsFileInsertionEvent.toTabletInsertionEvents()) {
+        transfer(tabletInsertionEvent);
+      }
+    } finally {
+      tsFileInsertionEvent.close();
     }
   }
 

iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipeProcessor.java

@@ -105,9 +105,13 @@ void process(TabletInsertionEvent tabletInsertionEvent, EventCollector eventColl
    */
   default void process(TsFileInsertionEvent tsFileInsertionEvent, EventCollector eventCollector)
       throws Exception {
-    for (final TabletInsertionEvent tabletInsertionEvent :
-        tsFileInsertionEvent.toTabletInsertionEvents()) {
-      process(tabletInsertionEvent, eventCollector);
+    try {
+      for (final TabletInsertionEvent tabletInsertionEvent :
+          tsFileInsertionEvent.toTabletInsertionEvents()) {
+        process(tabletInsertionEvent, eventCollector);
+      }
+    } finally {
+      tsFileInsertionEvent.close();
     }
   }
 

iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/event/dml/insertion/TsFileInsertionEvent.java

@@ -25,7 +25,7 @@
  * {@link TsFileInsertionEvent} is used to define the event of writing TsFile. Event data stores in
  * disks, which is compressed and encoded, and requires IO cost for computational processing.
  */
-public interface TsFileInsertionEvent extends Event {
+public interface TsFileInsertionEvent extends Event, AutoCloseable {
 
   /**
    * The method is used to convert the TsFileInsertionEvent into several TabletInsertionEvents.

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBAirGapConnector.java

@@ -237,8 +237,12 @@ public void transfer(TsFileInsertionEvent tsFileInsertionEvent) throws Exception
     }
 
     if (((EnrichedEvent) tsFileInsertionEvent).shouldParsePatternOrTime()) {
-      for (final TabletInsertionEvent event : tsFileInsertionEvent.toTabletInsertionEvents()) {
-        transfer(event);
+      try {
+        for (final TabletInsertionEvent event : tsFileInsertionEvent.toTabletInsertionEvents()) {
+          transfer(event);
+        }
+      } finally {
+        tsFileInsertionEvent.close();
       }
       return;
     }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncConnector.java

@@ -292,8 +292,12 @@ public void transfer(TsFileInsertionEvent tsFileInsertionEvent) throws Exception
     }
 
     if (((EnrichedEvent) tsFileInsertionEvent).shouldParsePatternOrTime()) {
-      for (final TabletInsertionEvent event : tsFileInsertionEvent.toTabletInsertionEvents()) {
-        transfer(event);
+      try {
+        for (final TabletInsertionEvent event : tsFileInsertionEvent.toTabletInsertionEvents()) {
+          transfer(event);
+        }
+      } finally {
+        tsFileInsertionEvent.close();
       }
       return;
     }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/websocket/WebSocketConnector.java

@@ -117,10 +117,14 @@ public void transfer(TsFileInsertionEvent tsFileInsertionEvent) throws Exception
           tsFileInsertionEvent);
       return;
     }
-    for (TabletInsertionEvent event : tsFileInsertionEvent.toTabletInsertionEvents()) {
-      long commitId = commitIdGenerator.incrementAndGet();
-      ((EnrichedEvent) event).increaseReferenceCount(WebSocketConnector.class.getName());
-      server.addEvent(new Pair<>(commitId, event));
+    try {
+      for (TabletInsertionEvent event : tsFileInsertionEvent.toTabletInsertionEvents()) {
+        long commitId = commitIdGenerator.incrementAndGet();
+        ((EnrichedEvent) event).increaseReferenceCount(WebSocketConnector.class.getName());
+        server.addEvent(new Pair<>(commitId, event));
+      }
+    } finally {
+      tsFileInsertionEvent.close();
     }
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java

@@ -185,19 +185,32 @@ public Iterable<TabletInsertionEvent> toTabletInsertionEvents() {
       }
       return dataContainer.toTabletInsertionEvents();
     } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      close();
+
       final String errorMsg =
           String.format(
               "Interrupted when waiting for closing TsFile %s.", resource.getTsFilePath());
       LOGGER.warn(errorMsg, e);
-      Thread.currentThread().interrupt();
       throw new PipeException(errorMsg);
     } catch (IOException e) {
+      close();
+
       final String errorMsg = String.format("Read TsFile %s error.", resource.getTsFilePath());
       LOGGER.warn(errorMsg, e);
       throw new PipeException(errorMsg);
     }
   }
 
+  /** Release the resource of data container. */
+  @Override
+  public void close() {
+    if (dataContainer != null) {
+      dataContainer.close();
+      dataContainer = null;
+    }
+  }
+
   /////////////////////////// Object ///////////////////////////
 
   @Override

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionDataContainer.java

@@ -163,6 +163,7 @@ public Iterable<TabletInsertionEvent> toTabletInsertionEvents() {
           public boolean hasNext() {
             while (tabletIterator == null || !tabletIterator.hasNext()) {
               if (!deviceMeasurementsMapIterator.hasNext()) {
+                close();
                 return false;
               }