This is an automated email from the ASF dual-hosted git repository.
jt2594838 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 7ffb1364b84 Add mutual SSL support for pipe sinks (#18080)
7ffb1364b84 is described below
commit 7ffb1364b843cfe13aaa57d693922adf333155eb
Author: Caideyipi <[email protected]>
AuthorDate: Mon Jul 13 17:28:31 2026 +0800
Add mutual SSL support for pipe sinks (#18080)
* Add mutual SSL support for pipe sinks
* Fix JDBC default user handling
* Update IoTDBPipeMutualSSLIT.java
* Validate SSL store aliases consistently
* Clarify SSL alias validation messages
* Move pipe SSL validation messages to i18n bundles
---
.../treemodel/auto/basic/IoTDBPipeMutualSSLIT.java | 176 +++++++++++++++++++
.../api/customizer/parameter/PipeParameters.java | 1 +
.../customizer/parameter/PipeParametersTest.java | 12 ++
.../org/apache/iotdb/jdbc/IoTDBConnection.java | 2 +-
.../client/IoTDBConfigNodeSyncClientManager.java | 4 +
.../pipe/sink/protocol/IoTDBConfigRegionSink.java | 4 +
.../pipe/sink/IoTDBConfigRegionSinkTest.java | 46 +++++
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 6 +
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 6 +-
.../client/IoTDBDataNodeSyncClientManager.java | 4 +
.../sink/protocol/legacy/IoTDBLegacyPipeSink.java | 123 +++++++++++--
.../thrift/async/IoTDBDataRegionAsyncSink.java | 27 ++-
.../thrift/sync/IoTDBDataNodeSyncSink.java | 4 +
.../apache/iotdb/db/pipe/sink/PipeSinkTest.java | 193 +++++++++++++++++++++
.../apache/iotdb/commons/i18n/PipeMessages.java | 6 +
.../apache/iotdb/commons/i18n/PipeMessages.java | 4 +
.../pipe/config/constant/PipeSinkConstant.java | 10 ++
.../commons/pipe/sink/client/IoTDBSyncClient.java | 17 +-
.../pipe/sink/client/IoTDBSyncClientManager.java | 10 +-
.../pipe/sink/protocol/IoTDBSslSyncSink.java | 125 +++++++++++--
20 files changed, 748 insertions(+), 32 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeMutualSSLIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeMutualSSLIT.java
new file mode 100644
index 00000000000..6362cca567c
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeMutualSSLIT.java
@@ -0,0 +1,176 @@
+/*
+ * 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.pipe.it.dual.treemodel.auto.basic;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
+import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.isession.SessionConfig;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.MultiClusterIT2DualTreeAutoBasic;
+import org.apache.iotdb.jdbc.Config;
+import
org.apache.iotdb.pipe.it.dual.treemodel.auto.AbstractPipeDualTreeModelAutoIT;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.awaitility.Awaitility;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.io.File;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2DualTreeAutoBasic.class})
+public class IoTDBPipeMutualSSLIT extends AbstractPipeDualTreeModelAutoIT {
+
+ private static final String STORE_PASSWORD = "thrift";
+
+ @Override
+ protected void setupConfig() {
+ super.setupConfig();
+
+ senderEnv
+ .getConfig()
+ .getCommonConfig()
+ .setTrustStorePath(trustStorePath())
+ .setTrustStorePwd(STORE_PASSWORD)
+ .setSslProtocol(SessionConfig.DEFAULT_SSL_PROTOCOL);
+ receiverEnv
+ .getConfig()
+ .getCommonConfig()
+ .setEnableThriftClientSSL(true)
+ .setThriftSSLClientAuth(true)
+ .setKeyStorePath(keyStorePath())
+ .setKeyStorePwd(STORE_PASSWORD)
+ .setTrustStorePath(trustStorePath())
+ .setTrustStorePwd(STORE_PASSWORD)
+ .setSslProtocol(SessionConfig.DEFAULT_SSL_PROTOCOL);
+ }
+
+ @Test
+ public void testPipeCanTransferWithMutualSSL() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient)
senderEnv.getLeaderConfigNodeConnection()) {
+ TestUtils.executeNonQueries(
+ senderEnv,
+ Arrays.asList(
+ "insert into root.pipe_mtls.d1(time, s1) values (1, 11)",
+ "insert into root.pipe_mtls.d1(time, s1) values (2, 22)",
+ "flush"),
+ null);
+
+ final Map<String, String> sourceAttributes = new HashMap<>();
+ final Map<String, String> sinkAttributes = new HashMap<>();
+
+ sourceAttributes.put("source.realtime.mode", "log");
+ sourceAttributes.put("source.user", "root");
+
+ sinkAttributes.put("sink", "iotdb-thrift-ssl-sink");
+ sinkAttributes.put("sink.batch.enable", "false");
+ sinkAttributes.put("sink.ip", receiverDataNode.getIp());
+ sinkAttributes.put("sink.port",
Integer.toString(receiverDataNode.getPort()));
+ sinkAttributes.put("sink.ssl.trust-store-path", trustStorePath());
+ sinkAttributes.put("sink.ssl.trust-store-pwd", STORE_PASSWORD);
+ sinkAttributes.put("sink.ssl.key-store-path", keyStorePath());
+ sinkAttributes.put("sink.ssl.key-store-pwd", STORE_PASSWORD);
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("testPipe", sinkAttributes)
+ .setExtractorAttributes(sourceAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
status.getCode());
+
+ TestUtils.executeNonQueries(
+ senderEnv,
+ Arrays.asList("insert into root.pipe_mtls.d1(time, s1) values (3,
33)", "flush"),
+ null);
+
+ try (final Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX
+ + receiverDataNode.getIpAndPortString()
+ + "?"
+ + Config.USE_SSL
+ + "=true&"
+ + Config.TRUST_STORE
+ + "="
+ + trustStorePath()
+ + "&"
+ + Config.TRUST_STORE_PWD
+ + "="
+ + STORE_PASSWORD
+ + "&"
+ + Config.KEY_STORE
+ + "="
+ + keyStorePath()
+ + "&"
+ + Config.KEY_STORE_PWD
+ + "="
+ + STORE_PASSWORD,
+ SessionConfig.DEFAULT_USER,
+ SessionConfig.DEFAULT_PASSWORD);
+ final Statement statement = connection.createStatement()) {
+ Awaitility.await()
+ .pollInSameThread()
+ .pollDelay(1L, TimeUnit.SECONDS)
+ .pollInterval(1L, TimeUnit.SECONDS)
+ .atMost(600, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ TestUtils.assertResultSetEqual(
+ statement.executeQuery("select s1 from
root.pipe_mtls.d1"),
+ "Time,root.pipe_mtls.d1.s1,",
+ Collections.unmodifiableSet(
+ new HashSet<>(Arrays.asList("1,11.0,", "2,22.0,",
"3,33.0,")))));
+ }
+ }
+ }
+
+ private static String keyStorePath() {
+ return keyDir() + "test-keystore";
+ }
+
+ private static String trustStorePath() {
+ return keyDir() + "test-truststore";
+ }
+
+ private static String keyDir() {
+ return System.getProperty("user.dir")
+ + File.separator
+ + "target"
+ + File.separator
+ + "test-classes"
+ + File.separator;
+ }
+}
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 7286d120991..16a1329a6bf 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
@@ -430,6 +430,7 @@ public class PipeParameters {
static {
KEYS.add("ssl.trust-store-pwd");
+ KEYS.add("ssl.key-store-pwd");
KEYS.add("scp.password");
KEYS.add("password");
}
diff --git
a/iotdb-api/pipe-api/src/test/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParametersTest.java
b/iotdb-api/pipe-api/src/test/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParametersTest.java
index d16b8e2879d..98eed48eace 100644
---
a/iotdb-api/pipe-api/src/test/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParametersTest.java
+++
b/iotdb-api/pipe-api/src/test/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParametersTest.java
@@ -39,4 +39,16 @@ public class PipeParametersTest {
parameters.addAttribute("opcua.sink.value-name", "false");
Assert.assertNull(parameters.getString("value-name"));
}
+
+ @Test
+ public void valueHiderShouldHideSslKeyStorePassword() {
+ Assert.assertEquals(
+ "******", PipeParameters.ValueHider.hide("sink.ssl.key-store-pwd",
"secret"));
+ Assert.assertEquals(
+ "******",
PipeParameters.ValueHider.hide("connector.ssl.key-store-pwd", "secret"));
+ Assert.assertEquals("******",
PipeParameters.ValueHider.hide("ssl.key-store-pwd", "secret"));
+ Assert.assertEquals(
+ "******",
PipeParameters.ValueHider.hide("connector.ssl.trust-store-pwd", "secret"));
+ Assert.assertEquals("secret",
PipeParameters.ValueHider.hide("ssl.key-store-path", "secret"));
+ }
}
diff --git
a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnection.java
b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnection.java
index 1192ceff25e..bf04cf69355 100644
--- a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnection.java
+++ b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnection.java
@@ -129,7 +129,7 @@ public class IoTDBConnection implements Connection {
}
params = Utils.parseUrl(url, info);
this.url = url;
- this.userName = info.get("user").toString();
+ this.userName = params.getUsername();
this.networkTimeout = params.getNetworkTimeout();
this.zoneId = ZoneId.of(params.getTimeZone());
this.charset = params.getCharset();
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/client/IoTDBConfigNodeSyncClientManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/client/IoTDBConfigNodeSyncClientManager.java
index c579fad73c1..25c0bc58b61 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/client/IoTDBConfigNodeSyncClientManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/client/IoTDBConfigNodeSyncClientManager.java
@@ -39,6 +39,8 @@ public class IoTDBConfigNodeSyncClientManager extends
IoTDBSyncClientManager {
boolean useSSL,
String trustStorePath,
String trustStorePwd,
+ String keyStorePath,
+ String keyStorePwd,
/* The following parameters are used locally. */
String loadBalanceStrategy,
/* The following parameters are used to handshake with the receiver. */
@@ -54,6 +56,8 @@ public class IoTDBConfigNodeSyncClientManager extends
IoTDBSyncClientManager {
useSSL,
trustStorePath,
trustStorePwd,
+ keyStorePath,
+ keyStorePwd,
false,
loadBalanceStrategy,
userEntity,
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/protocol/IoTDBConfigRegionSink.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/protocol/IoTDBConfigRegionSink.java
index 07e64321de7..636e28a455e 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/protocol/IoTDBConfigRegionSink.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/protocol/IoTDBConfigRegionSink.java
@@ -68,6 +68,8 @@ public class IoTDBConfigRegionSink extends IoTDBSslSyncSink {
final boolean useSSL,
final String trustStorePath,
final String trustStorePwd,
+ final String keyStorePath,
+ final String keyStorePwd,
/* The following parameters are used locally. */
final boolean useLeaderCache,
final String loadBalanceStrategy,
@@ -84,6 +86,8 @@ public class IoTDBConfigRegionSink extends IoTDBSslSyncSink {
useSSL,
Objects.nonNull(trustStorePath) ?
ConfigNodeConfig.addHomeDir(trustStorePath) : null,
trustStorePwd,
+ Objects.nonNull(keyStorePath) ?
ConfigNodeConfig.addHomeDir(keyStorePath) : null,
+ keyStorePwd,
loadBalanceStrategy,
userEntity,
password,
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/sink/IoTDBConfigRegionSinkTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/sink/IoTDBConfigRegionSinkTest.java
index b88862c4cae..01b7c1de668 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/sink/IoTDBConfigRegionSinkTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/sink/IoTDBConfigRegionSinkTest.java
@@ -51,4 +51,50 @@ public class IoTDBConfigRegionSinkTest {
Assert.fail();
}
}
+
+ @Test
+ public void testIoTDBConfigRegionSinkAcceptsMutualSslParameters() {
+ try (IoTDBConfigRegionSink connector = new IoTDBConfigRegionSink()) {
+ connector.validate(
+ new PipeParameterValidator(
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(
+ PipeSinkConstant.CONNECTOR_KEY,
+
BuiltinPipePlugin.IOTDB_THRIFT_SSL_CONNECTOR.getPipePluginName());
+ put(PipeSinkConstant.CONNECTOR_IOTDB_NODE_URLS_KEY,
"127.0.0.1:6668");
+
put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_TRUST_STORE_PATH_KEY, "truststore");
+
put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_TRUST_STORE_PWD_KEY, "trustpwd");
+
put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_KEY_STORE_PATH_KEY, "keystore");
+
put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_KEY_STORE_PWD_KEY, "keypwd");
+ }
+ })));
+ } catch (Exception e) {
+ Assert.fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testIoTDBConfigRegionSinkRejectsIncompleteKeyStoreParameters() {
+ try (IoTDBConfigRegionSink connector = new IoTDBConfigRegionSink()) {
+ connector.validate(
+ new PipeParameterValidator(
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(
+ PipeSinkConstant.CONNECTOR_KEY,
+
BuiltinPipePlugin.IOTDB_THRIFT_SSL_CONNECTOR.getPipePluginName());
+ put(PipeSinkConstant.CONNECTOR_IOTDB_NODE_URLS_KEY,
"127.0.0.1:6668");
+
put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_TRUST_STORE_PATH_KEY, "truststore");
+
put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_TRUST_STORE_PWD_KEY, "trustpwd");
+
put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_KEY_STORE_PATH_KEY, "keystore");
+ }
+ })));
+ Assert.fail();
+ } catch (Exception e) {
+
Assert.assertTrue(e.getMessage().contains(PipeSinkConstant.SINK_IOTDB_SSL_KEY_STORE_PWD_KEY));
+ }
+ }
}
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index 4dbdbd8803f..4297e241712 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -945,6 +945,12 @@ public final class DataNodePipeMessages {
"Network failed to receive tsFile %s, status: %s";
public static final String SECURITY_DIR = "security dir: {}";
public static final String SECURITY_PKI_DIR = "security pki dir: {}";
+ public static final String SSL_TRUST_STORE_PAIR_REQUIRED_WHEN_SSL_ENABLED =
+ "When %s or %s is true, specify a complete trust-store pair under the
same "
+ + "alias: %s and %s, %s and %s, or %s and %s";
+ public static final String
SSL_KEY_STORE_PATH_AND_PASSWORD_MUST_BE_SPECIFIED_TOGETHER =
+ "SSL key-store path and password must be specified together under the
same "
+ + "alias: %s and %s, %s and %s, or %s and %s";
public static final String SUCCESSFULLY_ADDED_ITEM = "Successfully added
item {}.";
public static final String SUCCESSFULLY_CONVERTED_PROGID_TO_CLSID =
"Successfully converted progID {} to CLSID: {{}}";
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index f97169709a7..1b38041791d 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -887,11 +887,15 @@ public final class DataNodePipeMessages {
public static final String RECEIVED_A_BIND_MESSAGE_FROM = "从 {}:{} 收到 bind
消息";
public static final String REDIRECT_FILE_POSITION_TO = "重定向 file position 到
{}。";
public static final String REDIRECT_TO_POSITION_IN_TRANSFERRING_TSFILE =
- "正在传输的 tsFile {} 中重定向到 position {}。";
+ "重定向到 position {}，正在传输的 TsFile 为 {}。";
public static final String NETWORK_FAILED_TO_RECEIVE_TSFILE_STATUS =
"网络接收 TsFile %s 失败，状态：%s";
public static final String SECURITY_DIR = "security 目录：{}";
public static final String SECURITY_PKI_DIR = "security pki 目录：{}";
+ public static final String SSL_TRUST_STORE_PAIR_REQUIRED_WHEN_SSL_ENABLED =
+ "当 %s 或 %s 为 true 时，请在同一别名下指定完整的 trust-store 参数对：%s 和 %s、%s 和 %s，或 %s 和
%s";
+ public static final String
SSL_KEY_STORE_PATH_AND_PASSWORD_MUST_BE_SPECIFIED_TOGETHER =
+ "SSL key-store 路径和密码必须在同一别名下同时指定：%s 和 %s、%s 和 %s，或 %s 和 %s";
public static final String SUCCESSFULLY_ADDED_ITEM = "成功添加 item {}。";
public static final String SUCCESSFULLY_CONVERTED_PROGID_TO_CLSID =
"成功将 progID {} 转换为 CLSID: {{}}";
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeSyncClientManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeSyncClientManager.java
index 841982accd3..df966c78acc 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeSyncClientManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeSyncClientManager.java
@@ -49,6 +49,8 @@ public class IoTDBDataNodeSyncClientManager extends
IoTDBSyncClientManager
final boolean useSSL,
final String trustStorePath,
final String trustStorePwd,
+ final String keyStorePath,
+ final String keyStorePwd,
/* The following parameters are used locally. */
final boolean useLeaderCache,
final String loadBalanceStrategy,
@@ -65,6 +67,8 @@ public class IoTDBDataNodeSyncClientManager extends
IoTDBSyncClientManager
useSSL,
trustStorePath,
trustStorePwd,
+ keyStorePath,
+ keyStorePwd,
useLeaderCache,
loadBalanceStrategy,
userEntity,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java
index bf37e02c8d5..a74cfcbd2d8 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java
@@ -83,6 +83,11 @@ import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CON
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_PASSWORD_DEFAULT_VALUE;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_PASSWORD_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_PORT_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_SSL_ENABLE_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_SSL_KEY_STORE_PATH_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_SSL_KEY_STORE_PWD_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_SSL_TRUST_STORE_PATH_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_SSL_TRUST_STORE_PWD_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_SYNC_CONNECTOR_VERSION_DEFAULT_VALUE;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_SYNC_CONNECTOR_VERSION_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_USERNAME_KEY;
@@ -92,6 +97,8 @@ import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SIN
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_PASSWORD_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_PORT_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_SSL_ENABLE_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_SSL_KEY_STORE_PATH_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_SSL_KEY_STORE_PWD_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_SYNC_CONNECTOR_VERSION_KEY;
@@ -112,6 +119,8 @@ public class IoTDBLegacyPipeSink implements PipeConnector {
private boolean useSSL;
private String trustStore;
private String trustStorePwd;
+ private String keyStore;
+ private String keyStorePwd;
private String user;
private String password;
@@ -149,13 +158,88 @@ public class IoTDBLegacyPipeSink implements PipeConnector
{
.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",
+
DataNodePipeMessages.SSL_TRUST_STORE_PAIR_REQUIRED_WHEN_SSL_ENABLED,
+ CONNECTOR_IOTDB_SSL_ENABLE_KEY,
SINK_IOTDB_SSL_ENABLE_KEY,
+ CONNECTOR_IOTDB_SSL_TRUST_STORE_PATH_KEY,
+ CONNECTOR_IOTDB_SSL_TRUST_STORE_PWD_KEY,
+ SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY,
+ SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY,
+
PipeParameters.KeyReducer.reduce(CONNECTOR_IOTDB_SSL_TRUST_STORE_PATH_KEY),
+
PipeParameters.KeyReducer.reduce(CONNECTOR_IOTDB_SSL_TRUST_STORE_PWD_KEY)),
+ parameters.getBooleanOrDefault(
+ Arrays.asList(CONNECTOR_IOTDB_SSL_ENABLE_KEY,
SINK_IOTDB_SSL_ENABLE_KEY), false),
+ hasCompleteAttributePair(
+ parameters,
+ CONNECTOR_IOTDB_SSL_TRUST_STORE_PATH_KEY,
+ CONNECTOR_IOTDB_SSL_TRUST_STORE_PWD_KEY,
SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY,
SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY),
- 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));
+ hasNoHalfAttributePair(
+ parameters,
+ CONNECTOR_IOTDB_SSL_TRUST_STORE_PATH_KEY,
+ CONNECTOR_IOTDB_SSL_TRUST_STORE_PWD_KEY,
+ SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY,
+ SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY))
+ .validate(
+ args -> (boolean) args[0] == (boolean) args[1],
+ String.format(
+
DataNodePipeMessages.SSL_KEY_STORE_PATH_AND_PASSWORD_MUST_BE_SPECIFIED_TOGETHER,
+ CONNECTOR_IOTDB_SSL_KEY_STORE_PATH_KEY,
+ CONNECTOR_IOTDB_SSL_KEY_STORE_PWD_KEY,
+ SINK_IOTDB_SSL_KEY_STORE_PATH_KEY,
+ SINK_IOTDB_SSL_KEY_STORE_PWD_KEY,
+
PipeParameters.KeyReducer.reduce(CONNECTOR_IOTDB_SSL_KEY_STORE_PATH_KEY),
+
PipeParameters.KeyReducer.reduce(CONNECTOR_IOTDB_SSL_KEY_STORE_PWD_KEY)),
+ true,
+ hasNoHalfAttributePair(
+ parameters,
+ CONNECTOR_IOTDB_SSL_KEY_STORE_PATH_KEY,
+ CONNECTOR_IOTDB_SSL_KEY_STORE_PWD_KEY,
+ SINK_IOTDB_SSL_KEY_STORE_PATH_KEY,
+ SINK_IOTDB_SSL_KEY_STORE_PWD_KEY));
+ }
+
+ private static boolean hasCompleteAttributePair(
+ final PipeParameters parameters,
+ final String connectorLeftKey,
+ final String connectorRightKey,
+ final String sinkLeftKey,
+ final String sinkRightKey) {
+ return hasExactAttributePair(parameters, connectorLeftKey,
connectorRightKey)
+ || hasExactAttributePair(parameters, sinkLeftKey, sinkRightKey)
+ || hasExactAttributePair(
+ parameters,
+ PipeParameters.KeyReducer.reduce(connectorLeftKey),
+ PipeParameters.KeyReducer.reduce(connectorRightKey));
+ }
+
+ private static boolean hasNoHalfAttributePair(
+ final PipeParameters parameters,
+ final String connectorLeftKey,
+ final String connectorRightKey,
+ final String sinkLeftKey,
+ final String sinkRightKey) {
+ return hasBothOrNoneExactAttributes(parameters, connectorLeftKey,
connectorRightKey)
+ && hasBothOrNoneExactAttributes(parameters, sinkLeftKey, sinkRightKey)
+ && hasBothOrNoneExactAttributes(
+ parameters,
+ PipeParameters.KeyReducer.reduce(connectorLeftKey),
+ PipeParameters.KeyReducer.reduce(connectorRightKey));
+ }
+
+ private static boolean hasExactAttributePair(
+ final PipeParameters parameters, final String leftKey, final String
rightKey) {
+ return hasExactAttribute(parameters, leftKey) &&
hasExactAttribute(parameters, rightKey);
+ }
+
+ private static boolean hasBothOrNoneExactAttributes(
+ final PipeParameters parameters, final String leftKey, final String
rightKey) {
+ return hasExactAttribute(parameters, leftKey) ==
hasExactAttribute(parameters, rightKey);
+ }
+
+ private static boolean hasExactAttribute(final PipeParameters parameters,
final String key) {
+ return parameters.getAttribute().containsKey(key);
}
private Set<TEndPoint> parseNodeUrls(final PipeParameters parameters) {
@@ -208,9 +292,21 @@ public class IoTDBLegacyPipeSink implements PipeConnector {
pipeName = configuration.getRuntimeEnvironment().getPipeName();
- useSSL = parameters.getBooleanOrDefault(SINK_IOTDB_SSL_ENABLE_KEY, false);
- trustStore = parameters.getString(SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY);
- trustStorePwd = parameters.getString(SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY);
+ useSSL =
+ parameters.getBooleanOrDefault(
+ Arrays.asList(CONNECTOR_IOTDB_SSL_ENABLE_KEY,
SINK_IOTDB_SSL_ENABLE_KEY), false);
+ trustStore =
+ parameters.getStringByKeys(
+ CONNECTOR_IOTDB_SSL_TRUST_STORE_PATH_KEY,
SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY);
+ trustStorePwd =
+ parameters.getStringByKeys(
+ CONNECTOR_IOTDB_SSL_TRUST_STORE_PWD_KEY,
SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY);
+ keyStore =
+ parameters.getStringByKeys(
+ CONNECTOR_IOTDB_SSL_KEY_STORE_PATH_KEY,
SINK_IOTDB_SSL_KEY_STORE_PATH_KEY);
+ keyStorePwd =
+ parameters.getStringByKeys(
+ CONNECTOR_IOTDB_SSL_KEY_STORE_PWD_KEY,
SINK_IOTDB_SSL_KEY_STORE_PWD_KEY);
final DataRegion dataRegion =
StorageEngine.getInstance()
@@ -235,7 +331,9 @@ public class IoTDBLegacyPipeSink implements PipeConnector {
port,
useSSL,
trustStore,
- trustStorePwd);
+ trustStorePwd,
+ keyStore,
+ keyStorePwd);
openClientSession();
final TSyncIdentityInfo identityInfo =
new TSyncIdentityInfo(
@@ -254,7 +352,7 @@ public class IoTDBLegacyPipeSink implements PipeConnector {
String.format(PipeConnectionException.CONNECTION_ERROR_FORMATTER,
ipAddress, port), e);
}
- sessionPool =
+ final SessionPool.Builder builder =
new SessionPool.Builder()
.host(ipAddress)
.port(port)
@@ -263,8 +361,11 @@ public class IoTDBLegacyPipeSink implements PipeConnector {
.maxSize(1)
.useSSL(useSSL)
.trustStore(trustStore)
- .trustStorePwd(trustStorePwd)
- .build();
+ .trustStorePwd(trustStorePwd);
+ if (keyStore != null) {
+ builder.keyStore(keyStore).keyStorePwd(keyStorePwd);
+ }
+ sessionPool = builder.build();
}
private void openClientSession() throws TException {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
index 213e43ab0b8..90e602af511 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
@@ -93,10 +93,17 @@ import java.util.concurrent.atomic.AtomicLong;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_ENABLE_SEND_TSFILE_LIMIT;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_ENABLE_SEND_TSFILE_LIMIT_DEFAULT_VALUE;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_SSL_ENABLE_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_SSL_KEY_STORE_PATH_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_SSL_KEY_STORE_PWD_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_SSL_TRUST_STORE_PATH_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_SSL_TRUST_STORE_PWD_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_LEADER_CACHE_ENABLE_DEFAULT_VALUE;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_LEADER_CACHE_ENABLE_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_ENABLE_SEND_TSFILE_LIMIT;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_SSL_ENABLE_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_SSL_KEY_STORE_PATH_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_SSL_KEY_STORE_PWD_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_LEADER_CACHE_ENABLE_KEY;
@@ -151,11 +158,23 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
final PipeParameters parameters = validator.getParameters();
validator.validate(
- args -> !((boolean) args[0] || (boolean) args[1] || (boolean) args[2]),
+ args ->
+ !((boolean) args[0]
+ || (boolean) args[1]
+ || (boolean) args[2]
+ || (boolean) args[3]
+ || (boolean) args[4]),
"Only 'iotdb-thrift-ssl-sink' supports SSL transmission currently.",
- 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));
+ parameters.getBooleanOrDefault(
+ Arrays.asList(CONNECTOR_IOTDB_SSL_ENABLE_KEY,
SINK_IOTDB_SSL_ENABLE_KEY), false),
+ parameters.hasAnyAttributes(
+ CONNECTOR_IOTDB_SSL_TRUST_STORE_PATH_KEY,
SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY),
+ parameters.hasAnyAttributes(
+ CONNECTOR_IOTDB_SSL_TRUST_STORE_PWD_KEY,
SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY),
+ parameters.hasAnyAttributes(
+ CONNECTOR_IOTDB_SSL_KEY_STORE_PATH_KEY,
SINK_IOTDB_SSL_KEY_STORE_PATH_KEY),
+ parameters.hasAnyAttributes(
+ CONNECTOR_IOTDB_SSL_KEY_STORE_PWD_KEY,
SINK_IOTDB_SSL_KEY_STORE_PWD_KEY));
}
@Override
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataNodeSyncSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataNodeSyncSink.java
index 72486d6c4e7..ff32bbbb06f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataNodeSyncSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataNodeSyncSink.java
@@ -43,6 +43,8 @@ public abstract class IoTDBDataNodeSyncSink extends
IoTDBSslSyncSink {
final boolean useSSL,
final String trustStorePath,
final String trustStorePwd,
+ final String keyStorePath,
+ final String keyStorePwd,
/* The following parameters are used locally. */
final boolean useLeaderCache,
final String loadBalanceStrategy,
@@ -60,6 +62,8 @@ public abstract class IoTDBDataNodeSyncSink extends
IoTDBSslSyncSink {
useSSL,
Objects.nonNull(trustStorePath) ?
IoTDBConfig.addDataHomeDir(trustStorePath) : null,
trustStorePwd,
+ Objects.nonNull(keyStorePath) ?
IoTDBConfig.addDataHomeDir(keyStorePath) : null,
+ keyStorePwd,
useLeaderCache,
loadBalanceStrategy,
userEntity,
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java
index fa18695ef1f..34bfc83763b 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java
@@ -114,6 +114,199 @@ public class PipeSinkTest {
}
}
+ @Test
+ public void testIoTDBThriftSyncSslSinkAcceptsMutualSslParameters() {
+ try (final IoTDBDataRegionSyncSink connector = new
IoTDBDataRegionSyncSink()) {
+ connector.validate(
+ new PipeParameterValidator(
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(
+ PipeSinkConstant.SINK_KEY,
+
BuiltinPipePlugin.IOTDB_THRIFT_SSL_SINK.getPipePluginName());
+ put(PipeSinkConstant.SINK_IOTDB_IP_KEY, "127.0.0.1");
+ put(PipeSinkConstant.SINK_IOTDB_PORT_KEY, "6668");
+
put(PipeSinkConstant.SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY, "truststore");
+ put(PipeSinkConstant.SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY,
"trustpwd");
+ put(PipeSinkConstant.SINK_IOTDB_SSL_KEY_STORE_PATH_KEY,
"keystore");
+ put(PipeSinkConstant.SINK_IOTDB_SSL_KEY_STORE_PWD_KEY,
"keypwd");
+ }
+ })));
+ } catch (Exception e) {
+ Assert.fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void
testIoTDBThriftSyncSslConnectorAcceptsConnectorMutualSslAliases() {
+ try (final IoTDBDataRegionSyncSink connector = new
IoTDBDataRegionSyncSink()) {
+ connector.validate(
+ new PipeParameterValidator(
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(
+ PipeSinkConstant.CONNECTOR_KEY,
+
BuiltinPipePlugin.IOTDB_THRIFT_SSL_CONNECTOR.getPipePluginName());
+ put(PipeSinkConstant.CONNECTOR_IOTDB_IP_KEY,
"127.0.0.1");
+ put(PipeSinkConstant.CONNECTOR_IOTDB_PORT_KEY, "6668");
+
put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_TRUST_STORE_PATH_KEY, "truststore");
+
put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_TRUST_STORE_PWD_KEY, "trustpwd");
+
put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_KEY_STORE_PATH_KEY, "keystore");
+
put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_KEY_STORE_PWD_KEY, "keypwd");
+ }
+ })));
+ } catch (Exception e) {
+ Assert.fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testIoTDBThriftSyncSslSinkRejectsIncompleteKeyStoreParameters() {
+ try (final IoTDBDataRegionSyncSink connector = new
IoTDBDataRegionSyncSink()) {
+ connector.validate(
+ new PipeParameterValidator(
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(
+ PipeSinkConstant.SINK_KEY,
+
BuiltinPipePlugin.IOTDB_THRIFT_SSL_SINK.getPipePluginName());
+ put(PipeSinkConstant.SINK_IOTDB_IP_KEY, "127.0.0.1");
+ put(PipeSinkConstant.SINK_IOTDB_PORT_KEY, "6668");
+
put(PipeSinkConstant.SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY, "truststore");
+ put(PipeSinkConstant.SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY,
"trustpwd");
+ put(PipeSinkConstant.SINK_IOTDB_SSL_KEY_STORE_PATH_KEY,
"keystore");
+ }
+ })));
+ Assert.fail();
+ } catch (Exception e) {
+
Assert.assertTrue(e.getMessage().contains(PipeSinkConstant.SINK_IOTDB_SSL_KEY_STORE_PWD_KEY));
+ }
+ }
+
+ @Test
+ public void
testIoTDBThriftSyncSslSinkRejectsCrossAliasTrustStoreParameters() {
+ try (final IoTDBDataRegionSyncSink connector = new
IoTDBDataRegionSyncSink()) {
+ connector.validate(
+ new PipeParameterValidator(
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(
+ PipeSinkConstant.CONNECTOR_KEY,
+
BuiltinPipePlugin.IOTDB_THRIFT_SSL_CONNECTOR.getPipePluginName());
+ put(PipeSinkConstant.CONNECTOR_IOTDB_IP_KEY,
"127.0.0.1");
+ put(PipeSinkConstant.CONNECTOR_IOTDB_PORT_KEY, "6668");
+
put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_TRUST_STORE_PATH_KEY, "truststore");
+ put(PipeSinkConstant.SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY,
"trustpwd");
+ }
+ })));
+ Assert.fail();
+ } catch (Exception e) {
+ Assert.assertTrue(
+
e.getMessage().contains(PipeSinkConstant.SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY));
+ }
+ }
+
+ @Test
+ public void testIoTDBThriftSyncSslSinkRejectsCrossAliasKeyStoreParameters() {
+ try (final IoTDBDataRegionSyncSink connector = new
IoTDBDataRegionSyncSink()) {
+ connector.validate(
+ new PipeParameterValidator(
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(
+ PipeSinkConstant.SINK_KEY,
+
BuiltinPipePlugin.IOTDB_THRIFT_SSL_SINK.getPipePluginName());
+ put(PipeSinkConstant.SINK_IOTDB_IP_KEY, "127.0.0.1");
+ put(PipeSinkConstant.SINK_IOTDB_PORT_KEY, "6668");
+
put(PipeSinkConstant.SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY, "truststore");
+ put(PipeSinkConstant.SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY,
"trustpwd");
+
put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_KEY_STORE_PATH_KEY, "keystore");
+ put(PipeSinkConstant.SINK_IOTDB_SSL_KEY_STORE_PWD_KEY,
"keypwd");
+ }
+ })));
+ Assert.fail();
+ } catch (Exception e) {
+
Assert.assertTrue(e.getMessage().contains(PipeSinkConstant.SINK_IOTDB_SSL_KEY_STORE_PWD_KEY));
+ }
+ }
+
+ @Test
+ public void testIoTDBThriftAsyncSinkRejectsSslKeyStoreParameters() {
+ try (final IoTDBDataRegionAsyncSink connector = new
IoTDBDataRegionAsyncSink()) {
+ connector.validate(
+ new PipeParameterValidator(
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(
+ PipeSinkConstant.CONNECTOR_KEY,
+
BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName());
+ put(PipeSinkConstant.CONNECTOR_IOTDB_NODE_URLS_KEY,
"127.0.0.1:6668");
+
put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_KEY_STORE_PATH_KEY, "keystore");
+
put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_KEY_STORE_PWD_KEY, "keypwd");
+ }
+ })));
+ Assert.fail();
+ } catch (Exception e) {
+ Assert.assertTrue(e.getMessage().contains("Only 'iotdb-thrift-ssl-sink'
supports SSL"));
+ }
+ }
+
+ @Test
+ public void testIoTDBLegacyPipeSinkAcceptsConnectorMutualSslAliases() {
+ try (final IoTDBLegacyPipeSink connector = new IoTDBLegacyPipeSink()) {
+ connector.validate(
+ new PipeParameterValidator(
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(
+ PipeSinkConstant.CONNECTOR_KEY,
+
BuiltinPipePlugin.IOTDB_LEGACY_PIPE_CONNECTOR.getPipePluginName());
+ put(PipeSinkConstant.CONNECTOR_IOTDB_IP_KEY,
"127.0.0.1");
+ put(PipeSinkConstant.CONNECTOR_IOTDB_PORT_KEY, "6668");
+ put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_ENABLE_KEY,
Boolean.TRUE.toString());
+
put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_TRUST_STORE_PATH_KEY, "truststore");
+
put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_TRUST_STORE_PWD_KEY, "trustpwd");
+
put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_KEY_STORE_PATH_KEY, "keystore");
+
put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_KEY_STORE_PWD_KEY, "keypwd");
+ }
+ })));
+ } catch (Exception e) {
+ Assert.fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testIoTDBLegacyPipeSinkRejectsCrossAliasSslParameters() {
+ try (final IoTDBLegacyPipeSink connector = new IoTDBLegacyPipeSink()) {
+ connector.validate(
+ new PipeParameterValidator(
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(
+ PipeSinkConstant.CONNECTOR_KEY,
+
BuiltinPipePlugin.IOTDB_LEGACY_PIPE_CONNECTOR.getPipePluginName());
+ put(PipeSinkConstant.CONNECTOR_IOTDB_IP_KEY,
"127.0.0.1");
+ put(PipeSinkConstant.CONNECTOR_IOTDB_PORT_KEY, "6668");
+ put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_ENABLE_KEY,
Boolean.TRUE.toString());
+
put(PipeSinkConstant.CONNECTOR_IOTDB_SSL_TRUST_STORE_PATH_KEY, "truststore");
+ put(PipeSinkConstant.SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY,
"trustpwd");
+ }
+ })));
+ Assert.fail();
+ } catch (Exception e) {
+ Assert.assertTrue(
+
e.getMessage().contains(PipeSinkConstant.SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY));
+ }
+ }
+
@Test
public void testAsyncSinkDropDoesNotRequeueDroppedPipeEvents() throws
Exception {
try (final IoTDBDataRegionAsyncSink connector = new
IoTDBDataRegionAsyncSink()) {
diff --git
a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java
b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java
index eb38e3524a2..696bb4485a9 100644
---
a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java
+++
b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java
@@ -484,6 +484,12 @@ public final class PipeMessages {
// ===================== IoTDBSslSyncSink =====================
+ public static final String SSL_TRUST_STORE_PAIR_REQUIRED_WHEN_SSL_ENABLED =
+ "When ssl transport is enabled, specify a complete trust-store pair
under the same "
+ + "alias: %s and %s, %s and %s, or %s and %s";
+ public static final String
SSL_KEY_STORE_PATH_AND_PASSWORD_MUST_BE_SPECIFIED_TOGETHER =
+ "SSL key-store path and password must be specified together under the
same "
+ + "alias: %s and %s, %s and %s, or %s and %s";
public static final String SYNC_CLIENT_MANAGER_CLOSED =
"IoTDB sync client manager has been closed";
public static final String REDIRECT_FILE_POSITION = "Redirect file position
to {}.";
diff --git
a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java
b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java
index 6914f4b1379..5126852bb6b 100644
---
a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java
+++
b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java
@@ -466,6 +466,10 @@ public final class PipeMessages {
// ===================== IoTDBSslSyncSink =====================
+ public static final String SSL_TRUST_STORE_PAIR_REQUIRED_WHEN_SSL_ENABLED =
+ "启用 SSL 传输时，请在同一别名下指定完整的 trust-store 参数对：%s 和 %s、%s 和 %s，或 %s 和 %s";
+ public static final String
SSL_KEY_STORE_PATH_AND_PASSWORD_MUST_BE_SPECIFIED_TOGETHER =
+ "SSL key-store 路径和密码必须在同一别名下同时指定：%s 和 %s、%s 和 %s，或 %s 和 %s";
public static final String SYNC_CLIENT_MANAGER_CLOSED =
"IoTDB sync 客户端管理器已关闭";
public static final String REDIRECT_FILE_POSITION = "重定向文件位置到 {}。";
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java
index 39be064f09a..aa09b712d6e 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java
@@ -50,9 +50,19 @@ public class PipeSinkConstant {
public static final String CONNECTOR_IOTDB_NODE_URLS_KEY =
"connector.node-urls";
public static final String SINK_IOTDB_NODE_URLS_KEY = "sink.node-urls";
+ public static final String CONNECTOR_IOTDB_SSL_ENABLE_KEY =
"connector.ssl.enable";
public static final String SINK_IOTDB_SSL_ENABLE_KEY = "sink.ssl.enable";
+ public static final String CONNECTOR_IOTDB_SSL_TRUST_STORE_PATH_KEY =
+ "connector.ssl.trust-store-path";
public static final String SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY =
"sink.ssl.trust-store-path";
+ public static final String CONNECTOR_IOTDB_SSL_TRUST_STORE_PWD_KEY =
+ "connector.ssl.trust-store-pwd";
public static final String SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY =
"sink.ssl.trust-store-pwd";
+ public static final String CONNECTOR_IOTDB_SSL_KEY_STORE_PATH_KEY =
+ "connector.ssl.key-store-path";
+ public static final String SINK_IOTDB_SSL_KEY_STORE_PATH_KEY =
"sink.ssl.key-store-path";
+ public static final String CONNECTOR_IOTDB_SSL_KEY_STORE_PWD_KEY =
"connector.ssl.key-store-pwd";
+ public static final String SINK_IOTDB_SSL_KEY_STORE_PWD_KEY =
"sink.ssl.key-store-pwd";
public static final String CONNECTOR_IOTDB_PARALLEL_TASKS_KEY =
"connector.parallel.tasks";
public static final String SINK_IOTDB_PARALLEL_TASKS_KEY =
"sink.parallel.tasks";
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClient.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClient.java
index b0407e0a7bf..296848629fd 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClient.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClient.java
@@ -55,6 +55,19 @@ public class IoTDBSyncClient extends IClientRPCService.Client
String trustStore,
String trustStorePwd)
throws TTransportException {
+ this(property, ipAddress, port, useSSL, trustStore, trustStorePwd, null,
null);
+ }
+
+ public IoTDBSyncClient(
+ ThriftClientProperty property,
+ String ipAddress,
+ int port,
+ boolean useSSL,
+ String trustStore,
+ String trustStorePwd,
+ String keyStore,
+ String keyStorePwd)
+ throws TTransportException {
super(
property
.getProtocolFactory()
@@ -65,7 +78,9 @@ public class IoTDBSyncClient extends IClientRPCService.Client
port,
property.getConnectionTimeoutMs(),
trustStore,
- trustStorePwd)
+ trustStorePwd,
+ keyStore,
+ keyStorePwd)
: DeepCopyRpcTransportFactory.INSTANCE.getTransport(
ipAddress, port, property.getConnectionTimeoutMs())));
this.ipAddress = ipAddress;
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClientManager.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClientManager.java
index c6ad14aac4d..83b85a6d583 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClientManager.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClientManager.java
@@ -57,6 +57,8 @@ public abstract class IoTDBSyncClientManager extends
IoTDBClientManager implemen
private final boolean useSSL;
private final String trustStorePath;
private final String trustStorePwd;
+ private final String keyStorePath;
+ private final String keyStorePwd;
protected final Map<TEndPoint, Pair<IoTDBSyncClient, Boolean>>
endPoint2ClientAndStatus =
new ConcurrentHashMap<>();
@@ -71,6 +73,8 @@ public abstract class IoTDBSyncClientManager extends
IoTDBClientManager implemen
boolean useSSL,
String trustStorePath,
String trustStorePwd,
+ String keyStorePath,
+ String keyStorePwd,
/* The following parameters are used locally. */
boolean useLeaderCache,
String loadBalanceStrategy,
@@ -96,6 +100,8 @@ public abstract class IoTDBSyncClientManager extends
IoTDBClientManager implemen
this.useSSL = useSSL;
this.trustStorePath = trustStorePath;
this.trustStorePwd = trustStorePwd;
+ this.keyStorePath = keyStorePath;
+ this.keyStorePwd = keyStorePwd;
for (final TEndPoint endPoint : endPoints) {
endPoint2ClientAndStatus.put(endPoint, new Pair<>(null, false));
@@ -205,7 +211,9 @@ public abstract class IoTDBSyncClientManager extends
IoTDBClientManager implemen
endPoint.getPort(),
useSSL,
trustStorePath,
- trustStorePwd));
+ trustStorePwd,
+ keyStorePath,
+ keyStorePwd));
return true;
} catch (Exception e) {
endPoint2HandshakeErrorMessage.put(endPoint, e.getMessage());
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSslSyncSink.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSslSyncSink.java
index e2dccccfd20..a045082b24e 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSslSyncSink.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSslSyncSink.java
@@ -53,10 +53,17 @@ import java.util.Map;
import static
org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR;
import static
org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin.IOTDB_THRIFT_SSL_CONNECTOR;
import static
org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin.IOTDB_THRIFT_SSL_SINK;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_SSL_ENABLE_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_SSL_KEY_STORE_PATH_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_SSL_KEY_STORE_PWD_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_SSL_TRUST_STORE_PATH_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_SSL_TRUST_STORE_PWD_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_LEADER_CACHE_ENABLE_DEFAULT_VALUE;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_LEADER_CACHE_ENABLE_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_SSL_ENABLE_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_SSL_KEY_STORE_PATH_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_SSL_KEY_STORE_PWD_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_KEY;
@@ -90,16 +97,93 @@ public abstract class IoTDBSslSyncSink extends IoTDBSink {
IOTDB_THRIFT_CONNECTOR.getPipePluginName())
.toLowerCase();
- validator.validate(
- args -> !((boolean) args[0]) || ((boolean) args[1] && (boolean)
args[2]),
- String.format(
- "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_THRIFT_SSL_CONNECTOR.getPipePluginName().equals(userSpecifiedConnectorName)
- ||
IOTDB_THRIFT_SSL_SINK.getPipePluginName().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));
+ validator
+ .validate(
+ args -> !((boolean) args[0]) || ((boolean) args[1] && (boolean)
args[2]),
+ String.format(
+ PipeMessages.SSL_TRUST_STORE_PAIR_REQUIRED_WHEN_SSL_ENABLED,
+ CONNECTOR_IOTDB_SSL_TRUST_STORE_PATH_KEY,
+ CONNECTOR_IOTDB_SSL_TRUST_STORE_PWD_KEY,
+ SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY,
+ SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY,
+
PipeParameters.KeyReducer.reduce(CONNECTOR_IOTDB_SSL_TRUST_STORE_PATH_KEY),
+
PipeParameters.KeyReducer.reduce(CONNECTOR_IOTDB_SSL_TRUST_STORE_PWD_KEY)),
+
IOTDB_THRIFT_SSL_CONNECTOR.getPipePluginName().equals(userSpecifiedConnectorName)
+ ||
IOTDB_THRIFT_SSL_SINK.getPipePluginName().equals(userSpecifiedConnectorName)
+ || parameters.getBooleanOrDefault(
+ Arrays.asList(CONNECTOR_IOTDB_SSL_ENABLE_KEY,
SINK_IOTDB_SSL_ENABLE_KEY),
+ false),
+ hasCompleteAttributePair(
+ parameters,
+ CONNECTOR_IOTDB_SSL_TRUST_STORE_PATH_KEY,
+ CONNECTOR_IOTDB_SSL_TRUST_STORE_PWD_KEY,
+ SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY,
+ SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY),
+ hasNoHalfAttributePair(
+ parameters,
+ CONNECTOR_IOTDB_SSL_TRUST_STORE_PATH_KEY,
+ CONNECTOR_IOTDB_SSL_TRUST_STORE_PWD_KEY,
+ SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY,
+ SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY))
+ .validate(
+ args -> (boolean) args[0] == (boolean) args[1],
+ String.format(
+
PipeMessages.SSL_KEY_STORE_PATH_AND_PASSWORD_MUST_BE_SPECIFIED_TOGETHER,
+ CONNECTOR_IOTDB_SSL_KEY_STORE_PATH_KEY,
+ CONNECTOR_IOTDB_SSL_KEY_STORE_PWD_KEY,
+ SINK_IOTDB_SSL_KEY_STORE_PATH_KEY,
+ SINK_IOTDB_SSL_KEY_STORE_PWD_KEY,
+
PipeParameters.KeyReducer.reduce(CONNECTOR_IOTDB_SSL_KEY_STORE_PATH_KEY),
+
PipeParameters.KeyReducer.reduce(CONNECTOR_IOTDB_SSL_KEY_STORE_PWD_KEY)),
+ true,
+ hasNoHalfAttributePair(
+ parameters,
+ CONNECTOR_IOTDB_SSL_KEY_STORE_PATH_KEY,
+ CONNECTOR_IOTDB_SSL_KEY_STORE_PWD_KEY,
+ SINK_IOTDB_SSL_KEY_STORE_PATH_KEY,
+ SINK_IOTDB_SSL_KEY_STORE_PWD_KEY));
+ }
+
+ private static boolean hasCompleteAttributePair(
+ final PipeParameters parameters,
+ final String connectorLeftKey,
+ final String connectorRightKey,
+ final String sinkLeftKey,
+ final String sinkRightKey) {
+ return hasExactAttributePair(parameters, connectorLeftKey,
connectorRightKey)
+ || hasExactAttributePair(parameters, sinkLeftKey, sinkRightKey)
+ || hasExactAttributePair(
+ parameters,
+ PipeParameters.KeyReducer.reduce(connectorLeftKey),
+ PipeParameters.KeyReducer.reduce(connectorRightKey));
+ }
+
+ private static boolean hasNoHalfAttributePair(
+ final PipeParameters parameters,
+ final String connectorLeftKey,
+ final String connectorRightKey,
+ final String sinkLeftKey,
+ final String sinkRightKey) {
+ return hasBothOrNoneExactAttributes(parameters, connectorLeftKey,
connectorRightKey)
+ && hasBothOrNoneExactAttributes(parameters, sinkLeftKey, sinkRightKey)
+ && hasBothOrNoneExactAttributes(
+ parameters,
+ PipeParameters.KeyReducer.reduce(connectorLeftKey),
+ PipeParameters.KeyReducer.reduce(connectorRightKey));
+ }
+
+ private static boolean hasExactAttributePair(
+ final PipeParameters parameters, final String leftKey, final String
rightKey) {
+ return hasExactAttribute(parameters, leftKey) &&
hasExactAttribute(parameters, rightKey);
+ }
+
+ private static boolean hasBothOrNoneExactAttributes(
+ final PipeParameters parameters, final String leftKey, final String
rightKey) {
+ return hasExactAttribute(parameters, leftKey) ==
hasExactAttribute(parameters, rightKey);
+ }
+
+ private static boolean hasExactAttribute(final PipeParameters parameters,
final String key) {
+ return parameters.getAttribute().containsKey(key);
}
@Override
@@ -118,9 +202,20 @@ public abstract class IoTDBSslSyncSink extends IoTDBSink {
final boolean useSSL =
IOTDB_THRIFT_SSL_CONNECTOR.getPipePluginName().equals(userSpecifiedConnectorName)
||
IOTDB_THRIFT_SSL_SINK.getPipePluginName().equals(userSpecifiedConnectorName)
- || parameters.getBooleanOrDefault(SINK_IOTDB_SSL_ENABLE_KEY,
false);
- final String trustStorePath =
parameters.getString(SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY);
- final String trustStorePwd =
parameters.getString(SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY);
+ || parameters.getBooleanOrDefault(
+ Arrays.asList(CONNECTOR_IOTDB_SSL_ENABLE_KEY,
SINK_IOTDB_SSL_ENABLE_KEY), false);
+ final String trustStorePath =
+ parameters.getStringByKeys(
+ CONNECTOR_IOTDB_SSL_TRUST_STORE_PATH_KEY,
SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY);
+ final String trustStorePwd =
+ parameters.getStringByKeys(
+ CONNECTOR_IOTDB_SSL_TRUST_STORE_PWD_KEY,
SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY);
+ final String keyStorePath =
+ parameters.getStringByKeys(
+ CONNECTOR_IOTDB_SSL_KEY_STORE_PATH_KEY,
SINK_IOTDB_SSL_KEY_STORE_PATH_KEY);
+ final String keyStorePwd =
+ parameters.getStringByKeys(
+ CONNECTOR_IOTDB_SSL_KEY_STORE_PWD_KEY,
SINK_IOTDB_SSL_KEY_STORE_PWD_KEY);
// leader cache configuration
final boolean useLeaderCache =
@@ -134,6 +229,8 @@ public abstract class IoTDBSslSyncSink extends IoTDBSink {
useSSL,
trustStorePath,
trustStorePwd,
+ keyStorePath,
+ keyStorePwd,
useLeaderCache,
loadBalanceStrategy,
userEntity,
@@ -150,6 +247,8 @@ public abstract class IoTDBSslSyncSink extends IoTDBSink {
final boolean useSSL,
final String trustStorePath,
final String trustStorePwd,
+ final String keyStorePath,
+ final String keyStorePwd,
/* The following parameters are used locally. */
final boolean useLeaderCache,
final String loadBalanceStrategy,
