[Pipe] Allow startup when a plugin JAR is unavailable

iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java

@@ -140,6 +140,25 @@ public final class DataNodePipeMessages {
       "Failed to get pipe metas, will be synced by configNode later...";
   public static final String FAILED_TO_GET_PIPE_PLUGIN_JAR_FROM =
       "Failed to get pipe plugin jar from config node.";
+  public static final String
+      LOG_FAILED_TO_FETCH_PIPE_PLUGIN_JARS_FROM_CONFIGNODE_PLUGINS_ARG_JARS_ARG_STATUS_ARG_RETRYING_EACH_PLUGIN_INDIVIDUALLY_574C0077 =
+          "Failed to fetch pipe plugin jars from ConfigNode. Plugins: {}, jars: {}, status: {}. "
+              + "Retrying each plugin individually.";
+  public static final String
+      LOG_CONFIGNODE_RETURNED_ARG_PIPE_PLUGIN_JARS_FOR_ARG_REQUESTED_PLUGINS_PLUGINS_ARG_JARS_ARG_RETRYING_EACH_PLUGIN_INDIVIDUALLY_27E32FDE =
+          "ConfigNode returned {} pipe plugin jars for {} requested plugins. Plugins: {}, jars: {}. "
+              + "Retrying each plugin individually.";
+  public static final String
+      EXCEPTION_FAILED_TO_FETCH_PIPE_PLUGIN_JAR_FROM_CONFIGNODE_FOR_PLUGIN_ARG_JAR_ARG_STATUS_ARG_B7C7FDE5 =
+          "Failed to fetch pipe plugin jar from ConfigNode for plugin %s (jar %s). Status: %s.";
+  public static final String
+      EXCEPTION_CONFIGNODE_RETURNED_ARG_JARS_FOR_PIPE_PLUGIN_ARG_WHILE_ONE_WAS_REQUESTED_A724E582 =
+          "ConfigNode returned %d jars for pipe plugin %s while one was requested.";
+  public static final String
+      LOG_FAILED_TO_FETCH_PIPE_PLUGIN_JAR_ARG_FOR_PIPE_PLUGIN_ARG_FROM_CONFIGNODE_4929C5D9 =
+          "Failed to fetch pipe plugin jar {} for pipe plugin {} from ConfigNode.";
+  public static final String LOG_FAILED_TO_SAVE_JAR_ARG_FOR_PIPE_PLUGIN_ARG_A64D1530 =
+      "Failed to save jar {} for pipe plugin {}.";
   public static final String FAILED_TO_GET_PIPE_TASK_META_FROM =
       "Failed to get pipe task meta from config node. Ignore the exception, because config "
           + "node may not be ready yet, and meta will be pushed by config node later.";

iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java

@@ -130,6 +130,23 @@ public final class DataNodePipeMessages {
       "获取 pipe metas 失败，稍后会从 CN 处再次同步。";
   public static final String FAILED_TO_GET_PIPE_PLUGIN_JAR_FROM =
       "从 CN 处获取 pipe 插件 jar 包失败。";
+  public static final String
+      LOG_FAILED_TO_FETCH_PIPE_PLUGIN_JARS_FROM_CONFIGNODE_PLUGINS_ARG_JARS_ARG_STATUS_ARG_RETRYING_EACH_PLUGIN_INDIVIDUALLY_574C0077 =
+          "从 ConfigNode 获取 pipe plugin jars 失败。插件：{}，jars：{}，状态：{}。将逐个重试每个插件。";
+  public static final String
+      LOG_CONFIGNODE_RETURNED_ARG_PIPE_PLUGIN_JARS_FOR_ARG_REQUESTED_PLUGINS_PLUGINS_ARG_JARS_ARG_RETRYING_EACH_PLUGIN_INDIVIDUALLY_27E32FDE =
+          "ConfigNode 为 {} 个请求的插件返回了 {} 个 pipe plugin jars。插件：{}，jars：{}。将逐个重试每个插件。";
+  public static final String
+      EXCEPTION_FAILED_TO_FETCH_PIPE_PLUGIN_JAR_FROM_CONFIGNODE_FOR_PLUGIN_ARG_JAR_ARG_STATUS_ARG_B7C7FDE5 =
+          "从 ConfigNode 获取插件 %s（jar %s）的 pipe plugin jar 失败。状态：%s。";
+  public static final String
+      EXCEPTION_CONFIGNODE_RETURNED_ARG_JARS_FOR_PIPE_PLUGIN_ARG_WHILE_ONE_WAS_REQUESTED_A724E582 =
+          "ConfigNode 返回了 %d 个 jars，而 pipe plugin %s 只请求了一个。";
+  public static final String
+      LOG_FAILED_TO_FETCH_PIPE_PLUGIN_JAR_ARG_FOR_PIPE_PLUGIN_ARG_FROM_CONFIGNODE_4929C5D9 =
+          "从 ConfigNode 获取 pipe plugin jar {}（pipe plugin {}）失败。";
+  public static final String LOG_FAILED_TO_SAVE_JAR_ARG_FOR_PIPE_PLUGIN_ARG_A64D1530 =
+      "保存 jar {}（pipe plugin {}）失败。";
   public static final String FAILED_TO_GET_PIPE_TASK_META_FROM =
       "获取 pipe task meta from config node. Ignore the exception 失败，原因：config node may not be "
           + "ready yet, and meta will be pushed by config node later.";

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeAgentLauncher.java

@@ -19,7 +19,6 @@
 
 package org.apache.iotdb.db.pipe.agent.runtime;
 
-import org.apache.iotdb.commons.client.exception.ClientManagerException;
 import org.apache.iotdb.commons.exception.StartupException;
 import org.apache.iotdb.commons.pipe.agent.plugin.meta.PipePluginMeta;
 import org.apache.iotdb.commons.pipe.agent.plugin.service.PipePluginClassLoaderManager;
@@ -39,14 +38,16 @@
 import org.apache.iotdb.pipe.api.exception.PipeException;
 import org.apache.iotdb.rpc.TSStatusCode;
 
-import org.apache.thrift.TException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 class PipeAgentLauncher {
@@ -70,6 +71,7 @@ public static synchronized void launchPipePluginAgent(
 
     final List<PipePluginMeta> uninstalledOrConflictedPipePluginMetaList =
         getUninstalledOrConflictedPipePluginMetaList(resourcesInformationHolder);
+    final Set<String> unavailablePipePluginNameSet = new HashSet<>();
     int index = 0;
     while (index < uninstalledOrConflictedPipePluginMetaList.size()) {
       List<PipePluginMeta> curList = new ArrayList<>();
@@ -80,20 +82,23 @@ public static synchronized void launchPipePluginAgent(
         offset++;
       }
       index += offset;
-      fetchAndSavePipePluginJars(curList);
+      unavailablePipePluginNameSet.addAll(fetchAndSavePipePluginJars(curList));
     }
 
     // create instances of pipe plugins and do registration
     for (PipePluginMeta meta : resourcesInformationHolder.getPipePluginMetaList()) {
       if (meta.isBuiltin()) {
         continue;
       }
+      if (unavailablePipePluginNameSet.contains(meta.getPluginName())) {
+        continue;
+      }
       try {
         PipeDataNodeAgent.plugin().doRegister(meta);
       } catch (Throwable e) {
         PipeDataNodeAgent.plugin().markPluginLoadFailure(meta, e);
         // Ignore a single broken plugin and continue startup.
-        LOGGER.warn(
+        LOGGER.error(
             DataNodePipeMessages.FAILURE_WHEN_REGISTER_PIPE_PLUGIN_SKIP_THIS,
             meta.getPluginName(),
             e);
@@ -137,28 +142,152 @@ private static List<PipePluginMeta> getUninstalledOrConflictedPipePluginMetaList
     return pipePluginMetaList;
   }
 
-  private static void fetchAndSavePipePluginJars(List<PipePluginMeta> pipePluginMetaList)
-      throws StartupException {
+  static Set<String> fetchAndSavePipePluginJars(List<PipePluginMeta> pipePluginMetaList) {
+    if (pipePluginMetaList.isEmpty()) {
+      return Collections.emptySet();
+    }
+
+    final List<String> pluginNameList =
+        pipePluginMetaList.stream().map(PipePluginMeta::getPluginName).collect(Collectors.toList());
+    final List<String> jarNameList =
+        pipePluginMetaList.stream().map(PipePluginMeta::getJarName).collect(Collectors.toList());
+    final TGetJarInListResp resp;
+
     try (ConfigNodeClient configNodeClient =
         ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
-      final List<String> jarNameList =
-          pipePluginMetaList.stream().map(PipePluginMeta::getJarName).collect(Collectors.toList());
-      final TGetJarInListResp resp =
-          configNodeClient.getPipePluginJar(new TGetJarInListReq(jarNameList));
-      if (resp.getStatus().getCode() == TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()) {
-        throw new StartupException(DataNodePipeMessages.FAILED_TO_GET_PIPE_PLUGIN_JAR_FROM);
+      resp = configNodeClient.getPipePluginJar(new TGetJarInListReq(jarNameList));
+    } catch (Exception e) {
+      LOGGER.error(
+          DataNodePipeMessages
+              .LOG_FAILED_TO_FETCH_PIPE_PLUGIN_JARS_FROM_CONFIGNODE_PLUGINS_ARG_JARS_ARG_STATUS_ARG_RETRYING_EACH_PLUGIN_INDIVIDUALLY_574C0077,
+          pluginNameList,
+          jarNameList,
+          null,
+          e);
+      return fetchAndSavePipePluginJarsIndividually(pipePluginMetaList);
+    }
+
+    if (resp == null
+        || resp.getStatus() == null
+        || resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+      LOGGER.error(
+          DataNodePipeMessages
+              .LOG_FAILED_TO_FETCH_PIPE_PLUGIN_JARS_FROM_CONFIGNODE_PLUGINS_ARG_JARS_ARG_STATUS_ARG_RETRYING_EACH_PLUGIN_INDIVIDUALLY_574C0077,
+          pluginNameList,
+          jarNameList,
+          resp == null ? null : resp.getStatus());
+      return fetchAndSavePipePluginJarsIndividually(pipePluginMetaList);
+    }
+
+    final List<ByteBuffer> jarList = resp.getJarList();
+    if (jarList == null || jarList.size() != pipePluginMetaList.size()) {
+      LOGGER.error(
+          DataNodePipeMessages
+              .LOG_CONFIGNODE_RETURNED_ARG_PIPE_PLUGIN_JARS_FOR_ARG_REQUESTED_PLUGINS_PLUGINS_ARG_JARS_ARG_RETRYING_EACH_PLUGIN_INDIVIDUALLY_27E32FDE,
+          jarList == null ? 0 : jarList.size(),
+          pipePluginMetaList.size(),
+          pluginNameList,
+          jarNameList);
+      return fetchAndSavePipePluginJarsIndividually(pipePluginMetaList);
+    }
+
+    return savePipePluginJars(pipePluginMetaList, jarList);
+  }
+
+  private static Set<String> fetchAndSavePipePluginJarsIndividually(
+      List<PipePluginMeta> pipePluginMetaList) {
+    final Set<String> unavailablePipePluginNameSet = new HashSet<>();
+    for (PipePluginMeta pipePluginMeta : pipePluginMetaList) {
+      if (!fetchAndSavePipePluginJarIndividually(pipePluginMeta)) {
+        unavailablePipePluginNameSet.add(pipePluginMeta.getPluginName());
       }
-      final List<ByteBuffer> jarList = resp.getJarList();
-      for (int i = 0; i < pipePluginMetaList.size(); i++) {
+    }
+    return unavailablePipePluginNameSet;
+  }
+
+  private static boolean fetchAndSavePipePluginJarIndividually(PipePluginMeta pipePluginMeta) {
+    final String pluginName = pipePluginMeta.getPluginName();
+    final String jarName = pipePluginMeta.getJarName();
+    final TGetJarInListResp resp;
+    try (ConfigNodeClient configNodeClient =
+        ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
+      resp = configNodeClient.getPipePluginJar(new TGetJarInListReq(List.of(jarName)));
+    } catch (Exception e) {
+      PipeDataNodeAgent.plugin().markPluginLoadFailure(pipePluginMeta, e);
+      LOGGER.error(
+          DataNodePipeMessages
+              .LOG_FAILED_TO_FETCH_PIPE_PLUGIN_JAR_ARG_FOR_PIPE_PLUGIN_ARG_FROM_CONFIGNODE_4929C5D9,
+          jarName,
+          pluginName,
+          e);
+      return false;
+    }
+
+    if (resp == null
+        || resp.getStatus() == null
+        || resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+      final PipeException exception =
+          new PipeException(
+              String.format(
+                  DataNodePipeMessages
+                      .EXCEPTION_FAILED_TO_FETCH_PIPE_PLUGIN_JAR_FROM_CONFIGNODE_FOR_PLUGIN_ARG_JAR_ARG_STATUS_ARG_B7C7FDE5,
+                  pluginName,
+                  jarName,
+                  resp == null ? null : resp.getStatus()));
+      PipeDataNodeAgent.plugin().markPluginLoadFailure(pipePluginMeta, exception);
+      LOGGER.error(exception.getMessage(), exception);
+      return false;
+    }
+
+    final List<ByteBuffer> jarList = resp.getJarList();
+    if (jarList == null || jarList.size() != 1) {
+      final PipeException exception =
+          new PipeException(
+              String.format(
+                  DataNodePipeMessages
+                      .EXCEPTION_CONFIGNODE_RETURNED_ARG_JARS_FOR_PIPE_PLUGIN_ARG_WHILE_ONE_WAS_REQUESTED_A724E582,
+                  jarList == null ? 0 : jarList.size(),
+                  pluginName));
+      PipeDataNodeAgent.plugin().markPluginLoadFailure(pipePluginMeta, exception);
+      LOGGER.error(exception.getMessage(), exception);
+      return false;
+    }
+
+    try {
+      PipePluginExecutableManager.getInstance()
+          .savePluginToInstallDir(jarList.get(0), pluginName, jarName);
+      return true;
+    } catch (Exception e) {
+      PipeDataNodeAgent.plugin().markPluginLoadFailure(pipePluginMeta, e);
+      LOGGER.error(
+          DataNodePipeMessages.LOG_FAILED_TO_SAVE_JAR_ARG_FOR_PIPE_PLUGIN_ARG_A64D1530,
+          jarName,
+          pluginName,
+          e);
+      return false;
+    }
+  }
+
+  private static Set<String> savePipePluginJars(
+      List<PipePluginMeta> pipePluginMetaList, List<ByteBuffer> jarList) {
+    final Set<String> unavailablePipePluginNameSet = new HashSet<>();
+    for (int i = 0; i < pipePluginMetaList.size(); i++) {
+      final PipePluginMeta pipePluginMeta = pipePluginMetaList.get(i);
+      try {
         PipePluginExecutableManager.getInstance()
             .savePluginToInstallDir(
-                jarList.get(i),
-                pipePluginMetaList.get(i).getPluginName(),
-                pipePluginMetaList.get(i).getJarName());
+                jarList.get(i), pipePluginMeta.getPluginName(), pipePluginMeta.getJarName());
+      } catch (Exception e) {
+        PipeDataNodeAgent.plugin().markPluginLoadFailure(pipePluginMeta, e);
+        LOGGER.error(
+            DataNodePipeMessages.LOG_FAILED_TO_SAVE_JAR_ARG_FOR_PIPE_PLUGIN_ARG_A64D1530,
+            pipePluginMeta.getJarName(),
+            pipePluginMeta.getPluginName(),
+            e);
+        unavailablePipePluginNameSet.add(pipePluginMeta.getPluginName());
       }
-    } catch (IOException | TException | ClientManagerException e) {
-      throw new StartupException(e);
     }
+    return unavailablePipePluginNameSet;
   }
 
   public static synchronized void launchPipeTaskAgent() {

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/plugin/PipeDataNodePluginAgentTest.java

@@ -20,12 +20,14 @@
 package org.apache.iotdb.db.pipe.agent.plugin;
 
 import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
+import org.apache.iotdb.commons.pipe.agent.plugin.meta.DataNodePipePluginMetaKeeper;
 import org.apache.iotdb.commons.pipe.agent.plugin.meta.PipePluginMeta;
 import org.apache.iotdb.commons.pipe.agent.plugin.service.PipePluginClassLoaderManager;
 import org.apache.iotdb.commons.pipe.agent.plugin.service.PipePluginExecutableManager;
 import org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant;
 import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant;
 import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
+import org.apache.iotdb.commons.pipe.datastructure.visibility.Visibility;
 import org.apache.iotdb.db.pipe.processor.iotconsensusv2.IoTConsensusV2Processor;
 import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.IoTConsensusV2AsyncSink;
 import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink;
@@ -40,6 +42,7 @@
 import org.junit.Test;
 
 import java.io.IOException;
+import java.lang.reflect.Field;
 import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.util.HashMap;
@@ -180,4 +183,25 @@ public void testPipePluginAgent() {
                     }))
             .getClass());
   }
+
+  @Test
+  public void testPluginLoadFailureIsRecordedForShowPipePlugins() throws Exception {
+    final PipeDataNodePluginAgent agent = new PipeDataNodePluginAgent();
+    final PipePluginMeta plugin =
+        new PipePluginMeta("failed", "test.class", false, "failed.jar", "test-md5");
+
+    agent.markPluginLoadFailure(plugin, new IOException("missing jar"));
+
+    final Field metaKeeperField =
+        PipeDataNodePluginAgent.class.getDeclaredField("pipePluginMetaKeeper");
+    metaKeeperField.setAccessible(true);
+    final DataNodePipePluginMetaKeeper metaKeeper =
+        (DataNodePipePluginMetaKeeper) metaKeeperField.get(agent);
+    final PipePluginMeta recordedPlugin = metaKeeper.getPipePluginMeta("FAILED");
+
+    Assert.assertEquals(
+        "IOException: missing jar", recordedPlugin.getPluginLoadingExceptionMessage());
+    Assert.assertEquals(
+        Visibility.BOTH, metaKeeper.getPipePluginNameToVisibilityMap().get("FAILED"));
+  }
 }