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 2023/11/15 14:30:45 UTC
(iotdb) branch master updated: Pipe: Make the pipe to detect illegal parameters when there is no data (#11541)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 36bb554e756 Pipe: Make the pipe to detect illegal parameters when there is no data (#11541)
36bb554e756 is described below
commit 36bb554e756408903ce46d46586a938b96d42c9e
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Wed Nov 15 22:30:38 2023 +0800
Pipe: Make the pipe to detect illegal parameters when there is no data (#11541)
Co-authored-by: Steve Yurong Su <ro...@apache.org>
---
.../apache/iotdb/pipe/it/IoTDBPipeSyntaxIT.java | 43 ++++----
.../agent/plugin/PipeConnectorConstructor.java | 112 +++++++++++++++++++++
.../agent/plugin/PipeExtractorConstructor.java | 58 +++++++++++
.../db/pipe/agent/plugin/PipePluginAgent.java | 111 ++++++++++----------
.../pipe/agent/plugin/PipePluginConstructor.java | 85 ++++++++++++++++
.../agent/plugin/PipeProcessorConstructor.java | 52 ++++++++++
.../db/pipe/task/stage/PipeTaskExtractorStage.java | 20 +---
.../db/pipe/task/stage/PipeTaskProcessorStage.java | 15 +--
.../connector/PipeConnectorSubtaskManager.java | 73 +-------------
.../config/executor/ClusterConfigTaskExecutor.java | 11 ++
.../db/pipe/agent/plugin/PipePluginAgentTest.java | 6 +-
11 files changed, 401 insertions(+), 185 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/IoTDBPipeSyntaxIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/IoTDBPipeSyntaxIT.java
index d3c679d13cd..d29b8d9ea50 100644
--- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/IoTDBPipeSyntaxIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/IoTDBPipeSyntaxIT.java
@@ -126,37 +126,28 @@ public class IoTDBPipeSyntaxIT {
}
@Test
- public void testRevertParameterOrder() throws Exception {
+ public void testRevertParameterOrder() {
DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
String receiverIp = receiverDataNode.getIp();
int receiverPort = receiverDataNode.getPort();
- try (SyncConfigNodeIServiceClient client =
- (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
-
- try (Connection connection = senderEnv.getConnection();
- Statement statement = connection.createStatement()) {
- statement.execute(
- String.format(
- "create pipe p1"
- + " with extractor ("
- + "'extractor.realtime.mode'='hybrid',"
- + "'extractor.history.enable'='false') "
- + " with connector ("
- + "'connector.batch.enable'='false', "
- + "'connector.port'='%s',"
- + "'connector.ip'='%s',"
- + "'connector'='iotdb-thrift-connector')",
- receiverIp, receiverPort));
- } catch (SQLException e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
-
- List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
- Assert.assertTrue(
- showPipeResult.stream().anyMatch((o) -> o.id.equals("p1") && o.state.equals("STOPPED")));
+ try (Connection connection = senderEnv.getConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe p1"
+ + " with extractor ("
+ + "'extractor.realtime.mode'='hybrid',"
+ + "'extractor.history.enable'='false') "
+ + " with connector ("
+ + "'connector.batch.enable'='false', "
+ + "'connector.port'='%s',"
+ + "'connector.ip'='%s',"
+ + "'connector'='iotdb-thrift-connector')",
+ receiverIp, receiverPort));
+ fail();
+ } catch (SQLException ignore) {
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipeConnectorConstructor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipeConnectorConstructor.java
new file mode 100644
index 00000000000..0137f1c3c7e
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipeConnectorConstructor.java
@@ -0,0 +1,112 @@
+/*
+ * 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.plugin;
+
+import org.apache.iotdb.commons.pipe.plugin.builtin.BuiltinPipePlugin;
+import org.apache.iotdb.commons.pipe.plugin.builtin.connector.DoNothingConnector;
+import org.apache.iotdb.commons.pipe.plugin.meta.DataNodePipePluginMetaKeeper;
+import org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant;
+import org.apache.iotdb.db.pipe.connector.protocol.airgap.IoTDBAirGapConnector;
+import org.apache.iotdb.db.pipe.connector.protocol.legacy.IoTDBLegacyPipeConnector;
+import org.apache.iotdb.db.pipe.connector.protocol.opcua.OpcUaConnector;
+import org.apache.iotdb.db.pipe.connector.protocol.thrift.async.IoTDBThriftAsyncConnector;
+import org.apache.iotdb.db.pipe.connector.protocol.thrift.sync.IoTDBThriftSyncConnector;
+import org.apache.iotdb.db.pipe.connector.protocol.websocket.WebSocketConnector;
+import org.apache.iotdb.db.pipe.connector.protocol.writeback.WriteBackConnector;
+import org.apache.iotdb.pipe.api.PipeConnector;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+import org.apache.iotdb.pipe.api.exception.PipeException;
+
+import java.util.Arrays;
+
+public class PipeConnectorConstructor extends PipePluginConstructor {
+
+ PipeConnectorConstructor(DataNodePipePluginMetaKeeper pipePluginMetaKeeper) {
+ super(pipePluginMetaKeeper);
+ }
+
+ @Override
+ protected void initConstructors() {
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR.getPipePluginName(),
+ IoTDBThriftAsyncConnector::new);
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.IOTDB_THRIFT_SYNC_CONNECTOR.getPipePluginName(),
+ IoTDBThriftSyncConnector::new);
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName(),
+ IoTDBThriftAsyncConnector::new);
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.IOTDB_LEGACY_PIPE_CONNECTOR.getPipePluginName(),
+ IoTDBLegacyPipeConnector::new);
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName(), IoTDBAirGapConnector::new);
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.WEBSOCKET_CONNECTOR.getPipePluginName(), WebSocketConnector::new);
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.OPC_UA_CONNECTOR.getPipePluginName(), OpcUaConnector::new);
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.DO_NOTHING_CONNECTOR.getPipePluginName(), DoNothingConnector::new);
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.WRITE_BACK_CONNECTOR.getPipePluginName(), WriteBackConnector::new);
+
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.IOTDB_THRIFT_SINK.getPipePluginName(), IoTDBThriftAsyncConnector::new);
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.IOTDB_THRIFT_SYNC_SINK.getPipePluginName(),
+ IoTDBThriftSyncConnector::new);
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_SINK.getPipePluginName(),
+ IoTDBThriftAsyncConnector::new);
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.IOTDB_LEGACY_PIPE_SINK.getPipePluginName(),
+ IoTDBLegacyPipeConnector::new);
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.IOTDB_AIR_GAP_SINK.getPipePluginName(), IoTDBAirGapConnector::new);
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.WEBSOCKET_SINK.getPipePluginName(), WebSocketConnector::new);
+ PLUGIN_CONSTRUCTORS.put(BuiltinPipePlugin.OPC_UA_SINK.getPipePluginName(), OpcUaConnector::new);
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.DO_NOTHING_SINK.getPipePluginName(), DoNothingConnector::new);
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.WRITE_BACK_SINK.getPipePluginName(), WriteBackConnector::new);
+ }
+
+ @Override
+ PipeConnector reflectPlugin(PipeParameters connectorParameters) {
+ if (!connectorParameters.hasAnyAttributes(
+ PipeConnectorConstant.CONNECTOR_KEY, PipeConnectorConstant.SINK_KEY)) {
+ throw new PipeException(
+ "Failed to reflect PipeConnector instance because "
+ + "'connector' is not specified in the parameters.");
+ }
+
+ return (PipeConnector)
+ reflectPluginByKey(
+ connectorParameters
+ .getStringOrDefault(
+ Arrays.asList(
+ PipeConnectorConstant.CONNECTOR_KEY, PipeConnectorConstant.SINK_KEY),
+ BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR.getPipePluginName())
+ // Convert the value of `CONNECTOR_KEY` or `SINK_KEY` to lowercase for matching in
+ // `PLUGIN_CONSTRUCTORS`
+ .toLowerCase());
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipeExtractorConstructor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipeExtractorConstructor.java
new file mode 100644
index 00000000000..b5adff57b27
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipeExtractorConstructor.java
@@ -0,0 +1,58 @@
+/*
+ * 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.plugin;
+
+import org.apache.iotdb.commons.pipe.plugin.builtin.BuiltinPipePlugin;
+import org.apache.iotdb.commons.pipe.plugin.meta.DataNodePipePluginMetaKeeper;
+import org.apache.iotdb.db.pipe.config.constant.PipeExtractorConstant;
+import org.apache.iotdb.db.pipe.extractor.IoTDBDataRegionExtractor;
+import org.apache.iotdb.pipe.api.PipeExtractor;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+
+import java.util.Arrays;
+
+public class PipeExtractorConstructor extends PipePluginConstructor {
+
+ PipeExtractorConstructor(DataNodePipePluginMetaKeeper pipePluginMetaKeeper) {
+ super(pipePluginMetaKeeper);
+ }
+
+ @Override
+ protected void initConstructors() {
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.IOTDB_EXTRACTOR.getPipePluginName(), IoTDBDataRegionExtractor::new);
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.IOTDB_SOURCE.getPipePluginName(), IoTDBDataRegionExtractor::new);
+ }
+
+ @Override
+ PipeExtractor reflectPlugin(PipeParameters extractorParameters) {
+ return (PipeExtractor)
+ reflectPluginByKey(
+ extractorParameters
+ .getStringOrDefault(
+ Arrays.asList(
+ PipeExtractorConstant.EXTRACTOR_KEY, PipeExtractorConstant.SOURCE_KEY),
+ BuiltinPipePlugin.IOTDB_EXTRACTOR.getPipePluginName())
+ // Convert the value of `EXTRACTOR_KEY` or `SOURCE_KEY` to lowercase for matching
+ // `IOTDB_EXTRACTOR`
+ .toLowerCase());
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipePluginAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipePluginAgent.java
index f3445c7a107..9ea0684e15f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipePluginAgent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipePluginAgent.java
@@ -19,19 +19,17 @@
package org.apache.iotdb.db.pipe.agent.plugin;
-import org.apache.iotdb.commons.pipe.plugin.builtin.BuiltinPipePlugin;
import org.apache.iotdb.commons.pipe.plugin.meta.DataNodePipePluginMetaKeeper;
import org.apache.iotdb.commons.pipe.plugin.meta.PipePluginMeta;
import org.apache.iotdb.commons.pipe.plugin.service.PipePluginClassLoader;
import org.apache.iotdb.commons.pipe.plugin.service.PipePluginClassLoaderManager;
import org.apache.iotdb.commons.pipe.plugin.service.PipePluginExecutableManager;
-import org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant;
-import org.apache.iotdb.db.pipe.config.constant.PipeExtractorConstant;
-import org.apache.iotdb.db.pipe.config.constant.PipeProcessorConstant;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.CreatePipeStatement;
import org.apache.iotdb.pipe.api.PipeConnector;
import org.apache.iotdb.pipe.api.PipeExtractor;
import org.apache.iotdb.pipe.api.PipePlugin;
import org.apache.iotdb.pipe.api.PipeProcessor;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.exception.PipeException;
@@ -41,7 +39,6 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.nio.ByteBuffer;
-import java.util.Arrays;
import java.util.concurrent.locks.ReentrantLock;
public class PipePluginAgent {
@@ -52,8 +49,15 @@ public class PipePluginAgent {
private final DataNodePipePluginMetaKeeper pipePluginMetaKeeper;
+ private final PipeExtractorConstructor pipeExtractorConstructor;
+ private final PipeProcessorConstructor pipeProcessorConstructor;
+ private final PipeConnectorConstructor pipeConnectorConstructor;
+
public PipePluginAgent() {
this.pipePluginMetaKeeper = new DataNodePipePluginMetaKeeper();
+ this.pipeExtractorConstructor = new PipeExtractorConstructor(pipePluginMetaKeeper);
+ this.pipeProcessorConstructor = new PipeProcessorConstructor(pipePluginMetaKeeper);
+ this.pipeConnectorConstructor = new PipeConnectorConstructor(pipePluginMetaKeeper);
}
/////////////////////////////// Lock ///////////////////////////////
@@ -200,61 +204,60 @@ public class PipePluginAgent {
}
}
+ /**
+ * Validation should have the granularity of "DataNode level", not "DataRegion level", because a
+ * DataNode may have no DataRegion at all when creating pipe
+ */
+ public void validate(CreatePipeStatement createPipeStatement) throws Exception {
+ final PipeParameters extractorParameters =
+ new PipeParameters(createPipeStatement.getExtractorAttributes());
+ final PipeExtractor temporaryExtractor = reflectExtractor(extractorParameters);
+ try {
+ temporaryExtractor.validate(new PipeParameterValidator(extractorParameters));
+ } finally {
+ try {
+ temporaryExtractor.close();
+ } catch (Exception e) {
+ LOGGER.warn("Failed to close temporary extractor: {}", e.getMessage());
+ }
+ }
+
+ final PipeParameters processorParameters =
+ new PipeParameters(createPipeStatement.getProcessorAttributes());
+ final PipeProcessor temporaryProcessor = reflectProcessor(processorParameters);
+ try {
+ temporaryProcessor.validate(new PipeParameterValidator(processorParameters));
+ } finally {
+ try {
+ temporaryProcessor.close();
+ } catch (Exception e) {
+ LOGGER.warn("Failed to close temporary processor: {}", e.getMessage());
+ }
+ }
+
+ final PipeParameters connectorParameters =
+ new PipeParameters(createPipeStatement.getConnectorAttributes());
+ final PipeConnector temporaryConnector = reflectConnector(connectorParameters);
+ try {
+ temporaryConnector.validate(new PipeParameterValidator(connectorParameters));
+ } finally {
+ try {
+ temporaryConnector.close();
+ } catch (Exception e) {
+ LOGGER.warn("Failed to close temporary connector: {}", e.getMessage());
+ }
+ }
+ }
+
public PipeExtractor reflectExtractor(PipeParameters extractorParameters) {
- return (PipeExtractor)
- reflect(
- extractorParameters.getStringOrDefault(
- Arrays.asList(
- PipeExtractorConstant.EXTRACTOR_KEY, PipeExtractorConstant.SOURCE_KEY),
- BuiltinPipePlugin.IOTDB_EXTRACTOR.getPipePluginName()));
+ return pipeExtractorConstructor.reflectPlugin(extractorParameters);
}
public PipeProcessor reflectProcessor(PipeParameters processorParameters) {
- return (PipeProcessor)
- reflect(
- processorParameters.getStringOrDefault(
- PipeProcessorConstant.PROCESSOR_KEY,
- BuiltinPipePlugin.DO_NOTHING_PROCESSOR.getPipePluginName()));
+ return pipeProcessorConstructor.reflectPlugin(processorParameters);
}
public PipeConnector reflectConnector(PipeParameters connectorParameters) {
- if (!connectorParameters.hasAnyAttributes(
- PipeConnectorConstant.CONNECTOR_KEY, PipeConnectorConstant.SINK_KEY)) {
- throw new PipeException(
- "Failed to reflect PipeConnector instance because "
- + "'connector' is not specified in the parameters.");
- }
- return (PipeConnector)
- reflect(
- connectorParameters.getStringByKeys(
- PipeConnectorConstant.CONNECTOR_KEY, PipeConnectorConstant.SINK_KEY));
- }
-
- private PipePlugin reflect(String pluginName) {
- PipePluginMeta information = pipePluginMetaKeeper.getPipePluginMeta(pluginName);
- if (information == null) {
- String errorMessage =
- String.format(
- "Failed to reflect PipePlugin instance, because "
- + "PipePlugin %s has not been registered.",
- pluginName.toUpperCase());
- LOGGER.warn(errorMessage);
- throw new PipeException(errorMessage);
- }
-
- try {
- return (PipePlugin)
- pipePluginMetaKeeper.getPluginClass(pluginName).getDeclaredConstructor().newInstance();
- } catch (InstantiationException
- | InvocationTargetException
- | NoSuchMethodException
- | IllegalAccessException e) {
- String errorMessage =
- String.format(
- "Failed to reflect PipePlugin %s(%s) instance, because %s",
- pluginName, information.getClassName(), e);
- LOGGER.warn(errorMessage, e);
- throw new PipeException(errorMessage);
- }
+ return pipeConnectorConstructor.reflectPlugin(connectorParameters);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipePluginConstructor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipePluginConstructor.java
new file mode 100644
index 00000000000..1296933613c
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipePluginConstructor.java
@@ -0,0 +1,85 @@
+/*
+ * 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.plugin;
+
+import org.apache.iotdb.commons.pipe.plugin.meta.DataNodePipePluginMetaKeeper;
+import org.apache.iotdb.commons.pipe.plugin.meta.PipePluginMeta;
+import org.apache.iotdb.pipe.api.PipePlugin;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+import org.apache.iotdb.pipe.api.exception.PipeException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Supplier;
+
+public abstract class PipePluginConstructor {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PipePluginConstructor.class);
+
+ private final DataNodePipePluginMetaKeeper pipePluginMetaKeeper;
+
+ protected static final Map<String, Supplier<PipePlugin>> PLUGIN_CONSTRUCTORS = new HashMap<>();
+
+ PipePluginConstructor(DataNodePipePluginMetaKeeper pipePluginMetaKeeper) {
+ this.pipePluginMetaKeeper = pipePluginMetaKeeper;
+ initConstructors();
+ }
+
+ // New plugins shall be put here
+ protected abstract void initConstructors();
+
+ abstract PipePlugin reflectPlugin(PipeParameters pipeParameters);
+
+ protected final PipePlugin reflectPluginByKey(String pluginKey) {
+ return PLUGIN_CONSTRUCTORS.getOrDefault(pluginKey, () -> reflect(pluginKey)).get();
+ }
+
+ private PipePlugin reflect(String pluginName) {
+ PipePluginMeta information = pipePluginMetaKeeper.getPipePluginMeta(pluginName);
+ if (information == null) {
+ String errorMessage =
+ String.format(
+ "Failed to reflect PipePlugin instance, because "
+ + "PipePlugin %s has not been registered.",
+ pluginName.toUpperCase());
+ LOGGER.warn(errorMessage);
+ throw new PipeException(errorMessage);
+ }
+
+ try {
+ return (PipePlugin)
+ pipePluginMetaKeeper.getPluginClass(pluginName).getDeclaredConstructor().newInstance();
+ } catch (InstantiationException
+ | InvocationTargetException
+ | NoSuchMethodException
+ | IllegalAccessException e) {
+ String errorMessage =
+ String.format(
+ "Failed to reflect PipePlugin %s(%s) instance, because %s",
+ pluginName, information.getClassName(), e);
+ LOGGER.warn(errorMessage, e);
+ throw new PipeException(errorMessage);
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipeProcessorConstructor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipeProcessorConstructor.java
new file mode 100644
index 00000000000..abd3e8669f9
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipeProcessorConstructor.java
@@ -0,0 +1,52 @@
+/*
+ * 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.plugin;
+
+import org.apache.iotdb.commons.pipe.plugin.builtin.BuiltinPipePlugin;
+import org.apache.iotdb.commons.pipe.plugin.builtin.processor.DoNothingProcessor;
+import org.apache.iotdb.commons.pipe.plugin.meta.DataNodePipePluginMetaKeeper;
+import org.apache.iotdb.db.pipe.config.constant.PipeProcessorConstant;
+import org.apache.iotdb.pipe.api.PipeProcessor;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+
+public class PipeProcessorConstructor extends PipePluginConstructor {
+ PipeProcessorConstructor(DataNodePipePluginMetaKeeper pipePluginMetaKeeper) {
+ super(pipePluginMetaKeeper);
+ }
+
+ @Override
+ protected void initConstructors() {
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.DO_NOTHING_PROCESSOR.getPipePluginName(), DoNothingProcessor::new);
+ }
+
+ @Override
+ PipeProcessor reflectPlugin(PipeParameters processorParameters) {
+ return (PipeProcessor)
+ reflectPluginByKey(
+ processorParameters
+ .getStringOrDefault(
+ PipeProcessorConstant.PROCESSOR_KEY,
+ BuiltinPipePlugin.DO_NOTHING_PROCESSOR.getPipePluginName())
+ // Convert the value of `PROCESSOR_KEY` to lowercase for matching
+ // `DO_NOTHING_PROCESSOR`
+ .toLowerCase());
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskExtractorStage.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskExtractorStage.java
index 1b4ed57251b..8155d9b53cb 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskExtractorStage.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskExtractorStage.java
@@ -20,21 +20,16 @@
package org.apache.iotdb.db.pipe.task.stage;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
-import org.apache.iotdb.commons.pipe.plugin.builtin.BuiltinPipePlugin;
import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
import org.apache.iotdb.db.pipe.agent.PipeAgent;
-import org.apache.iotdb.db.pipe.config.constant.PipeExtractorConstant;
import org.apache.iotdb.db.pipe.config.plugin.configuraion.PipeTaskRuntimeConfiguration;
import org.apache.iotdb.db.pipe.config.plugin.env.PipeTaskExtractorRuntimeEnvironment;
-import org.apache.iotdb.db.pipe.extractor.IoTDBDataRegionExtractor;
import org.apache.iotdb.db.pipe.task.connection.EventSupplier;
import org.apache.iotdb.pipe.api.PipeExtractor;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.exception.PipeException;
-import java.util.Arrays;
-
public class PipeTaskExtractorStage extends PipeTaskStage {
private final PipeExtractor pipeExtractor;
@@ -45,20 +40,7 @@ public class PipeTaskExtractorStage extends PipeTaskStage {
PipeParameters extractorParameters,
TConsensusGroupId dataRegionId,
PipeTaskMeta pipeTaskMeta) {
- // Convert the value of `EXTRACTOR_KEY` or `SOURCE_KEY` to lowercase for matching
- // `IOTDB_EXTRACTOR`
- final String pluginName =
- extractorParameters
- .getStringOrDefault(
- Arrays.asList(
- PipeExtractorConstant.EXTRACTOR_KEY, PipeExtractorConstant.SOURCE_KEY),
- BuiltinPipePlugin.IOTDB_EXTRACTOR.getPipePluginName())
- .toLowerCase();
- pipeExtractor =
- pluginName.equals(BuiltinPipePlugin.IOTDB_EXTRACTOR.getPipePluginName())
- || pluginName.equals(BuiltinPipePlugin.IOTDB_SOURCE.getPipePluginName())
- ? new IoTDBDataRegionExtractor()
- : PipeAgent.plugin().reflectExtractor(extractorParameters);
+ pipeExtractor = PipeAgent.plugin().reflectExtractor(extractorParameters);
// Validate and customize should be called before createSubtask. this allows extractor exposing
// exceptions in advance.
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskProcessorStage.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskProcessorStage.java
index 6face1a58fd..d050594a9ff 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskProcessorStage.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskProcessorStage.java
@@ -20,10 +20,7 @@
package org.apache.iotdb.db.pipe.task.stage;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
-import org.apache.iotdb.commons.pipe.plugin.builtin.BuiltinPipePlugin;
-import org.apache.iotdb.commons.pipe.plugin.builtin.processor.DoNothingProcessor;
import org.apache.iotdb.db.pipe.agent.PipeAgent;
-import org.apache.iotdb.db.pipe.config.constant.PipeProcessorConstant;
import org.apache.iotdb.db.pipe.config.plugin.configuraion.PipeTaskRuntimeConfiguration;
import org.apache.iotdb.db.pipe.config.plugin.env.PipeTaskRuntimeEnvironment;
import org.apache.iotdb.db.pipe.execution.executor.PipeProcessorSubtaskExecutor;
@@ -62,18 +59,10 @@ public class PipeTaskProcessorStage extends PipeTaskStage {
TConsensusGroupId dataRegionId,
EventSupplier pipeExtractorInputEventSupplier,
BoundedBlockingPendingQueue<Event> pipeConnectorOutputPendingQueue) {
- // Convert the value of `PROCESSOR_KEY` to lowercase for matching `DO_NOTHING_PROCESSOR`
final PipeProcessor pipeProcessor =
- pipeProcessorParameters
- .getStringOrDefault(
- PipeProcessorConstant.PROCESSOR_KEY,
- BuiltinPipePlugin.DO_NOTHING_PROCESSOR.getPipePluginName())
- .toLowerCase()
- .equals(BuiltinPipePlugin.DO_NOTHING_PROCESSOR.getPipePluginName())
- ? new DoNothingProcessor()
- : PipeAgent.plugin().reflectProcessor(pipeProcessorParameters);
+ PipeAgent.plugin().reflectProcessor(pipeProcessorParameters);
- // validate and customize should be called before createSubtask. this allows extractor exposing
+ // Validate and customize should be called before createSubtask. this allows extractor exposing
// exceptions in advance.
try {
// 1. validate processor parameters
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/connector/PipeConnectorSubtaskManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/connector/PipeConnectorSubtaskManager.java
index c1c75b49ba4..6f1f488785c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/connector/PipeConnectorSubtaskManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/connector/PipeConnectorSubtaskManager.java
@@ -20,18 +20,9 @@
package org.apache.iotdb.db.pipe.task.subtask.connector;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
-import org.apache.iotdb.commons.pipe.plugin.builtin.BuiltinPipePlugin;
-import org.apache.iotdb.commons.pipe.plugin.builtin.connector.DoNothingConnector;
import org.apache.iotdb.db.pipe.agent.PipeAgent;
import org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant;
import org.apache.iotdb.db.pipe.config.plugin.configuraion.PipeTaskRuntimeConfiguration;
-import org.apache.iotdb.db.pipe.connector.protocol.airgap.IoTDBAirGapConnector;
-import org.apache.iotdb.db.pipe.connector.protocol.legacy.IoTDBLegacyPipeConnector;
-import org.apache.iotdb.db.pipe.connector.protocol.opcua.OpcUaConnector;
-import org.apache.iotdb.db.pipe.connector.protocol.thrift.async.IoTDBThriftAsyncConnector;
-import org.apache.iotdb.db.pipe.connector.protocol.thrift.sync.IoTDBThriftSyncConnector;
-import org.apache.iotdb.db.pipe.connector.protocol.websocket.WebSocketConnector;
-import org.apache.iotdb.db.pipe.connector.protocol.writeback.WriteBackConnector;
import org.apache.iotdb.db.pipe.execution.executor.PipeConnectorSubtaskExecutor;
import org.apache.iotdb.db.pipe.task.connection.BoundedBlockingPendingQueue;
import org.apache.iotdb.pipe.api.PipeConnector;
@@ -47,13 +38,9 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
-import java.util.function.Supplier;
public class PipeConnectorSubtaskManager {
- private static final Map<String, Supplier<PipeConnector>> CONNECTOR_CONSTRUCTORS =
- new HashMap<>();
-
private static final String FAILED_TO_DEREGISTER_EXCEPTION_MESSAGE =
"Failed to deregister PipeConnectorSubtask. No such subtask: ";
@@ -77,15 +64,6 @@ public class PipeConnectorSubtaskManager {
final List<PipeConnectorSubtaskLifeCycle> pipeConnectorSubtaskLifeCycleList =
new ArrayList<>(connectorNum);
- // Convert the value of `CONNECTOR_KEY` or `SINK_KEY` to lowercase for matching in
- // `CONNECTOR_CONSTRUCTORS`
- final String connectorKey =
- pipeConnectorParameters
- .getStringOrDefault(
- Arrays.asList(
- PipeConnectorConstant.CONNECTOR_KEY, PipeConnectorConstant.SINK_KEY),
- BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR.getPipePluginName())
- .toLowerCase();
// Shared pending queue for all subtasks
final BoundedBlockingPendingQueue<Event> pendingQueue =
new BoundedBlockingPendingQueue<>(
@@ -93,11 +71,7 @@ public class PipeConnectorSubtaskManager {
for (int connectorIndex = 0; connectorIndex < connectorNum; connectorIndex++) {
final PipeConnector pipeConnector =
- CONNECTOR_CONSTRUCTORS
- .getOrDefault(
- connectorKey,
- () -> PipeAgent.plugin().reflectConnector(pipeConnectorParameters))
- .get();
+ PipeAgent.plugin().reflectConnector(pipeConnectorParameters);
// 1. Construct, validate and customize PipeConnector, and then handshake (create
// connection) with the target
@@ -194,50 +168,7 @@ public class PipeConnectorSubtaskManager {
///////////////////////// Singleton Instance Holder /////////////////////////
private PipeConnectorSubtaskManager() {
- CONNECTOR_CONSTRUCTORS.put(
- BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR.getPipePluginName(),
- IoTDBThriftAsyncConnector::new);
- CONNECTOR_CONSTRUCTORS.put(
- BuiltinPipePlugin.IOTDB_THRIFT_SYNC_CONNECTOR.getPipePluginName(),
- IoTDBThriftSyncConnector::new);
- CONNECTOR_CONSTRUCTORS.put(
- BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName(),
- IoTDBThriftAsyncConnector::new);
- CONNECTOR_CONSTRUCTORS.put(
- BuiltinPipePlugin.IOTDB_LEGACY_PIPE_CONNECTOR.getPipePluginName(),
- IoTDBLegacyPipeConnector::new);
- CONNECTOR_CONSTRUCTORS.put(
- BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName(), IoTDBAirGapConnector::new);
- CONNECTOR_CONSTRUCTORS.put(
- BuiltinPipePlugin.WEBSOCKET_CONNECTOR.getPipePluginName(), WebSocketConnector::new);
- CONNECTOR_CONSTRUCTORS.put(
- BuiltinPipePlugin.OPC_UA_CONNECTOR.getPipePluginName(), OpcUaConnector::new);
- CONNECTOR_CONSTRUCTORS.put(
- BuiltinPipePlugin.DO_NOTHING_CONNECTOR.getPipePluginName(), DoNothingConnector::new);
- CONNECTOR_CONSTRUCTORS.put(
- BuiltinPipePlugin.WRITE_BACK_CONNECTOR.getPipePluginName(), WriteBackConnector::new);
-
- CONNECTOR_CONSTRUCTORS.put(
- BuiltinPipePlugin.IOTDB_THRIFT_SINK.getPipePluginName(), IoTDBThriftAsyncConnector::new);
- CONNECTOR_CONSTRUCTORS.put(
- BuiltinPipePlugin.IOTDB_THRIFT_SYNC_SINK.getPipePluginName(),
- IoTDBThriftSyncConnector::new);
- CONNECTOR_CONSTRUCTORS.put(
- BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_SINK.getPipePluginName(),
- IoTDBThriftAsyncConnector::new);
- CONNECTOR_CONSTRUCTORS.put(
- BuiltinPipePlugin.IOTDB_LEGACY_PIPE_SINK.getPipePluginName(),
- IoTDBLegacyPipeConnector::new);
- CONNECTOR_CONSTRUCTORS.put(
- BuiltinPipePlugin.IOTDB_AIR_GAP_SINK.getPipePluginName(), IoTDBAirGapConnector::new);
- CONNECTOR_CONSTRUCTORS.put(
- BuiltinPipePlugin.WEBSOCKET_SINK.getPipePluginName(), WebSocketConnector::new);
- CONNECTOR_CONSTRUCTORS.put(
- BuiltinPipePlugin.OPC_UA_SINK.getPipePluginName(), OpcUaConnector::new);
- CONNECTOR_CONSTRUCTORS.put(
- BuiltinPipePlugin.DO_NOTHING_SINK.getPipePluginName(), DoNothingConnector::new);
- CONNECTOR_CONSTRUCTORS.put(
- BuiltinPipePlugin.WRITE_BACK_SINK.getPipePluginName(), WriteBackConnector::new);
+ // Do nothing
}
private static class PipeSubtaskManagerHolder {
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 5ad9b4ea985..5312eb28655 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
@@ -97,6 +97,7 @@ import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.PathNotExistException;
import org.apache.iotdb.db.exception.metadata.SchemaQuotaExceededException;
import org.apache.iotdb.db.exception.sql.SemanticException;
+import org.apache.iotdb.db.pipe.agent.PipeAgent;
import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
@@ -1545,6 +1546,16 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
@Override
public SettableFuture<ConfigTaskResult> createPipe(CreatePipeStatement createPipeStatement) {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+
+ // Validate before creation
+ try {
+ PipeAgent.plugin().validate(createPipeStatement);
+ } catch (Exception e) {
+ future.setException(
+ new IoTDBException(e.getMessage(), TSStatusCode.PIPE_ERROR.getStatusCode()));
+ return future;
+ }
+
try (ConfigNodeClient configNodeClient =
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
TCreatePipeReq req =
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/plugin/PipePluginAgentTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/plugin/PipePluginAgentTest.java
index 5c656cecf83..50e21d280b9 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/plugin/PipePluginAgentTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/plugin/PipePluginAgentTest.java
@@ -25,6 +25,8 @@ import org.apache.iotdb.commons.pipe.plugin.service.PipePluginClassLoaderManager
import org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant;
import org.apache.iotdb.db.pipe.config.constant.PipeExtractorConstant;
import org.apache.iotdb.db.pipe.config.constant.PipeProcessorConstant;
+import org.apache.iotdb.db.pipe.connector.protocol.thrift.async.IoTDBThriftAsyncConnector;
+import org.apache.iotdb.db.pipe.extractor.IoTDBDataRegionExtractor;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.junit.After;
@@ -76,7 +78,7 @@ public class PipePluginAgentTest {
Assert.fail();
}
Assert.assertEquals(
- BuiltinPipePlugin.IOTDB_EXTRACTOR.getPipePluginClass(),
+ IoTDBDataRegionExtractor.class,
agent
.reflectExtractor(
new PipeParameters(
@@ -102,7 +104,7 @@ public class PipePluginAgentTest {
}))
.getClass());
Assert.assertEquals(
- BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR.getPipePluginClass(),
+ IoTDBThriftAsyncConnector.class,
agent
.reflectConnector(
new PipeParameters(