(iotdb) branch master updated: Pipe: Fix batch type can not be changed (Introduced in #15377) & Enhance close() in PipeTsFileResource (#15401)

[rong]

This is an automated email from the ASF dual-hosted git repository.

rong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/master by this push:
     new dd429d9539f Pipe: Fix batch type can not be changed (Introduced in 
#15377) & Enhance close() in PipeTsFileResource (#15401)
dd429d9539f is described below

commit dd429d9539f93c0c0f800a0e0aa73aa9b3a5510e
Author: Steve Yurong Su <r...@apache.org>
AuthorDate: Thu Apr 24 14:46:50 2025 +0800

    Pipe: Fix batch type can not be changed (Introduced in #15377) & Enhance 
close() in PipeTsFileResource (#15401)
---
 .../evolvable/batch/PipeTransferBatchReqBuilder.java       |  5 +++--
 .../iotdb/db/pipe/resource/tsfile/PipeTsFileResource.java  | 14 +++++++++++---
 .../db/pipe/resource/tsfile/PipeTsFileResourceManager.java |  2 +-
 3 files changed, 15 insertions(+), 6 deletions(-)

diff --git 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/batch/PipeTransferBatchReqBuilder.java
 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/batch/PipeTransferBatchReqBuilder.java
index 7d8162b0df2..50160ba61f6 100644
--- 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/batch/PipeTransferBatchReqBuilder.java
+++ 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/batch/PipeTransferBatchReqBuilder.java
@@ -104,7 +104,6 @@ public class PipeTransferBatchReqBuilder implements 
AutoCloseable {
       requestMaxDelayInMs =
           requestMaxDelayInMillis < 0 ? Integer.MAX_VALUE : 
requestMaxDelayInMillis;
     }
-
     requestMaxBatchSizeInBytes =
         parameters.getLongOrDefault(
             Arrays.asList(CONNECTOR_IOTDB_BATCH_SIZE_KEY, 
SINK_IOTDB_BATCH_SIZE_KEY),
@@ -112,7 +111,9 @@ public class PipeTransferBatchReqBuilder implements 
AutoCloseable {
                 ? CONNECTOR_IOTDB_TS_FILE_BATCH_SIZE_DEFAULT_VALUE
                 : CONNECTOR_IOTDB_PLAIN_BATCH_SIZE_DEFAULT_VALUE);
     this.defaultBatch =
-        new PipeTabletEventTsFileBatch(requestMaxDelayInMs, 
requestMaxBatchSizeInBytes);
+        usingTsFileBatch
+            ? new PipeTabletEventTsFileBatch(requestMaxDelayInMs, 
requestMaxBatchSizeInBytes)
+            : new PipeTabletEventPlainBatch(requestMaxDelayInMs, 
requestMaxBatchSizeInBytes);
   }
 
   /**
diff --git 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResource.java
 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResource.java
index dd3ca9c5bc0..bf789b9732d 100644
--- 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResource.java
+++ 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResource.java
@@ -122,7 +122,7 @@ public class PipeTsFileResource implements AutoCloseable {
     return finalReferenceCount;
   }
 
-  public synchronized boolean closeIfOutOfTimeToLive() throws IOException {
+  public synchronized boolean closeIfOutOfTimeToLive() {
     if (referenceCount.get() <= 0
         && (deviceMeasurementsMap == null // Not cached yet.
             || System.currentTimeMillis() - lastUnpinToZeroTime.get()
@@ -135,7 +135,7 @@ public class PipeTsFileResource implements AutoCloseable {
   }
 
   @Override
-  public synchronized void close() throws IOException {
+  public synchronized void close() {
     if (deviceMeasurementsMap != null) {
       deviceMeasurementsMap = null;
     }
@@ -153,7 +153,15 @@ public class PipeTsFileResource implements AutoCloseable {
       allocatedMemoryBlock = null;
     }
 
-    Files.deleteIfExists(hardlinkOrCopiedFile.toPath());
+    try {
+      Files.deleteIfExists(hardlinkOrCopiedFile.toPath());
+    } catch (final Exception e) {
+      LOGGER.error(
+          "PipeTsFileResource: Failed to delete tsfile {} when closing, 
because {}. Please MANUALLY delete it.",
+          hardlinkOrCopiedFile,
+          e.getMessage(),
+          e);
+    }
 
     LOGGER.info("PipeTsFileResource: Closed tsfile {} and cleaned up.", 
hardlinkOrCopiedFile);
   }
diff --git 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java
 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java
index d851b58985c..6d04e909cb1 100644
--- 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java
+++ 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java
@@ -103,7 +103,7 @@ public class PipeTsFileResourceManager {
                   entry.getValue().getReferenceCount(),
                   entry.getValue().getFileSize()));
         }
-      } catch (final IOException e) {
+      } catch (final Exception e) {
         LOGGER.warn("failed to close PipeTsFileResource when checking TTL: ", 
e);
       } finally {
         segmentLock.unlock(new File(hardlinkOrCopiedFile));