Pipe: Refactor some features for user access

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

@@ -91,7 +91,8 @@ public void releasePipePluginInfoLock() {
 
   /////////////////////////////// Validator ///////////////////////////////
 
-  public void validateBeforeCreatingPipePlugin(String pluginName, String jarName, String jarMD5) {
+  public void validateBeforeCreatingPipePlugin(
+      final String pluginName, final String jarName, final String jarMD5) {
     // both build-in and user defined pipe plugin should be unique
     if (pipePluginMetaKeeper.containsPipePlugin(pluginName)) {
       throw new PipeException(
@@ -108,24 +109,28 @@ public void validateBeforeCreatingPipePlugin(String pluginName, String jarName,
     }
   }
 
-  public void validateBeforeDroppingPipePlugin(String pluginName) {
-    if (pipePluginMetaKeeper.containsPipePlugin(pluginName)
-        && pipePluginMetaKeeper.getPipePluginMeta(pluginName).isBuiltin()) {
+  public void validateBeforeDroppingPipePlugin(final String pluginName) {
+    if (!pipePluginMetaKeeper.containsPipePlugin(pluginName)) {
+      throw new PipeException(
+          String.format(
+              "Failed to drop PipePlugin [%s], this PipePlugin has not been created", pluginName));
+    }
+    if (pipePluginMetaKeeper.getPipePluginMeta(pluginName).isBuiltin()) {
       throw new PipeException(
           String.format(
               "Failed to drop PipePlugin [%s], the PipePlugin is a built-in PipePlugin",
               pluginName));
     }
   }
 
-  public boolean isJarNeededToBeSavedWhenCreatingPipePlugin(String jarName) {
+  public boolean isJarNeededToBeSavedWhenCreatingPipePlugin(final String jarName) {
     return !pipePluginMetaKeeper.containsJar(jarName);
   }
 
   public void checkPipePluginExistence(
-      Map<String, String> extractorAttributes,
-      Map<String, String> processorAttributes,
-      Map<String, String> connectorAttributes) {
+      final Map<String, String> extractorAttributes,
+      final Map<String, String> processorAttributes,
+      final Map<String, String> connectorAttributes) {
     final PipeParameters extractorParameters = new PipeParameters(extractorAttributes);
     final String extractorPluginName =
         extractorParameters.getStringOrDefault(
@@ -170,7 +175,7 @@ public void checkPipePluginExistence(
 
   /////////////////////////////// Pipe Plugin Management ///////////////////////////////
 
-  public TSStatus createPipePlugin(CreatePipePluginPlan createPipePluginPlan) {
+  public TSStatus createPipePlugin(final CreatePipePluginPlan createPipePluginPlan) {
     try {
       final PipePluginMeta pipePluginMeta = createPipePluginPlan.getPipePluginMeta();
 
@@ -188,7 +193,7 @@ public TSStatus createPipePlugin(CreatePipePluginPlan createPipePluginPlan) {
       }
 
       return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
-    } catch (Exception e) {
+    } catch (final Exception e) {
       final String errorMessage =
           String.format(
               "Failed to execute createPipePlugin(%s) on config nodes, because of %s",
@@ -199,7 +204,7 @@ public TSStatus createPipePlugin(CreatePipePluginPlan createPipePluginPlan) {
     }
   }
 
-  public TSStatus dropPipePlugin(DropPipePluginPlan dropPipePluginPlan) {
+  public TSStatus dropPipePlugin(final DropPipePluginPlan dropPipePluginPlan) {
     final String pluginName = dropPipePluginPlan.getPluginName();
 
     if (pipePluginMetaKeeper.containsPipePlugin(pluginName)) {
@@ -217,17 +222,17 @@ public DataSet showPipePlugins() {
         Arrays.asList(pipePluginMetaKeeper.getAllPipePluginMeta()));
   }
 
-  public JarResp getPipePluginJar(GetPipePluginJarPlan getPipePluginJarPlan) {
+  public JarResp getPipePluginJar(final GetPipePluginJarPlan getPipePluginJarPlan) {
     try {
       final List<ByteBuffer> jarList = new ArrayList<>();
-      for (String jarName : getPipePluginJarPlan.getJarNames()) {
+      for (final String jarName : getPipePluginJarPlan.getJarNames()) {
         jarList.add(
             ExecutableManager.transferToBytebuffer(
                 PipePluginExecutableManager.getInstance()
                     .getFileStringUnderInstallByName(jarName)));
       }
       return new JarResp(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()), jarList);
-    } catch (Exception e) {
+    } catch (final Exception e) {
       LOGGER.error("Get PipePlugin_Jar failed", e);
       return new JarResp(
           new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
@@ -239,7 +244,7 @@ public JarResp getPipePluginJar(GetPipePluginJarPlan getPipePluginJarPlan) {
   /////////////////////////////// Snapshot Processor ///////////////////////////////
 
   @Override
-  public boolean processTakeSnapshot(File snapshotDir) throws IOException {
+  public boolean processTakeSnapshot(final File snapshotDir) throws IOException {
     acquirePipePluginInfoLock();
     try {
       final File snapshotFile = new File(snapshotDir, SNAPSHOT_FILE_NAME);
@@ -261,7 +266,7 @@ public boolean processTakeSnapshot(File snapshotDir) throws IOException {
   }
 
   @Override
-  public void processLoadSnapshot(File snapshotDir) throws IOException {
+  public void processLoadSnapshot(final File snapshotDir) throws IOException {
     acquirePipePluginInfoLock();
     try {
       final File snapshotFile = new File(snapshotDir, SNAPSHOT_FILE_NAME);
@@ -288,7 +293,7 @@ public int hashCode() {
   }
 
   @Override
-  public boolean equals(Object obj) {
+  public boolean equals(final Object obj) {
     if (this == obj) {
       return true;
     }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java

@@ -469,7 +469,7 @@ public SettableFuture<ConfigTaskResult> createFunction(
           }
         } catch (IOException | URISyntaxException e) {
           LOGGER.warn(
-              "Failed to get executable for UDF({}) using URI: {}, the cause is: {}",
+              "Failed to get executable for UDF({}) using URI: {}.",
               createFunctionStatement.getUdfName(),
               createFunctionStatement.getUriString(),
               e);
@@ -505,14 +505,14 @@ public SettableFuture<ConfigTaskResult> createFunction(
           | InvocationTargetException
           | ClassCastException e) {
         LOGGER.warn(
-            "Failed to create function when try to create UDF({}) instance first, the cause is: {}",
+            "Failed to create function when try to create UDF({}) instance first.",
             createFunctionStatement.getUdfName(),
             e);
         future.setException(
             new IoTDBException(
                 "Failed to load class '"
                     + createFunctionStatement.getClassName()
-                    + "', because it's not found in jar file: "
+                    + "', because it's not found in jar file or is invalid: "
                     + createFunctionStatement.getUriString(),
                 TSStatusCode.UDF_LOAD_CLASS_ERROR.getStatusCode()));
         return future;
@@ -642,7 +642,7 @@ public SettableFuture<ConfigTaskResult> createTrigger(
           }
         } catch (IOException | URISyntaxException e) {
           LOGGER.warn(
-              "Failed to get executable for Trigger({}) using URI: {}, the cause is: {}",
+              "Failed to get executable for Trigger({}) using URI: {}.",
               createTriggerStatement.getTriggerName(),
               createTriggerStatement.getUriString(),
               e);
@@ -679,14 +679,14 @@ public SettableFuture<ConfigTaskResult> createTrigger(
           | InvocationTargetException
           | ClassCastException e) {
         LOGGER.warn(
-            "Failed to create trigger when try to create trigger({}) instance first, the cause is: {}",
+            "Failed to create trigger when try to create trigger({}) instance first.",
             createTriggerStatement.getTriggerName(),
             e);
         future.setException(
             new IoTDBException(
                 "Failed to load class '"
                     + createTriggerStatement.getClassName()
-                    + "', because it's not found in jar file: "
+                    + "', because it's not found in jar file or is invalid: "
                     + createTriggerStatement.getUriString(),
                 TSStatusCode.TRIGGER_LOAD_CLASS_ERROR.getStatusCode()));
         return future;
@@ -752,7 +752,7 @@ public SettableFuture<ConfigTaskResult> showTriggers() {
 
   @Override
   public SettableFuture<ConfigTaskResult> createPipePlugin(
-      CreatePipePluginStatement createPipePluginStatement) {
+      final CreatePipePluginStatement createPipePluginStatement) {
     final SettableFuture<ConfigTaskResult> future = SettableFuture.create();
     final String pluginName = createPipePluginStatement.getPluginName();
     final String className = createPipePluginStatement.getClassName();
@@ -768,13 +768,13 @@ public SettableFuture<ConfigTaskResult> createPipePlugin(
 
     try (final ConfigNodeClient client =
         CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
-      String libRoot;
-      ByteBuffer jarFile;
-      String jarMd5;
+      final String libRoot;
+      final ByteBuffer jarFile;
+      final String jarMd5;
 
       final String jarFileName = new File(uriString).getName();
       try {
-        URI uri = new URI(uriString);
+        final URI uri = new URI(uriString);
         if (uri.getScheme() == null) {
           future.setException(
               new IoTDBException(
@@ -784,10 +784,10 @@ public SettableFuture<ConfigTaskResult> createPipePlugin(
         }
         if (!uri.getScheme().equals("file")) {
           // Download executable
-          ExecutableResource resource =
+          final ExecutableResource resource =
               PipePluginExecutableManager.getInstance()
                   .request(Collections.singletonList(uriString));
-          String jarFilePathUnderTempDir =
+          final String jarFilePathUnderTempDir =
               PipePluginExecutableManager.getInstance()
                       .getDirStringUnderTempRootByRequestId(resource.getRequestId())
                   + File.separator
@@ -805,9 +805,9 @@ public SettableFuture<ConfigTaskResult> createPipePlugin(
           // Set md5 of the jar file
           jarMd5 = DigestUtils.md5Hex(Files.newInputStream(Paths.get(libRoot)));
         }
-      } catch (IOException | URISyntaxException e) {
+      } catch (final IOException | URISyntaxException e) {
         LOGGER.warn(
-            "Failed to get executable for PipePlugin({}) using URI: {}, the cause is: {}",
+            "Failed to get executable for PipePlugin({}) using URI: {}.",
             createPipePluginStatement.getPluginName(),
             createPipePluginStatement.getUriString(),
             e);
@@ -821,25 +821,26 @@ public SettableFuture<ConfigTaskResult> createPipePlugin(
       }
 
       // try to create instance, this request will fail if creation is not successful
-      try (PipePluginClassLoader classLoader = new PipePluginClassLoader(libRoot)) {
+      try (final PipePluginClassLoader classLoader = new PipePluginClassLoader(libRoot)) {
         // ensure that jar file contains the class and the class is a pipe plugin
-        Class<?> clazz = Class.forName(createPipePluginStatement.getClassName(), true, classLoader);
-        PipePlugin ignored = (PipePlugin) clazz.getDeclaredConstructor().newInstance();
-      } catch (ClassNotFoundException
+        final Class<?> clazz =
+            Class.forName(createPipePluginStatement.getClassName(), true, classLoader);
+        final PipePlugin ignored = (PipePlugin) clazz.getDeclaredConstructor().newInstance();
+      } catch (final ClassNotFoundException
           | NoSuchMethodException
           | InstantiationException
           | IllegalAccessException
           | InvocationTargetException
           | ClassCastException e) {
         LOGGER.warn(
-            "Failed to create function when try to create PipePlugin({}) instance first, the cause is: {}",
+            "Failed to create function when try to create PipePlugin({}) instance first.",
             createPipePluginStatement.getPluginName(),
             e);
         future.setException(
             new IoTDBException(
                 "Failed to load class '"
                     + createPipePluginStatement.getClassName()
-                    + "', because it's not found in jar file: "
+                    + "', because it's not found in jar file or is invalid: "
                     + createPipePluginStatement.getUriString(),
                 TSStatusCode.PIPE_PLUGIN_LOAD_CLASS_ERROR.getStatusCode()));
         return future;
@@ -868,7 +869,7 @@ public SettableFuture<ConfigTaskResult> createPipePlugin(
       } else {
         future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
       }
-    } catch (ClientManagerException | TException | IOException e) {
+    } catch (final ClientManagerException | TException | IOException e) {
       future.setException(e);
     }
     return future;

iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties

@@ -1658,7 +1658,7 @@ data_replication_factor=1
 # The total bytes that all pipe sinks can transfer per second.
 # When given a value less than or equal to 0, it means no limit.
 # default value is -1, which means no limit.
-# Datatype: int
+# Datatype: double
 # pipe_all_sinks_rate_limit_bytes_per_second=-1
 
 ####################

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java

@@ -562,15 +562,14 @@ private void loadPipeProps(Properties properties) {
             properties.getProperty(
                 "two_stage_aggregate_sender_end_points_cache_in_ms",
                 String.valueOf(config.getTwoStageAggregateSenderEndPointsCacheInMs()))));
+  }
 
+  private void loadSubscriptionProps(Properties properties) {
     config.setSubscriptionCacheMemoryUsagePercentage(
         Float.parseFloat(
             properties.getProperty(
                 "subscription_cache_memory_usage_percentage",
                 String.valueOf(config.getSubscriptionCacheMemoryUsagePercentage()))));
-  }
-
-  private void loadSubscriptionProps(Properties properties) {
     config.setSubscriptionSubtaskExecutorMaxThreadNum(
         Integer.parseInt(
             properties.getProperty(

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java

@@ -117,7 +117,7 @@ public abstract class IoTDBConnector implements PipeConnector {
   protected PipeReceiverStatusHandler receiverStatusHandler;
 
   @Override
-  public void validate(PipeParameterValidator validator) throws Exception {
+  public void validate(final PipeParameterValidator validator) throws Exception {
     final PipeParameters parameters = validator.getParameters();
 
     validator.validate(
@@ -232,7 +232,8 @@ public void validate(PipeParameterValidator validator) throws Exception {
   }
 
   @Override
-  public void customize(PipeParameters parameters, PipeConnectorRuntimeConfiguration configuration)
+  public void customize(
+      final PipeParameters parameters, final PipeConnectorRuntimeConfiguration configuration)
       throws Exception {
     nodeUrls.clear();
     nodeUrls.addAll(parseNodeUrls(parameters));
@@ -276,7 +277,7 @@ public void customize(PipeParameters parameters, PipeConnectorRuntimeConfigurati
                 CONNECTOR_EXCEPTION_OTHERS_RECORD_IGNORED_DATA_DEFAULT_VALUE));
   }
 
-  protected LinkedHashSet<TEndPoint> parseNodeUrls(PipeParameters parameters)
+  protected LinkedHashSet<TEndPoint> parseNodeUrls(final PipeParameters parameters)
       throws PipeParameterNotValidException {
     final LinkedHashSet<TEndPoint> givenNodeUrls = new LinkedHashSet<>(nodeUrls);
 
@@ -317,15 +318,22 @@ protected LinkedHashSet<TEndPoint> parseNodeUrls(PipeParameters parameters)
         givenNodeUrls.addAll(
             NodeUrlUtils.parseTEndPointUrls(
                 Arrays.asList(
-                    parameters.getStringByKeys(CONNECTOR_IOTDB_NODE_URLS_KEY).split(","))));
+                    parameters
+                        .getStringByKeys(CONNECTOR_IOTDB_NODE_URLS_KEY)
+                        .replace(" ", "")
+                        .split(","))));
       }
 
       if (parameters.hasAttribute(SINK_IOTDB_NODE_URLS_KEY)) {
         givenNodeUrls.addAll(
             NodeUrlUtils.parseTEndPointUrls(
-                Arrays.asList(parameters.getStringByKeys(SINK_IOTDB_NODE_URLS_KEY).split(","))));
+                Arrays.asList(
+                    parameters
+                        .getStringByKeys(SINK_IOTDB_NODE_URLS_KEY)
+                        .replace(" ", "")
+                        .split(","))));
       }
-    } catch (Exception e) {
+    } catch (final Exception e) {
       LOGGER.warn(PARSE_URL_ERROR_FORMATTER, e.toString());
       throw new PipeParameterNotValidException(PARSE_URL_ERROR_MESSAGE);
     }
@@ -335,8 +343,8 @@ protected LinkedHashSet<TEndPoint> parseNodeUrls(PipeParameters parameters)
     return givenNodeUrls;
   }
 
-  private void checkNodeUrls(Set<TEndPoint> nodeUrls) throws PipeParameterNotValidException {
-    for (TEndPoint nodeUrl : nodeUrls) {
+  private void checkNodeUrls(final Set<TEndPoint> nodeUrls) throws PipeParameterNotValidException {
+    for (final TEndPoint nodeUrl : nodeUrls) {
       if (Objects.isNull(nodeUrl.ip) || nodeUrl.ip.isEmpty()) {
         LOGGER.warn(PARSE_URL_ERROR_FORMATTER, "host cannot be empty");
         throw new PipeParameterNotValidException(PARSE_URL_ERROR_MESSAGE);
@@ -354,13 +362,13 @@ public void close() {
     PIPE_END_POINT_RATE_LIMITER_MAP.clear();
   }
 
-  protected TPipeTransferReq compressIfNeeded(TPipeTransferReq req) throws IOException {
+  protected TPipeTransferReq compressIfNeeded(final TPipeTransferReq req) throws IOException {
     return isRpcCompressionEnabled
         ? PipeTransferCompressedReq.toTPipeTransferReq(req, compressors)
         : req;
   }
 
-  protected byte[] compressIfNeeded(byte[] reqInBytes) throws IOException {
+  protected byte[] compressIfNeeded(final byte[] reqInBytes) throws IOException {
     return isRpcCompressionEnabled
         ? PipeTransferCompressedReq.toTPipeTransferReqBytes(reqInBytes, compressors)
         : reqInBytes;