Pipe: Fixed the hard-link lock problem & Some pipe CIs on master (#16006)

* Update IoTDBPipePermissionIT.java

* refacotr

* revert=pom

* fix-lock

Author: Caideyipi

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipePermissionIT.java

@@ -207,7 +207,11 @@ public void testSourcePermission() {
       return;
     }
 
-    TableModelUtils.createDataBaseAndTable(receiverEnv, "test", "test");
+    try {
+      TableModelUtils.createDataBaseAndTable(receiverEnv, "test", "test");
+    } catch (final Exception ignore) {
+      // Ignore because the db/table may be transferred because sender user may see these
+    }
 
     // Exception, block here
     TableModelUtils.assertCountDataAlwaysOnEnv("test", "test", 0, receiverEnv);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeSinglePipeMetrics.java

@@ -404,18 +404,18 @@ public Pair<Long, Double> getRemainingEventAndTime(
 
   //////////////////////////// singleton ////////////////////////////
 
-  private static class PipeDataNodeRemainingEventAndTimeMetricsHolder {
+  private static class PipeDataNodeSinglePipeMetricsHolder {
 
     private static final PipeDataNodeSinglePipeMetrics INSTANCE =
         new PipeDataNodeSinglePipeMetrics();
 
-    private PipeDataNodeRemainingEventAndTimeMetricsHolder() {
+    private PipeDataNodeSinglePipeMetricsHolder() {
       // Empty constructor
     }
   }
 
   public static PipeDataNodeSinglePipeMetrics getInstance() {
-    return PipeDataNodeRemainingEventAndTimeMetricsHolder.INSTANCE;
+    return PipeDataNodeSinglePipeMetricsHolder.INSTANCE;
   }
 
   private PipeDataNodeSinglePipeMetrics() {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java

@@ -91,30 +91,27 @@ private File increaseFileReference(
       throws IOException {
     // If the file is already a hardlink or copied file,
     // just increase reference count and return it
-    segmentLock.lock(file);
-    try {
-      if (increaseReferenceIfExists(file, pipeName, isTsFile)) {
-        return file;
-      }
-    } finally {
-      segmentLock.unlock(file);
+    if (increaseReferenceIfExists(file, pipeName, isTsFile)) {
+      return file;
     }
 
     // If the file is not a hardlink or copied file, check if there is a related hardlink or
     // copied file in pipe dir. if so, increase reference count and return it
     final File hardlinkOrCopiedFile =
         Objects.isNull(sourceFile) ? getHardlinkOrCopiedFileInPipeDir(file, pipeName) : file;
-    segmentLock.lock(hardlinkOrCopiedFile);
-    try {
-      if (increaseReferenceIfExists(hardlinkOrCopiedFile, pipeName, isTsFile)) {
-        return getResourceMap(pipeName).get(hardlinkOrCopiedFile.getPath()).getFile();
-      }
 
-      // If the file is a tsfile, create a hardlink in pipe dir and will return it.
-      // otherwise, copy the file (.mod or .resource) to pipe dir and will return it.
-      final File source = Objects.isNull(sourceFile) ? file : sourceFile;
+    if (increaseReferenceIfExists(hardlinkOrCopiedFile, pipeName, isTsFile)) {
+      return getResourceMap(pipeName).get(hardlinkOrCopiedFile.getPath()).getFile();
+    }
+
+    // If the file is a tsfile, create a hardlink in pipe dir and will return it.
+    // otherwise, copy the file (.mod or .resource) to pipe dir and will return it.
+    final File source = Objects.isNull(sourceFile) ? file : sourceFile;
+    final File resultFile;
 
-      final File resultFile =
+    segmentLock.lock(hardlinkOrCopiedFile);
+    try {
+      resultFile =
           isTsFile
               ? FileUtils.createHardLink(source, hardlinkOrCopiedFile)
               : FileUtils.copyFile(source, hardlinkOrCopiedFile);
@@ -130,25 +127,29 @@ private File increaseFileReference(
         hardlinkOrCopiedFileToTsFilePublicResourceMap.put(
             resultFile.getPath(), new PipeTsFilePublicResource(resultFile));
       }
-
-      increasePublicReference(resultFile, pipeName, isTsFile);
-
-      return resultFile;
     } finally {
       segmentLock.unlock(hardlinkOrCopiedFile);
     }
+    increasePublicReference(resultFile, pipeName, isTsFile);
+    return resultFile;
   }
 
   private boolean increaseReferenceIfExists(
       final File file, final @Nullable String pipeName, final boolean isTsFile) throws IOException {
-    final String path = file.getPath();
-    final PipeTsFileResource resource = getResourceMap(pipeName).get(path);
-    if (resource != null) {
-      resource.increaseReferenceCount();
-      increasePublicReference(file, pipeName, isTsFile);
-      return true;
+    segmentLock.lock(file);
+    try {
+      final String path = file.getPath();
+      final PipeTsFileResource resource = getResourceMap(pipeName).get(path);
+      if (resource != null) {
+        resource.increaseReferenceCount();
+      } else {
+        return false;
+      }
+    } finally {
+      segmentLock.unlock(file);
     }
-    return false;
+    increasePublicReference(file, pipeName, isTsFile);
+    return true;
   }
 
   private void increasePublicReference(
@@ -221,12 +222,13 @@ public void decreaseFileReference(
       if (resource != null && resource.decreaseReferenceCount()) {
         getResourceMap(pipeName).remove(filePath);
       }
-      // Decrease the assigner's file to clear hard-link and memory cache
-      // Note that it does not exist for historical files
-      decreasePublicReferenceIfExists(hardlinkOrCopiedFile, pipeName);
     } finally {
       segmentLock.unlock(hardlinkOrCopiedFile);
     }
+
+    // Decrease the assigner's file to clear hard-link and memory cache
+    // Note that it does not exist for historical files
+    decreasePublicReferenceIfExists(hardlinkOrCopiedFile, pipeName);
   }
 
   private void decreasePublicReferenceIfExists(final File file, final @Nullable String pipeName) {