You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2024/10/29 15:08:08 UTC
(iotdb) branch rc/1.3.3 updated: Pipe: validate all pipe parameters before alter (#13932) (#13958)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch rc/1.3.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rc/1.3.3 by this push:
new a19fda5010f Pipe: validate all pipe parameters before alter (#13932) (#13958)
a19fda5010f is described below
commit a19fda5010fcac0d8f70e00f9c4f55264c6043fb
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Tue Oct 29 23:08:01 2024 +0800
Pipe: validate all pipe parameters before alter (#13932) (#13958)
(cherry picked from commit d35da563bd026e94e69d2765dc8c316ed60cdbf4)
---
.../impl/pipe/task/AlterPipeProcedureV2.java | 2 +-
.../config/executor/ClusterConfigTaskExecutor.java | 110 ++++++++++++++++++---
2 files changed, 98 insertions(+), 14 deletions(-)
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java
index d4ab2a2cf5b..ccdd0e76b31 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java
@@ -63,7 +63,7 @@ public class AlterPipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
private PipeRuntimeMeta currentPipeRuntimeMeta;
private PipeRuntimeMeta updatedPipeRuntimeMeta;
- private ProcedureType procedureType;
+ private final ProcedureType procedureType;
public AlterPipeProcedureV2(ProcedureType procedureType) {
super();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
index ed10039d425..d59d9deb55e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
@@ -45,6 +45,7 @@ import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathPatternTree;
import org.apache.iotdb.commons.pipe.agent.plugin.service.PipePluginClassLoader;
import org.apache.iotdb.commons.pipe.agent.plugin.service.PipePluginExecutableManager;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
import org.apache.iotdb.commons.pipe.connector.payload.airgap.AirGapPseudoTPipeTransferRequest;
import org.apache.iotdb.commons.schema.view.LogicalViewSchema;
@@ -80,6 +81,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TDropPipePluginReq;
import org.apache.iotdb.confignode.rpc.thrift.TDropPipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TDropTopicReq;
import org.apache.iotdb.confignode.rpc.thrift.TDropTriggerReq;
+import org.apache.iotdb.confignode.rpc.thrift.TGetAllPipeInfoResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetDatabaseReq;
import org.apache.iotdb.confignode.rpc.thrift.TGetPipePluginTableResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetRegionIdReq;
@@ -231,6 +233,7 @@ import org.apache.iotdb.db.storageengine.dataregion.compaction.repair.RepairTask
import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.CompactionScheduleTaskManager;
import org.apache.iotdb.db.trigger.service.TriggerClassLoader;
import org.apache.iotdb.pipe.api.PipePlugin;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -1806,21 +1809,102 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
return future;
}
- // Validate pipe plugin before alteration - only validate replace mode
+ // Validate pipe existence
+ final PipeMeta pipeMetaFromCoordinator;
+ try (final ConfigNodeClient configNodeClient =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
+ final TGetAllPipeInfoResp getAllPipeInfoResp = configNodeClient.getAllPipeInfo();
+ if (getAllPipeInfoResp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ String exceptionMessage =
+ String.format(
+ "Failed to get pipe info from config node, status is %s.",
+ getAllPipeInfoResp.getStatus());
+ LOGGER.warn(exceptionMessage);
+ future.setException(
+ new IoTDBException(exceptionMessage, TSStatusCode.PIPE_ERROR.getStatusCode()));
+ return future;
+ }
+
+ pipeMetaFromCoordinator =
+ getAllPipeInfoResp.getAllPipeInfo().stream()
+ .map(PipeMeta::deserialize)
+ .filter(
+ pipeMeta ->
+ pipeMeta
+ .getStaticMeta()
+ .getPipeName()
+ .equals(alterPipeStatement.getPipeName()))
+ .findFirst()
+ .orElse(null);
+ if (pipeMetaFromCoordinator == null) {
+ final String exceptionMessage =
+ String.format(
+ "Failed to alter pipe %s, pipe not found in system.",
+ alterPipeStatement.getPipeName());
+ LOGGER.warn(exceptionMessage);
+ future.setException(
+ new IoTDBException(exceptionMessage, TSStatusCode.PIPE_ERROR.getStatusCode()));
+ return future;
+ }
+ } catch (Exception e) {
+ final String exceptionMessage =
+ String.format(
+ "Failed to alter pipe %s, because %s",
+ alterPipeStatement.getPipeName(), e.getMessage());
+ LOGGER.warn(exceptionMessage, e);
+ future.setException(
+ new IoTDBException(exceptionMessage, TSStatusCode.PIPE_ERROR.getStatusCode()));
+ return future;
+ }
+
+ // Construct temporary pipe static meta for validation
final String pipeName = alterPipeStatement.getPipeName();
try {
- if (!alterPipeStatement.getExtractorAttributes().isEmpty()
- && alterPipeStatement.isReplaceAllExtractorAttributes()) {
- PipeDataNodeAgent.plugin().validateExtractor(alterPipeStatement.getExtractorAttributes());
- }
- if (!alterPipeStatement.getProcessorAttributes().isEmpty()
- && alterPipeStatement.isReplaceAllProcessorAttributes()) {
- PipeDataNodeAgent.plugin().validateProcessor(alterPipeStatement.getProcessorAttributes());
- }
- if (!alterPipeStatement.getConnectorAttributes().isEmpty()
- && alterPipeStatement.isReplaceAllConnectorAttributes()) {
- PipeDataNodeAgent.plugin()
- .validateConnector(pipeName, alterPipeStatement.getConnectorAttributes());
+ if (!alterPipeStatement.getExtractorAttributes().isEmpty()) {
+ if (alterPipeStatement.isReplaceAllExtractorAttributes()) {
+ PipeDataNodeAgent.plugin().validateExtractor(alterPipeStatement.getExtractorAttributes());
+ } else {
+ pipeMetaFromCoordinator
+ .getStaticMeta()
+ .getExtractorParameters()
+ .addOrReplaceEquivalentAttributes(
+ new PipeParameters(alterPipeStatement.getExtractorAttributes()));
+ PipeDataNodeAgent.plugin()
+ .validateExtractor(
+ pipeMetaFromCoordinator.getStaticMeta().getExtractorParameters().getAttribute());
+ }
+ }
+
+ if (!alterPipeStatement.getProcessorAttributes().isEmpty()) {
+ if (alterPipeStatement.isReplaceAllProcessorAttributes()) {
+ PipeDataNodeAgent.plugin().validateProcessor(alterPipeStatement.getProcessorAttributes());
+ } else {
+ pipeMetaFromCoordinator
+ .getStaticMeta()
+ .getProcessorParameters()
+ .addOrReplaceEquivalentAttributes(
+ new PipeParameters(alterPipeStatement.getProcessorAttributes()));
+ PipeDataNodeAgent.plugin()
+ .validateProcessor(
+ pipeMetaFromCoordinator.getStaticMeta().getProcessorParameters().getAttribute());
+ }
+ }
+
+ if (!alterPipeStatement.getConnectorAttributes().isEmpty()) {
+ if (alterPipeStatement.isReplaceAllConnectorAttributes()) {
+ PipeDataNodeAgent.plugin()
+ .validateConnector(pipeName, alterPipeStatement.getConnectorAttributes());
+ } else {
+ pipeMetaFromCoordinator
+ .getStaticMeta()
+ .getConnectorParameters()
+ .addOrReplaceEquivalentAttributes(
+ new PipeParameters(alterPipeStatement.getConnectorAttributes()));
+ PipeDataNodeAgent.plugin()
+ .validateConnector(
+ pipeName,
+ pipeMetaFromCoordinator.getStaticMeta().getConnectorParameters().getAttribute());
+ }
}
} catch (Exception e) {
LOGGER.info("Failed to validate alter pipe statement, because {}", e.getMessage(), e);