Pipe: Fixed the hardlink bug of plugin meta (#16937)

Author: Caideyipi

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipePluginInfo.java

@@ -215,16 +215,8 @@ public TSStatus createPipePlugin(final CreatePipePluginPlan createPipePluginPlan
       final String className = pipePluginMeta.getClassName();
       final String jarName = pipePluginMeta.getJarName();
 
-      // try to drop the old pipe plugin if exists to reduce the effect of the inconsistency
-      dropPipePlugin(new DropPipePluginPlan(pluginName));
-
-      pipePluginMetaKeeper.addPipePluginMeta(pluginName, pipePluginMeta);
-      pipePluginMetaKeeper.addJarNameAndMd5(jarName, pipePluginMeta.getJarMD5());
-
       if (createPipePluginPlan.getJarFile() != null) {
-        pipePluginExecutableManager.savePluginToInstallDir(
-            ByteBuffer.wrap(createPipePluginPlan.getJarFile().getValues()), pluginName, jarName);
-        computeFromPluginClass(pluginName, className);
+        savePipePluginWithRollback(createPipePluginPlan);
       } else {
         final String existed = pipePluginMetaKeeper.getPluginNameByJarName(jarName);
         if (Objects.nonNull(existed)) {
@@ -238,6 +230,12 @@ public TSStatus createPipePlugin(final CreatePipePluginPlan createPipePluginPlan
         }
       }
 
+      // try to drop the old pipe plugin if exists to reduce the effect of the inconsistency
+      dropPipePlugin(new DropPipePluginPlan(pluginName));
+
+      pipePluginMetaKeeper.addPipePluginMeta(pluginName, pipePluginMeta);
+      pipePluginMetaKeeper.addJarNameAndMd5(jarName, pipePluginMeta.getJarMD5());
+
       return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
     } catch (final Exception e) {
       final String errorMessage =
@@ -250,6 +248,23 @@ public TSStatus createPipePlugin(final CreatePipePluginPlan createPipePluginPlan
     }
   }
 
+  private void savePipePluginWithRollback(final CreatePipePluginPlan createPipePluginPlan)
+      throws Exception {
+    final PipePluginMeta pipePluginMeta = createPipePluginPlan.getPipePluginMeta();
+    final String pluginName = pipePluginMeta.getPluginName();
+    final String className = pipePluginMeta.getClassName();
+    final String jarName = pipePluginMeta.getJarName();
+    try {
+      pipePluginExecutableManager.savePluginToInstallDir(
+          ByteBuffer.wrap(createPipePluginPlan.getJarFile().getValues()), pluginName, jarName);
+      computeFromPluginClass(pluginName, className);
+    } catch (final Exception e) {
+      // We need to rollback if the creation has failed
+      pipePluginExecutableManager.removePluginFileUnderLibRoot(pluginName, jarName);
+      throw e;
+    }
+  }
+
   private void computeFromPluginClass(final String pluginName, final String className)
       throws Exception {
     final String pluginDirPath = pipePluginExecutableManager.getPluginsDirPath(pluginName);

iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/PipeInfoTest.java

@@ -31,6 +31,8 @@
 import org.apache.iotdb.confignode.consensus.request.write.pipe.task.DropPipePlanV2;
 import org.apache.iotdb.confignode.consensus.request.write.pipe.task.SetPipeStatusPlanV2;
 import org.apache.iotdb.confignode.persistence.pipe.PipeInfo;
+import org.apache.iotdb.pipe.api.exception.PipeException;
+import org.apache.iotdb.rpc.TSStatusCode;
 
 import org.apache.thrift.TException;
 import org.apache.tsfile.common.conf.TSFileConfig;
@@ -149,10 +151,18 @@ public void testManagement() {
         new CreatePipePluginPlan(
             new PipePluginMeta(pluginName, "org.apache.iotdb.TestJar", false, "test.jar", "???"),
             new Binary("123", TSFileConfig.STRING_CHARSET));
-    pipeInfo.getPipePluginInfo().createPipePlugin(createPipePluginPlan);
 
-    // Drop pipe plugin test plugin
-    pipeInfo.getPipePluginInfo().validateBeforeDroppingPipePlugin(pluginName, false);
+    // Shall fail due to validation
+    Assert.assertEquals(
+        TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode(),
+        pipeInfo.getPipePluginInfo().createPipePlugin(createPipePluginPlan).getCode());
+
+    // Drop pipe plugin test plugin, validation failure
+    Assert.assertThrows(
+        PipeException.class,
+        () -> pipeInfo.getPipePluginInfo().validateBeforeDroppingPipePlugin(pluginName, false));
+
+    // Idempotent
     DropPipePluginPlan dropPipePluginPlan = new DropPipePluginPlan(pluginName);
     pipeInfo.getPipePluginInfo().dropPipePlugin(dropPipePluginPlan);
   }