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 f94565d1919 Pipe: rename plugins *-extractor -> *-source, *-connector
-> *-sink (#11440)
f94565d1919 is described below
commit f94565d1919a50f03551b80ede1226a964966818
Author: Steve Yurong Su <[email protected]>
AuthorDate: Tue Oct 31 19:49:43 2023 +0800
Pipe: rename plugins *-extractor -> *-source, *-connector -> *-sink (#11440)
---
.../db/pipe/task/stage/PipeTaskExtractorStage.java | 12 +++++------
.../connector/PipeConnectorSubtaskManager.java | 23 ++++++++++++++++++++++
.../pipe/plugin/builtin/BuiltinPipePlugin.java | 11 +++++++++++
3 files changed, 40 insertions(+), 6 deletions(-)
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 2798d18fa61..83fd4d8e098 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
@@ -45,13 +45,13 @@ public class PipeTaskExtractorStage extends PipeTaskStage {
PipeParameters extractorParameters,
TConsensusGroupId dataRegionId,
PipeTaskMeta pipeTaskMeta) {
+ final String pluginName =
+ extractorParameters.getStringOrDefault(
+ Arrays.asList(PipeExtractorConstant.EXTRACTOR_KEY,
PipeExtractorConstant.SOURCE_KEY),
+ BuiltinPipePlugin.IOTDB_EXTRACTOR.getPipePluginName());
pipeExtractor =
- extractorParameters
- .getStringOrDefault(
- Arrays.asList(
- PipeExtractorConstant.EXTRACTOR_KEY,
PipeExtractorConstant.SOURCE_KEY),
- BuiltinPipePlugin.IOTDB_EXTRACTOR.getPipePluginName())
- .equals(BuiltinPipePlugin.IOTDB_EXTRACTOR.getPipePluginName())
+
pluginName.equals(BuiltinPipePlugin.IOTDB_EXTRACTOR.getPipePluginName())
+ ||
pluginName.equals(BuiltinPipePlugin.IOTDB_SOURCE.getPipePluginName())
? new IoTDBDataRegionExtractor()
: PipeAgent.plugin().reflectExtractor(extractorParameters);
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 6450247a53f..52fbf18aef4 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
@@ -21,6 +21,7 @@ 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;
@@ -199,6 +200,28 @@ public class PipeConnectorSubtaskManager {
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.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);
}
private static class PipeSubtaskManagerHolder {
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/BuiltinPipePlugin.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/BuiltinPipePlugin.java
index 9218e29df24..54e21941cac 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/BuiltinPipePlugin.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/BuiltinPipePlugin.java
@@ -35,6 +35,8 @@ public enum BuiltinPipePlugin {
// extractors
IOTDB_EXTRACTOR("iotdb-extractor", IoTDBExtractor.class),
+ IOTDB_SOURCE("iotdb-source", IoTDBExtractor.class),
+
// processors
DO_NOTHING_PROCESSOR("do-nothing-processor", DoNothingProcessor.class),
@@ -47,6 +49,15 @@ public enum BuiltinPipePlugin {
IOTDB_AIR_GAP_CONNECTOR("iotdb-air-gap-connector",
IoTDBAirGapConnector.class),
WEBSOCKET_CONNECTOR("websocket-connector", WebSocketConnector.class),
OPC_UA_CONNECTOR("opc-ua-connector", OpcUaConnector.class),
+
+ DO_NOTHING_SINK("do-nothing-sink", DoNothingConnector.class),
+ IOTDB_THRIFT_SINK("iotdb-thrift-sink", IoTDBThriftConnector.class),
+ IOTDB_THRIFT_SYNC_SINK("iotdb-thrift-sync-sink",
IoTDBThriftSyncConnector.class),
+ IOTDB_THRIFT_ASYNC_SINK("iotdb-thrift-async-sink",
IoTDBThriftAsyncConnector.class),
+ IOTDB_LEGACY_PIPE_SINK("iotdb-legacy-pipe-sink",
IoTDBLegacyPipeConnector.class),
+ IOTDB_AIR_GAP_SINK("iotdb-air-gap-sink", IoTDBAirGapConnector.class),
+ WEBSOCKET_SINK("websocket-sink", WebSocketConnector.class),
+ OPC_UA_SINK("opc-ua-sink", OpcUaConnector.class),
;
private final String pipePluginName;
