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/10/31 11:38:27 UTC
(iotdb) branch pipe-plugin-rename created (now 538e014e352)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a change to branch pipe-plugin-rename
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 538e014e352 Merge branch 'master' of https://github.com/apache/iotdb into pipe-plugin-rename
This branch includes the following new commits:
new f17a85f92e7 Pipe: rename plugins *-extractor -> *-source, *-connector -> *-sink
new 538e014e352 Merge branch 'master' of https://github.com/apache/iotdb into pipe-plugin-rename
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 02/02: Merge branch 'master' of https://github.com/apache/iotdb into pipe-plugin-rename
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch pipe-plugin-rename
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 538e014e352cec7471baaf145745479f24769aca
Merge: f17a85f92e7 ad980dbf973
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Tue Oct 31 19:38:01 2023 +0800
Merge branch 'master' of https://github.com/apache/iotdb into pipe-plugin-rename
.../apache/iotdb/db/storageengine/StorageEngine.java | 18 +++++++++++-------
.../iotdb/db/storageengine/dataregion/DataRegion.java | 2 +-
.../InsertionCrossSpaceCompactionRecoverTest.java | 2 +-
.../InsertionCrossSpaceCompactionSelectorTest.java | 2 +-
4 files changed, 14 insertions(+), 10 deletions(-)
(iotdb) 01/02: Pipe: rename plugins *-extractor -> *-source, *-connector -> *-sink
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch pipe-plugin-rename
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit f17a85f92e7432808bd074c5c9d9e5a8020088bd
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Tue Oct 31 19:37:19 2023 +0800
Pipe: rename plugins *-extractor -> *-source, *-connector -> *-sink
---
.../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;