This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 02f8d5c0756 [Pipe] Allow startup when a plugin JAR is unavailable
(#18496) (#18600)
02f8d5c0756 is described below
commit 02f8d5c07567aabb9c5911ffe61354fe7b708c8b
Author: Caideyipi <[email protected]>
AuthorDate: Tue Sep 8 17:13:15 2026 +0800
[Pipe] Allow startup when a plugin JAR is unavailable (#18496) (#18600)
* fix(pipe): allow startup when plugin jar is unavailable
* fix(pipe): isolate unavailable plugin jars during startup
* Update PipeAgentLauncherTest.java
* test(pipe): move plugin load failure test out of PowerMock
* fix(pipe): address review comments
(cherry picked from commit 8a789d09a6046691b3b6003740c48334a0b79aa4)
---
.../db/pipe/agent/runtime/PipeAgentLauncher.java | 172 +++++++++++++++++----
.../pipe/agent/runtime/PipeAgentLauncherTest.java | 168 ++++++++++++++++++++
2 files changed, 314 insertions(+), 26 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeAgentLauncher.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeAgentLauncher.java
index 01289109371..c7a669fefbf 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeAgentLauncher.java
+++
b/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;
@@ -38,14 +37,16 @@ import
org.apache.iotdb.db.service.ResourcesInformationHolder;
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 {
@@ -69,6 +70,7 @@ class PipeAgentLauncher {
final List<PipePluginMeta> uninstalledOrConflictedPipePluginMetaList =
getUninstalledOrConflictedPipePluginMetaList(resourcesInformationHolder);
+ final Set<String> unavailablePipePluginNameSet = new HashSet<>();
int index = 0;
while (index < uninstalledOrConflictedPipePluginMetaList.size()) {
List<PipePluginMeta> curList = new ArrayList<>();
@@ -79,20 +81,26 @@ class PipeAgentLauncher {
offset++;
}
index += offset;
- fetchAndSavePipePluginJars(curList);
+ unavailablePipePluginNameSet.addAll(fetchAndSavePipePluginJars(curList));
}
// create instances of pipe plugins and do registration
- try {
- for (PipePluginMeta meta :
resourcesInformationHolder.getPipePluginMetaList()) {
- if (meta.isBuiltin()) {
- continue;
- }
+ for (PipePluginMeta meta :
resourcesInformationHolder.getPipePluginMetaList()) {
+ if (meta.isBuiltin()) {
+ continue;
+ }
+ if (unavailablePipePluginNameSet.contains(meta.getPluginName())) {
+ continue;
+ }
+ try {
PipeDataNodeAgent.plugin().doRegister(meta);
+ } catch (Throwable e) {
+ // Ignore a single broken plugin and continue startup.
+ LOGGER.error(
+ "Failure when register pipe plugin {}. Skip this plugin and
continue startup.",
+ meta.getPluginName(),
+ e);
}
- } catch (Exception e) {
- // Ignore the pipe plugin errors and continue to start
- LOGGER.warn("Failure when register pipe plugins, will ignore.", e);
}
}
@@ -132,28 +140,140 @@ class PipeAgentLauncher {
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("Failed to get pipe plugin jar from config
node.");
+ resp = configNodeClient.getPipePluginJar(new
TGetJarInListReq(jarNameList));
+ } catch (Exception e) {
+ LOGGER.error(
+ "Failed to fetch pipe plugin jars from ConfigNode. Plugins: {},
jars: {}, status: {}. "
+ + "Retrying each plugin individually.",
+ pluginNameList,
+ jarNameList,
+ null,
+ e);
+ return fetchAndSavePipePluginJarsIndividually(pipePluginMetaList);
+ }
+
+ if (resp == null
+ || resp.getStatus() == null
+ || resp.getStatus().getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ LOGGER.error(
+ "Failed to fetch pipe plugin jars from ConfigNode. Plugins: {},
jars: {}, status: {}. "
+ + "Retrying each plugin individually.",
+ 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(
+ "ConfigNode returned {} pipe plugin jars for {} requested plugins. "
+ + "Plugins: {}, jars: {}. Retrying each plugin individually.",
+ 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(Collections.singletonList(jarName)));
+ } catch (Exception e) {
+ LOGGER.error(
+ "Failed to fetch pipe plugin jar {} for pipe plugin {} from
ConfigNode.",
+ 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(
+ "Failed to fetch pipe plugin jar from ConfigNode for plugin
%s (jar %s). "
+ + "Status: %s.",
+ pluginName, jarName, resp == null ? null :
resp.getStatus()));
+ 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(
+ "ConfigNode returned %d jars for pipe plugin %s while one
was requested.",
+ jarList == null ? 0 : jarList.size(), pluginName));
+ LOGGER.error(exception.getMessage(), exception);
+ return false;
+ }
+
+ try {
+ PipePluginExecutableManager.getInstance()
+ .savePluginToInstallDir(jarList.get(0), pluginName, jarName);
+ return true;
+ } catch (Exception e) {
+ LOGGER.error("Failed to save jar {} for pipe plugin {}.", 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) {
+ LOGGER.error(
+ "Failed to save jar {} for pipe plugin {}.",
+ 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() {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/runtime/PipeAgentLauncherTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/runtime/PipeAgentLauncherTest.java
new file mode 100644
index 00000000000..49526ad51be
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/runtime/PipeAgentLauncherTest.java
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.agent.runtime;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.consensus.ConfigRegionId;
+import org.apache.iotdb.commons.pipe.agent.plugin.meta.PipePluginMeta;
+import
org.apache.iotdb.commons.pipe.agent.plugin.service.PipePluginExecutableManager;
+import org.apache.iotdb.confignode.rpc.thrift.TGetJarInListReq;
+import org.apache.iotdb.confignode.rpc.thrift.TGetJarInListResp;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
+import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+@PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*",
"javax.management.*"})
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({ConfigNodeClientManager.class,
PipePluginExecutableManager.class})
+public class PipeAgentLauncherTest {
+
+ private IClientManager<ConfigRegionId, ConfigNodeClient>
configNodeClientManager;
+ private ConfigNodeClient configNodeClient;
+ private PipePluginExecutableManager pipePluginExecutableManager;
+
+ @Before
+ public void setUp() throws Exception {
+ configNodeClientManager = Mockito.mock(IClientManager.class);
+ configNodeClient = Mockito.mock(ConfigNodeClient.class);
+ pipePluginExecutableManager =
Mockito.mock(PipePluginExecutableManager.class);
+
+ PowerMockito.mockStatic(ConfigNodeClientManager.class);
+ PowerMockito.mockStatic(PipePluginExecutableManager.class);
+
PowerMockito.when(ConfigNodeClientManager.getInstance()).thenReturn(configNodeClientManager);
+ PowerMockito.when(PipePluginExecutableManager.getInstance())
+ .thenReturn(pipePluginExecutableManager);
+
Mockito.when(configNodeClientManager.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID))
+ .thenReturn(configNodeClient);
+ }
+
+ @Test
+ public void
testBatchFailureRetriesIndividuallyAndOnlyMarksFailedPluginUnavailable()
+ throws Exception {
+ final PipePluginMeta healthyPlugin = pipePluginMeta("healthy",
"healthy.jar");
+ final PipePluginMeta missingPlugin = pipePluginMeta("missing",
"missing.jar");
+ final TGetJarInListResp failedResponse = failureResponse();
+
+
Mockito.when(configNodeClient.getPipePluginJar(Mockito.any(TGetJarInListReq.class)))
+ .thenAnswer(
+ invocation -> {
+ final List<String> jarNames =
+ ((TGetJarInListReq)
invocation.getArgument(0)).getJarNameList();
+ if (jarNames.size() == 2 ||
"missing.jar".equals(jarNames.get(0))) {
+ return failedResponse;
+ }
+ return
successResponse(Collections.singletonList(ByteBuffer.wrap(new byte[] {1})));
+ });
+
+ final Set<String> unavailablePlugins =
+
PipeAgentLauncher.fetchAndSavePipePluginJars(Arrays.asList(healthyPlugin,
missingPlugin));
+
+ Assert.assertEquals(Collections.singleton("MISSING"), unavailablePlugins);
+ Mockito.verify(pipePluginExecutableManager)
+ .savePluginToInstallDir(
+ Mockito.any(ByteBuffer.class), Mockito.eq("HEALTHY"),
Mockito.eq("healthy.jar"));
+ }
+
+ @Test
+ public void testBatchJarCountMismatchRetriesEachPluginIndividually() throws
Exception {
+ final PipePluginMeta firstPlugin = pipePluginMeta("first", "first.jar");
+ final PipePluginMeta secondPlugin = pipePluginMeta("second", "second.jar");
+
+
Mockito.when(configNodeClient.getPipePluginJar(Mockito.any(TGetJarInListReq.class)))
+
.thenReturn(successResponse(Collections.singletonList(ByteBuffer.wrap(new
byte[] {1}))));
+
+ final Set<String> unavailablePlugins =
+
PipeAgentLauncher.fetchAndSavePipePluginJars(Arrays.asList(firstPlugin,
secondPlugin));
+
+ Assert.assertTrue(unavailablePlugins.isEmpty());
+ final ArgumentCaptor<TGetJarInListReq> requestCaptor =
+ ArgumentCaptor.forClass(TGetJarInListReq.class);
+ Mockito.verify(configNodeClient,
Mockito.times(3)).getPipePluginJar(requestCaptor.capture());
+ Assert.assertEquals(
+ Arrays.asList("first.jar", "second.jar"),
+ requestCaptor.getAllValues().get(0).getJarNameList());
+ Assert.assertEquals(
+ Collections.singletonList("first.jar"),
+ requestCaptor.getAllValues().get(1).getJarNameList());
+ Assert.assertEquals(
+ Collections.singletonList("second.jar"),
+ requestCaptor.getAllValues().get(2).getJarNameList());
+ Mockito.verify(pipePluginExecutableManager, Mockito.times(2))
+ .savePluginToInstallDir(
+ Mockito.any(ByteBuffer.class), Mockito.anyString(),
Mockito.anyString());
+ }
+
+ @Test
+ public void testSaveFailureDoesNotBlockSubsequentPlugin() throws Exception {
+ final PipePluginMeta failedPlugin = pipePluginMeta("failed", "failed.jar");
+ final PipePluginMeta healthyPlugin = pipePluginMeta("healthy",
"healthy.jar");
+
Mockito.when(configNodeClient.getPipePluginJar(Mockito.any(TGetJarInListReq.class)))
+ .thenReturn(
+ successResponse(
+ Arrays.asList(ByteBuffer.wrap(new byte[] {1}),
ByteBuffer.wrap(new byte[] {2}))));
+ Mockito.doThrow(new IOException("injected save failure"))
+ .when(pipePluginExecutableManager)
+ .savePluginToInstallDir(
+ Mockito.any(ByteBuffer.class), Mockito.eq("FAILED"),
Mockito.eq("failed.jar"));
+
+ final Set<String> unavailablePlugins =
+
PipeAgentLauncher.fetchAndSavePipePluginJars(Arrays.asList(failedPlugin,
healthyPlugin));
+
+ Assert.assertEquals(Collections.singleton("FAILED"), unavailablePlugins);
+ Mockito.verify(pipePluginExecutableManager)
+ .savePluginToInstallDir(
+ Mockito.any(ByteBuffer.class), Mockito.eq("HEALTHY"),
Mockito.eq("healthy.jar"));
+ }
+
+ private static PipePluginMeta pipePluginMeta(final String pluginName, final
String jarName) {
+ return new PipePluginMeta(pluginName, "test.class", false, jarName,
"test-md5");
+ }
+
+ private static TGetJarInListResp successResponse(final List<ByteBuffer>
jarList) {
+ return new TGetJarInListResp(
+ new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()), jarList);
+ }
+
+ private static TGetJarInListResp failureResponse() {
+ return new TGetJarInListResp(
+ new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()),
+ Collections.emptyList());
+ }
+}
