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/12/14 04:01:58 UTC
(iotdb) branch pipe-ssl-refactor created (now 80361e8d0f7)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a change to branch pipe-ssl-refactor
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 80361e8d0f7 Pipe: support kv attrs that dont have sink. processor. sink. prefixes
This branch includes the following new commits:
new b72075ec006 Pipe: iotdb-ssl-sink
new c639d2cdd71 Pipe: allow not specifying connector in kv attrs
new b1a0d76fff6 Update PipeParameters.java
new 80361e8d0f7 Pipe: support kv attrs that dont have sink. processor. sink. prefixes
The 4 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) 01/04: Pipe: iotdb-ssl-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-ssl-refactor
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit b72075ec006f0ef5fe5cc7f291128f6d4ce1f54b
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Wed Dec 13 16:32:46 2023 +0800
Pipe: iotdb-ssl-sink
---
.../agent/plugin/PipeConnectorConstructor.java | 4 +++
.../thrift/sync/IoTDBThriftSyncConnector.java | 32 +++++++++++++++-------
.../pipe/plugin/builtin/BuiltinPipePlugin.java | 8 +++++-
.../builtin/connector/IoTDBSslConnector.java | 28 +++++++++++++++++++
4 files changed, 61 insertions(+), 11 deletions(-)
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
index 0137f1c3c7e..6ab65c67282 100644
--- 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
@@ -56,6 +56,8 @@ public class PipeConnectorConstructor extends PipePluginConstructor {
PLUGIN_CONSTRUCTORS.put(
BuiltinPipePlugin.IOTDB_LEGACY_PIPE_CONNECTOR.getPipePluginName(),
IoTDBLegacyPipeConnector::new);
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.IOTDB_SSL_CONNECTOR.getPipePluginName(), IoTDBThriftSyncConnector::new);
PLUGIN_CONSTRUCTORS.put(
BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName(), IoTDBAirGapConnector::new);
PLUGIN_CONSTRUCTORS.put(
@@ -78,6 +80,8 @@ public class PipeConnectorConstructor extends PipePluginConstructor {
PLUGIN_CONSTRUCTORS.put(
BuiltinPipePlugin.IOTDB_LEGACY_PIPE_SINK.getPipePluginName(),
IoTDBLegacyPipeConnector::new);
+ PLUGIN_CONSTRUCTORS.put(
+ BuiltinPipePlugin.IOTDB_SSL_SINK.getPipePluginName(), IoTDBThriftSyncConnector::new);
PLUGIN_CONSTRUCTORS.put(
BuiltinPipePlugin.IOTDB_AIR_GAP_SINK.getPipePluginName(), IoTDBAirGapConnector::new);
PLUGIN_CONSTRUCTORS.put(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncConnector.java
index 2f50a7081b0..c85d2cfa6f6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncConnector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncConnector.java
@@ -67,9 +67,13 @@ import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
+import static org.apache.iotdb.commons.pipe.plugin.builtin.BuiltinPipePlugin.IOTDB_SSL_CONNECTOR;
+import static org.apache.iotdb.commons.pipe.plugin.builtin.BuiltinPipePlugin.IOTDB_SSL_SINK;
+import static org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_KEY;
import static org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_SSL_ENABLE_KEY;
import static org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY;
import static org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY;
+import static org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.SINK_KEY;
public class IoTDBThriftSyncConnector extends IoTDBConnector {
@@ -81,7 +85,7 @@ public class IoTDBThriftSyncConnector extends IoTDBConnector {
private final List<Boolean> isClientAlive = new ArrayList<>();
private boolean useSSL;
- private String trustStore;
+ private String trustStorePath;
private String trustStorePwd;
private long currentClientIndex = 0;
@@ -98,7 +102,10 @@ public class IoTDBThriftSyncConnector extends IoTDBConnector {
final IoTDBConfig ioTDBConfig = IoTDBDescriptor.getInstance().getConfig();
final PipeParameters parameters = validator.getParameters();
- Set<TEndPoint> givenNodeUrls = parseNodeUrls(parameters);
+
+ final String userSpecifiedConnectorName =
+ parameters.getStringByKeys(CONNECTOR_KEY, SINK_KEY).toLowerCase();
+ final Set<TEndPoint> givenNodeUrls = parseNodeUrls(parameters);
validator
.validate(
@@ -122,11 +129,11 @@ public class IoTDBThriftSyncConnector extends IoTDBConnector {
.validate(
args -> !((boolean) args[0]) || ((boolean) args[1] && (boolean) args[2]),
String.format(
- "When %s is specified to true, %s and %s must be specified",
- SINK_IOTDB_SSL_ENABLE_KEY,
- SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY,
- SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY),
- parameters.getBooleanOrDefault(SINK_IOTDB_SSL_ENABLE_KEY, false),
+ "When ssl transport is enabled, %s and %s must be specified",
+ SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY, SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY),
+ IOTDB_SSL_CONNECTOR.getPipePluginName().equals(userSpecifiedConnectorName)
+ || IOTDB_SSL_SINK.getClassName().equals(userSpecifiedConnectorName)
+ || parameters.getBooleanOrDefault(SINK_IOTDB_SSL_ENABLE_KEY, false),
parameters.hasAttribute(SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY),
parameters.hasAttribute(SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY));
}
@@ -145,8 +152,13 @@ public class IoTDBThriftSyncConnector extends IoTDBConnector {
tabletBatchBuilder = new IoTDBThriftSyncPipeTransferBatchReqBuilder(parameters);
}
- useSSL = parameters.getBooleanOrDefault(SINK_IOTDB_SSL_ENABLE_KEY, false);
- trustStore = parameters.getString(SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY);
+ final String userSpecifiedConnectorName =
+ parameters.getStringByKeys(CONNECTOR_KEY, SINK_KEY).toLowerCase();
+ useSSL =
+ IOTDB_SSL_CONNECTOR.getPipePluginName().equals(userSpecifiedConnectorName)
+ || IOTDB_SSL_SINK.getClassName().equals(userSpecifiedConnectorName)
+ || parameters.getBooleanOrDefault(SINK_IOTDB_SSL_ENABLE_KEY, false);
+ trustStorePath = parameters.getString(SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY);
trustStorePwd = parameters.getString(SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY);
}
@@ -184,7 +196,7 @@ public class IoTDBThriftSyncConnector extends IoTDBConnector {
ip,
port,
useSSL,
- trustStore,
+ trustStorePath,
trustStorePwd));
try {
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 f2063aa80e2..d0e4c6af6af 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
@@ -22,6 +22,7 @@ package org.apache.iotdb.commons.pipe.plugin.builtin;
import org.apache.iotdb.commons.pipe.plugin.builtin.connector.DoNothingConnector;
import org.apache.iotdb.commons.pipe.plugin.builtin.connector.IoTDBAirGapConnector;
import org.apache.iotdb.commons.pipe.plugin.builtin.connector.IoTDBLegacyPipeConnector;
+import org.apache.iotdb.commons.pipe.plugin.builtin.connector.IoTDBSslConnector;
import org.apache.iotdb.commons.pipe.plugin.builtin.connector.IoTDBThriftAsyncConnector;
import org.apache.iotdb.commons.pipe.plugin.builtin.connector.IoTDBThriftConnector;
import org.apache.iotdb.commons.pipe.plugin.builtin.connector.IoTDBThriftSyncConnector;
@@ -54,6 +55,7 @@ public enum BuiltinPipePlugin {
IOTDB_THRIFT_SYNC_CONNECTOR("iotdb-thrift-sync-connector", IoTDBThriftSyncConnector.class),
IOTDB_THRIFT_ASYNC_CONNECTOR("iotdb-thrift-async-connector", IoTDBThriftAsyncConnector.class),
IOTDB_LEGACY_PIPE_CONNECTOR("iotdb-legacy-pipe-connector", IoTDBLegacyPipeConnector.class),
+ IOTDB_SSL_CONNECTOR("iotdb-ssl-connector", IoTDBSslConnector.class),
IOTDB_AIR_GAP_CONNECTOR("iotdb-air-gap-connector", IoTDBAirGapConnector.class),
WEBSOCKET_CONNECTOR("websocket-connector", WebSocketConnector.class),
OPC_UA_CONNECTOR("opc-ua-connector", OpcUaConnector.class),
@@ -64,6 +66,7 @@ public enum BuiltinPipePlugin {
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_SSL_SINK("iotdb-ssl-sink", IoTDBSslConnector.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),
@@ -106,6 +109,7 @@ public enum BuiltinPipePlugin {
IOTDB_THRIFT_SYNC_CONNECTOR.getPipePluginName().toUpperCase(),
IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName().toUpperCase(),
IOTDB_LEGACY_PIPE_CONNECTOR.getPipePluginName().toUpperCase(),
+ IOTDB_SSL_CONNECTOR.getPipePluginName().toUpperCase(),
IOTDB_AIR_GAP_CONNECTOR.getPipePluginName().toUpperCase(),
WEBSOCKET_CONNECTOR.getPipePluginName().toUpperCase(),
OPC_UA_CONNECTOR.getPipePluginName().toUpperCase(),
@@ -114,5 +118,7 @@ public enum BuiltinPipePlugin {
IOTDB_THRIFT_SYNC_SINK.getPipePluginName().toUpperCase(),
IOTDB_THRIFT_ASYNC_SINK.getPipePluginName().toUpperCase(),
IOTDB_LEGACY_PIPE_SINK.getPipePluginName().toUpperCase(),
- WEBSOCKET_SINK.getPipePluginName().toUpperCase())));
+ WEBSOCKET_SINK.getPipePluginName().toUpperCase(),
+ OPC_UA_SINK.getPipePluginName().toUpperCase(),
+ WRITE_BACK_SINK.getPipePluginName().toUpperCase())));
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/connector/IoTDBSslConnector.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/connector/IoTDBSslConnector.java
new file mode 100644
index 00000000000..00d8790185a
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/connector/IoTDBSslConnector.java
@@ -0,0 +1,28 @@
+/*
+ * 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.commons.pipe.plugin.builtin.connector;
+
+/**
+ * This class is a placeholder and should not be initialized. It represents the IoTDB SSL connector.
+ * There is a real implementation in the server module but cannot be imported here. The pipe agent
+ * in the server module will replace this class with the real implementation when initializing the
+ * IoTDB SSL connector.
+ */
+public class IoTDBSslConnector extends PlaceholderConnector {}
(iotdb) 02/04: Pipe: allow not specifying connector in kv attrs
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch pipe-ssl-refactor
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit c639d2cdd71aec68425144a4174f68b4adcf66d9
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Wed Dec 13 16:42:19 2023 +0800
Pipe: allow not specifying connector in kv attrs
---
.../iotdb/db/pipe/agent/plugin/PipeConnectorConstructor.java | 8 --------
1 file changed, 8 deletions(-)
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
index 6ab65c67282..5422d6fc29f 100644
--- 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
@@ -32,7 +32,6 @@ 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;
@@ -95,13 +94,6 @@ public class PipeConnectorConstructor extends PipePluginConstructor {
@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
(iotdb) 04/04: Pipe: support kv attrs that dont have sink. processor. sink. prefixes
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch pipe-ssl-refactor
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 80361e8d0f745e14304c462ff453fc22b4691445
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Thu Dec 14 12:01:50 2023 +0800
Pipe: support kv attrs that dont have sink. processor. sink. prefixes
---
.../api/customizer/parameter/PipeParameters.java | 97 ++++++++++++++--------
1 file changed, 63 insertions(+), 34 deletions(-)
diff --git a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameters.java b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameters.java
index 1fe6a635c4d..89ebaa6bcb7 100644
--- a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameters.java
+++ b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameters.java
@@ -26,18 +26,20 @@ import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeCo
import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeProcessorRuntimeConfiguration;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
/**
* Used in {@link PipeExtractor#customize(PipeParameters, PipeExtractorRuntimeConfiguration)} ,
* {@link PipeProcessor#customize(PipeParameters, PipeProcessorRuntimeConfiguration)} and {@link
* PipeConnector#customize(PipeParameters, PipeConnectorRuntimeConfiguration)}.
*
- * <p>This class is used to parse the parameters in WITH PROCESSOR and WITH CONNECTOR when creating
- * a pipe.
+ * <p>This class is used to parse the parameters in WITH SOURCE, WITH PROCESSOR and WITH SINK when
+ * creating a pipe.
*
- * <p>The input parameters is the key-value pair attributes for customization.
+ * <p>The input parameters are the key-value pair attributes for customization.
*/
public class PipeParameters {
@@ -52,12 +54,12 @@ public class PipeParameters {
}
public boolean hasAttribute(String key) {
- return attributes.containsKey(key);
+ return attributes.containsKey(key) || attributes.containsKey(KeyReducer.reduce(key));
}
public boolean hasAnyAttributes(String... keys) {
for (final String key : keys) {
- if (attributes.containsKey(key)) {
+ if (hasAttribute(key)) {
return true;
}
}
@@ -65,37 +67,38 @@ public class PipeParameters {
}
public String getString(String key) {
- return attributes.get(key);
+ final String value = attributes.get(key);
+ return value != null ? value : attributes.get(KeyReducer.reduce(key));
}
public Boolean getBoolean(String key) {
- String value = attributes.get(key);
+ final String value = getString(key);
return value == null ? null : Boolean.parseBoolean(value);
}
public Integer getInt(String key) {
- String value = attributes.get(key);
+ final String value = getString(key);
return value == null ? null : Integer.parseInt(value);
}
public Long getLong(String key) {
- String value = attributes.get(key);
+ final String value = getString(key);
return value == null ? null : Long.parseLong(value);
}
public Float getFloat(String key) {
- String value = attributes.get(key);
+ final String value = getString(key);
return value == null ? null : Float.parseFloat(value);
}
public Double getDouble(String key) {
- String value = attributes.get(key);
+ final String value = getString(key);
return value == null ? null : Double.parseDouble(value);
}
public String getStringByKeys(String... keys) {
for (final String key : keys) {
- final String value = attributes.get(key);
+ final String value = getString(key);
if (value != null) {
return value;
}
@@ -105,9 +108,9 @@ public class PipeParameters {
public Boolean getBooleanByKeys(String... keys) {
for (final String key : keys) {
- final String value = attributes.get(key);
+ final Boolean value = getBoolean(key);
if (value != null) {
- return Boolean.parseBoolean(value);
+ return value;
}
}
return null;
@@ -115,9 +118,9 @@ public class PipeParameters {
public Integer getIntByKeys(String... keys) {
for (final String key : keys) {
- final String value = attributes.get(key);
+ final Integer value = getInt(key);
if (value != null) {
- return Integer.parseInt(value);
+ return value;
}
}
return null;
@@ -125,9 +128,9 @@ public class PipeParameters {
public Long getLongByKeys(String... keys) {
for (final String key : keys) {
- final String value = attributes.get(key);
+ final Long value = getLong(key);
if (value != null) {
- return Long.parseLong(value);
+ return value;
}
}
return null;
@@ -135,9 +138,9 @@ public class PipeParameters {
public Float getFloatByKeys(String... keys) {
for (final String key : keys) {
- final String value = attributes.get(key);
+ final Float value = getFloat(key);
if (value != null) {
- return Float.parseFloat(value);
+ return value;
}
}
return null;
@@ -145,47 +148,47 @@ public class PipeParameters {
public Double getDoubleByKeys(String... keys) {
for (final String key : keys) {
- final String value = attributes.get(key);
+ final Double value = getDouble(key);
if (value != null) {
- return Double.parseDouble(value);
+ return value;
}
}
return null;
}
public String getStringOrDefault(String key, String defaultValue) {
- String value = attributes.get(key);
+ final String value = getString(key);
return value == null ? defaultValue : value;
}
public boolean getBooleanOrDefault(String key, boolean defaultValue) {
- String value = attributes.get(key);
+ final String value = getString(key);
return value == null ? defaultValue : Boolean.parseBoolean(value);
}
public int getIntOrDefault(String key, int defaultValue) {
- String value = attributes.get(key);
+ final String value = getString(key);
return value == null ? defaultValue : Integer.parseInt(value);
}
public long getLongOrDefault(String key, long defaultValue) {
- String value = attributes.get(key);
+ final String value = getString(key);
return value == null ? defaultValue : Long.parseLong(value);
}
public float getFloatOrDefault(String key, float defaultValue) {
- String value = attributes.get(key);
+ final String value = getString(key);
return value == null ? defaultValue : Float.parseFloat(value);
}
public double getDoubleOrDefault(String key, double defaultValue) {
- String value = attributes.get(key);
+ final String value = getString(key);
return value == null ? defaultValue : Double.parseDouble(value);
}
public String getStringOrDefault(List<String> keys, String defaultValue) {
for (final String key : keys) {
- final String value = attributes.get(key);
+ final String value = getString(key);
if (value != null) {
return value;
}
@@ -195,7 +198,7 @@ public class PipeParameters {
public boolean getBooleanOrDefault(List<String> keys, boolean defaultValue) {
for (final String key : keys) {
- final String value = attributes.get(key);
+ final String value = getString(key);
if (value != null) {
return Boolean.parseBoolean(value);
}
@@ -205,7 +208,7 @@ public class PipeParameters {
public int getIntOrDefault(List<String> keys, int defaultValue) {
for (final String key : keys) {
- final String value = attributes.get(key);
+ final String value = getString(key);
if (value != null) {
return Integer.parseInt(value);
}
@@ -215,7 +218,7 @@ public class PipeParameters {
public long getLongOrDefault(List<String> keys, long defaultValue) {
for (final String key : keys) {
- final String value = attributes.get(key);
+ final String value = getString(key);
if (value != null) {
return Long.parseLong(value);
}
@@ -225,7 +228,7 @@ public class PipeParameters {
public float getFloatOrDefault(List<String> keys, float defaultValue) {
for (final String key : keys) {
- final String value = attributes.get(key);
+ final String value = getString(key);
if (value != null) {
return Float.parseFloat(value);
}
@@ -235,7 +238,7 @@ public class PipeParameters {
public double getDoubleOrDefault(List<String> keys, double defaultValue) {
for (final String key : keys) {
- final String value = attributes.get(key);
+ final String value = getString(key);
if (value != null) {
return Double.parseDouble(value);
}
@@ -264,4 +267,30 @@ public class PipeParameters {
public String toString() {
return attributes.toString();
}
+
+ private static class KeyReducer {
+
+ private static final Set<String> PREFIXES = new HashSet<>();
+
+ static {
+ PREFIXES.add("extractor.");
+ PREFIXES.add("source.");
+ PREFIXES.add("processor.");
+ PREFIXES.add("connector.");
+ PREFIXES.add("sink.");
+ }
+
+ static String reduce(String key) {
+ if (key == null) {
+ return null;
+ }
+ final String lowerCaseKey = key.toLowerCase();
+ for (final String prefix : PREFIXES) {
+ if (lowerCaseKey.startsWith(prefix)) {
+ return key.substring(prefix.length());
+ }
+ }
+ return key;
+ }
+ }
}
(iotdb) 03/04: Update PipeParameters.java
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch pipe-ssl-refactor
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit b1a0d76fff6f98bce5fb15a844942be23eebbef8
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Wed Dec 13 19:09:30 2023 +0800
Update PipeParameters.java
---
.../apache/iotdb/pipe/api/customizer/parameter/PipeParameters.java | 7 +++++--
1 file changed, 5 insertions(+), 2 deletions(-)
diff --git a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameters.java b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameters.java
index c896825f342..1fe6a635c4d 100644
--- a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameters.java
+++ b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameters.java
@@ -20,16 +20,19 @@
package org.apache.iotdb.pipe.api.customizer.parameter;
import org.apache.iotdb.pipe.api.PipeConnector;
+import org.apache.iotdb.pipe.api.PipeExtractor;
import org.apache.iotdb.pipe.api.PipeProcessor;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
+import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeProcessorRuntimeConfiguration;
import java.util.List;
import java.util.Map;
/**
- * Used in {@link PipeProcessor#customize(PipeParameters, PipeProcessorRuntimeConfiguration)} and
- * {@link PipeConnector#customize(PipeParameters, PipeConnectorRuntimeConfiguration)}.
+ * Used in {@link PipeExtractor#customize(PipeParameters, PipeExtractorRuntimeConfiguration)} ,
+ * {@link PipeProcessor#customize(PipeParameters, PipeProcessorRuntimeConfiguration)} and {@link
+ * PipeConnector#customize(PipeParameters, PipeConnectorRuntimeConfiguration)}.
*
* <p>This class is used to parse the parameters in WITH PROCESSOR and WITH CONNECTOR when creating
* a pipe.