This is an automated email from the ASF dual-hosted git repository. haonan pushed a commit to branch ssl_between_nodes in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 314467e90acbae97f5b5ebe85dc93d6e80af6073 Author: HTHou <[email protected]> AuthorDate: Thu Jun 19 14:28:30 2025 +0800 Fix cndn heart beat --- .../apache/iotdb/rpc/BaseRpcTransportFactory.java | 17 ++++++ .../sync/SyncDataNodeHeartbeatClientPool.java | 70 ++++++++++++++++++++++ .../manager/load/service/HeartbeatService.java | 14 ++++- .../manager/load/service/TopologyService.java | 9 ++- .../service/thrift/ConfigNodeRPCService.java | 1 + .../iot/service/IoTConsensusRPCService.java | 39 ++++++++---- .../pipe/service/PipeConsensusRPCService.java | 39 ++++++++---- .../iotdb/db/protocol/client/ConfigNodeClient.java | 19 +++++- .../db/service/DataNodeInternalRPCService.java | 4 +- .../iotdb/commons/client/ClientPoolFactory.java | 24 ++++++++ .../client/sync/SyncConfigNodeIServiceClient.java | 20 +++++-- .../sync/SyncDataNodeInternalServiceClient.java | 24 ++++++-- .../SyncDataNodeMPPDataExchangeServiceClient.java | 20 +++++-- .../sync/SyncPipeConsensusServiceClient.java | 20 +++++-- .../service/AbstractThriftServiceThread.java | 4 +- 15 files changed, 274 insertions(+), 50 deletions(-) diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/BaseRpcTransportFactory.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/BaseRpcTransportFactory.java index 61fcf52671f..1175b16a935 100644 --- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/BaseRpcTransportFactory.java +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/BaseRpcTransportFactory.java @@ -84,6 +84,23 @@ public class BaseRpcTransportFactory extends TTransportFactory { return inner.getTransport(transport); } + public TTransport getTransport( + String ip, + int port, + int timeout, + String trustStore, + String trustStorePwd, + String keyStore, + String keyStorePwd) + throws TTransportException { + TSSLTransportFactory.TSSLTransportParameters params = + new TSSLTransportFactory.TSSLTransportParameters(); + params.setTrustStore(trustStore, trustStorePwd); + params.setKeyStore(keyStore, keyStorePwd); + TTransport transport = TSSLTransportFactory.getClientSocket(ip, port, timeout, params); + return inner.getTransport(transport); + } + public TTransport getTransport(String ip, int port, int timeout) throws TTransportException { return inner.getTransport( new TSocket(TConfigurationConst.defaultTConfiguration, ip, port, timeout)); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeHeartbeatClientPool.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeHeartbeatClientPool.java new file mode 100644 index 00000000000..b32f023dac2 --- /dev/null +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeHeartbeatClientPool.java @@ -0,0 +1,70 @@ +/* + * 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.confignode.client.sync; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.client.ClientPoolFactory; +import org.apache.iotdb.commons.client.IClientManager; +import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient; +import org.apache.iotdb.confignode.client.async.handlers.heartbeat.DataNodeHeartbeatHandler; +import org.apache.iotdb.mpp.rpc.thrift.TDataNodeHeartbeatReq; +import org.apache.iotdb.mpp.rpc.thrift.TDataNodeHeartbeatResp; + +/** Synchronously send RPC requests to DataNodes. See queryengine.thrift for more details. */ +public class SyncDataNodeHeartbeatClientPool { + + private final IClientManager<TEndPoint, SyncDataNodeInternalServiceClient> clientManager; + + private SyncDataNodeHeartbeatClientPool() { + clientManager = + new IClientManager.Factory<TEndPoint, SyncDataNodeInternalServiceClient>() + .createClientManager( + new ClientPoolFactory.SyncDataNodeHeartbeatServiceClientPoolFactory()); + } + + /** + * Only used in LoadManager. + * + * @param endPoint The specific DataNode + */ + public void getDataNodeHeartBeat( + TEndPoint endPoint, TDataNodeHeartbeatReq req, DataNodeHeartbeatHandler handler) { + try (SyncDataNodeInternalServiceClient client = clientManager.borrowClient(endPoint)) { + TDataNodeHeartbeatResp resp = client.getDataNodeHeartBeat(req); + handler.onComplete(resp); + } catch (Exception e) { + handler.onError(e); + } + } + + private static class SyncDataNodeHeartbeatClientPoolHolder { + + private static final SyncDataNodeHeartbeatClientPool INSTANCE = + new SyncDataNodeHeartbeatClientPool(); + + private SyncDataNodeHeartbeatClientPoolHolder() { + // Empty constructor + } + } + + public static SyncDataNodeHeartbeatClientPool getInstance() { + return SyncDataNodeHeartbeatClientPoolHolder.INSTANCE; + } +} diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java index 9e630733025..40418a00efc 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java @@ -27,6 +27,7 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.commons.concurrent.ThreadName; import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil; +import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.confignode.client.async.AsyncAINodeHeartbeatClientPool; import org.apache.iotdb.confignode.client.async.AsyncConfigNodeHeartbeatClientPool; @@ -34,6 +35,7 @@ import org.apache.iotdb.confignode.client.async.AsyncDataNodeHeartbeatClientPool import org.apache.iotdb.confignode.client.async.handlers.heartbeat.AINodeHeartbeatHandler; import org.apache.iotdb.confignode.client.async.handlers.heartbeat.ConfigNodeHeartbeatHandler; import org.apache.iotdb.confignode.client.async.handlers.heartbeat.DataNodeHeartbeatHandler; +import org.apache.iotdb.confignode.client.sync.SyncDataNodeHeartbeatClientPool; import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; import org.apache.iotdb.confignode.manager.IManager; import org.apache.iotdb.confignode.manager.consensus.ConsensusManager; @@ -276,9 +278,15 @@ public class HeartbeatService { configManager.getPipeManager().getPipeRuntimeCoordinator()); configManager.getClusterQuotaManager().updateSpaceQuotaUsage(); addConfigNodeLocationsToReq(dataNodeId, heartbeatReq); - AsyncDataNodeHeartbeatClientPool.getInstance() - .getDataNodeHeartBeat( - dataNodeInfo.getLocation().getInternalEndPoint(), heartbeatReq, handler); + if (CommonDescriptor.getInstance().getConfig().isEnableSSL()) { + SyncDataNodeHeartbeatClientPool.getInstance() + .getDataNodeHeartBeat( + dataNodeInfo.getLocation().getInternalEndPoint(), heartbeatReq, handler); + } else { + AsyncDataNodeHeartbeatClientPool.getInstance() + .getDataNodeHeartBeat( + dataNodeInfo.getLocation().getInternalEndPoint(), heartbeatReq, handler); + } } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/TopologyService.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/TopologyService.java index 3a4ff73ecae..e05f00415bd 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/TopologyService.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/TopologyService.java @@ -27,6 +27,7 @@ import org.apache.iotdb.common.rpc.thrift.TTestConnectionResult; import org.apache.iotdb.commons.cluster.NodeStatus; import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.commons.concurrent.ThreadName; +import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType; import org.apache.iotdb.confignode.client.async.CnToDnInternalServiceAsyncRequestManager; import org.apache.iotdb.confignode.client.async.handlers.DataNodeAsyncRequestContext; @@ -181,8 +182,12 @@ public class TopologyService implements Runnable, IClusterStatusSubscriber { CnToDnAsyncRequestType.SUBMIT_TEST_DN_INTERNAL_CONNECTION_TASK, nodeLocations, dataNodeLocationMap); - CnToDnInternalServiceAsyncRequestManager.getInstance() - .sendAsyncRequestWithTimeoutInMs(dataNodeAsyncRequestContext, PROBING_TIMEOUT_MS); + if (CommonDescriptor.getInstance().getConfig().isEnableSSL()) { + // TODO: Haonan do it syncly + } else { + CnToDnInternalServiceAsyncRequestManager.getInstance() + .sendAsyncRequestWithTimeoutInMs(dataNodeAsyncRequestContext, PROBING_TIMEOUT_MS); + } final List<TTestConnectionResult> results = new ArrayList<>(); dataNodeAsyncRequestContext .getResponseMap() diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCService.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCService.java index 33710cb7594..c828fad4c08 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCService.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCService.java @@ -90,6 +90,7 @@ public class ConfigNodeRPCService extends ThriftService implements ConfigNodeRPC commonConfig.isRpcThriftCompressionEnabled(), DeepCopyRpcTransportFactory.INSTANCE); } catch (RPCServiceException e) { + e.printStackTrace(); throw new IllegalAccessException(e.getMessage()); } thriftServiceThread.setName(ThreadName.CONFIGNODE_RPC_SERVICE.getName()); diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCService.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCService.java index 8f7c484af4b..2398e69a389 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCService.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCService.java @@ -74,17 +74,34 @@ public class IoTConsensusRPCService extends ThriftService implements IoTConsensu throws IllegalAccessException, InstantiationException, ClassNotFoundException { try { thriftServiceThread = - new ThriftServiceThread( - processor, - getID().getName(), - ThreadName.IOT_CONSENSUS_RPC_PROCESSOR.getName(), - getBindIP(), - getBindPort(), - config.getRpc().getRpcMaxConcurrentClientNum(), - config.getRpc().getThriftServerAwaitTimeForStopService(), - new IoTConsensusRPCServiceHandler(iotConsensusRPCServiceProcessor), - config.getRpc().isRpcThriftCompressionEnabled(), - ZeroCopyRpcTransportFactory.INSTANCE); + config.getRpc().isEnableSSL() + ? new ThriftServiceThread( + processor, + getID().getName(), + ThreadName.IOT_CONSENSUS_RPC_PROCESSOR.getName(), + getBindIP(), + getBindPort(), + config.getRpc().getRpcMaxConcurrentClientNum(), + config.getRpc().getThriftServerAwaitTimeForStopService(), + new IoTConsensusRPCServiceHandler(iotConsensusRPCServiceProcessor), + config.getRpc().isRpcThriftCompressionEnabled(), + config.getRpc().getSslKeyStorePath(), + config.getRpc().getSslKeyStorePassword(), + config.getRpc().getSslTrustStorePath(), + config.getRpc().getSslTrustStorePassword(), + config.getRpc().getConnectionTimeoutInMs(), + ZeroCopyRpcTransportFactory.INSTANCE) + : new ThriftServiceThread( + processor, + getID().getName(), + ThreadName.IOT_CONSENSUS_RPC_PROCESSOR.getName(), + getBindIP(), + getBindPort(), + config.getRpc().getRpcMaxConcurrentClientNum(), + config.getRpc().getThriftServerAwaitTimeForStopService(), + new IoTConsensusRPCServiceHandler(iotConsensusRPCServiceProcessor), + config.getRpc().isRpcThriftCompressionEnabled(), + ZeroCopyRpcTransportFactory.INSTANCE); } catch (RPCServiceException e) { throw new IllegalAccessException(e.getMessage()); } diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/service/PipeConsensusRPCService.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/service/PipeConsensusRPCService.java index 66bded8d13d..1f3240e5e50 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/service/PipeConsensusRPCService.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/service/PipeConsensusRPCService.java @@ -61,17 +61,34 @@ public class PipeConsensusRPCService extends ThriftService implements PipeConsen public void initThriftServiceThread() throws IllegalAccessException { try { thriftServiceThread = - new ThriftServiceThread( - processor, - getID().getName(), - ThreadName.PIPE_CONSENSUS_RPC_PROCESSOR.getName(), - getBindIP(), - getBindPort(), - config.getRpc().getRpcMaxConcurrentClientNum(), - config.getRpc().getThriftServerAwaitTimeForStopService(), - new PipeConsensusRPCServiceHandler(pipeConsensusRPCServiceProcessor), - config.getRpc().isRpcThriftCompressionEnabled(), - ZeroCopyRpcTransportFactory.INSTANCE); + config.getRpc().isEnableSSL() + ? new ThriftServiceThread( + processor, + getID().getName(), + ThreadName.PIPE_CONSENSUS_RPC_PROCESSOR.getName(), + getBindIP(), + getBindPort(), + config.getRpc().getRpcMaxConcurrentClientNum(), + config.getRpc().getThriftServerAwaitTimeForStopService(), + new PipeConsensusRPCServiceHandler(pipeConsensusRPCServiceProcessor), + config.getRpc().isRpcThriftCompressionEnabled(), + config.getRpc().getSslKeyStorePath(), + config.getRpc().getSslKeyStorePassword(), + config.getRpc().getSslTrustStorePath(), + config.getRpc().getSslTrustStorePassword(), + config.getRpc().getConnectionTimeoutInMs(), + ZeroCopyRpcTransportFactory.INSTANCE) + : new ThriftServiceThread( + processor, + getID().getName(), + ThreadName.PIPE_CONSENSUS_RPC_PROCESSOR.getName(), + getBindIP(), + getBindPort(), + config.getRpc().getRpcMaxConcurrentClientNum(), + config.getRpc().getThriftServerAwaitTimeForStopService(), + new PipeConsensusRPCServiceHandler(pipeConsensusRPCServiceProcessor), + config.getRpc().isRpcThriftCompressionEnabled(), + ZeroCopyRpcTransportFactory.INSTANCE); } catch (RPCServiceException e) { throw new IllegalAccessException(e.getMessage()); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java index 9d0dcd22f91..aa089da3ed4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java @@ -39,6 +39,8 @@ import org.apache.iotdb.commons.client.factory.ThriftClientFactory; import org.apache.iotdb.commons.client.property.ThriftClientProperty; import org.apache.iotdb.commons.client.request.TestConnectionUtils; import org.apache.iotdb.commons.client.sync.SyncThriftClientWithErrorHandler; +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.consensus.ConfigRegionId; import org.apache.iotdb.confignode.rpc.thrift.IConfigNodeRPCService; import org.apache.iotdb.confignode.rpc.thrift.TAINodeConfigurationResp; @@ -238,6 +240,8 @@ public class ConfigNodeClient implements IConfigNodeRPCService.Iface, ThriftClie private final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); + private final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig(); + ClientManager<ConfigRegionId, ConfigNodeClient> clientManager; ConfigRegionId configRegionId = ConfigNodeInfo.CONFIG_REGION_ID; @@ -260,9 +264,18 @@ public class ConfigNodeClient implements IConfigNodeRPCService.Iface, ThriftClie public void connect(TEndPoint endpoint, int timeoutMs) throws TException { try { transport = - DeepCopyRpcTransportFactory.INSTANCE.getTransport( - // As there is a try-catch already, we do not need to use TSocket.wrap - endpoint.getIp(), endpoint.getPort(), timeoutMs); + commonConfig.isEnableSSL() + ? DeepCopyRpcTransportFactory.INSTANCE.getTransport( + endpoint.getIp(), + endpoint.getPort(), + timeoutMs, + commonConfig.getTrustStorePath(), + commonConfig.getTrustStorePwd(), + commonConfig.getKeyStorePath(), + commonConfig.getKeyStorePwd()) + : DeepCopyRpcTransportFactory.INSTANCE.getTransport( + // As there is a try-catch already, we do not need to use TSocket.wrap + endpoint.getIp(), endpoint.getPort(), timeoutMs); if (!transport.isOpen()) { transport.open(); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeInternalRPCService.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeInternalRPCService.java index 08904b4e35a..b95adc0d764 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeInternalRPCService.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeInternalRPCService.java @@ -100,12 +100,12 @@ public class DataNodeInternalRPCService extends ThriftService @Override public String getBindIP() { - return IoTDBDescriptor.getInstance().getConfig().getInternalAddress(); + return config.getInternalAddress(); } @Override public int getBindPort() { - return IoTDBDescriptor.getInstance().getConfig().getInternalPort(); + return config.getInternalPort(); } public DataNodeInternalRPCServiceImpl getImpl() { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientPoolFactory.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientPoolFactory.java index 32c6345dc27..cecd4289bdb 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientPoolFactory.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientPoolFactory.java @@ -220,6 +220,30 @@ public class ClientPoolFactory { } } + public static class SyncDataNodeHeartbeatServiceClientPoolFactory + implements IClientPoolFactory<TEndPoint, SyncDataNodeInternalServiceClient> { + @Override + public GenericKeyedObjectPool<TEndPoint, SyncDataNodeInternalServiceClient> createClientPool( + ClientManager<TEndPoint, SyncDataNodeInternalServiceClient> manager) { + GenericKeyedObjectPool<TEndPoint, SyncDataNodeInternalServiceClient> clientPool = + new GenericKeyedObjectPool<>( + new SyncDataNodeInternalServiceClient.Factory( + manager, + new ThriftClientProperty.Builder() + .setConnectionTimeoutMs(conf.getCnConnectionTimeoutInMS()) + .setRpcThriftCompressionEnabled(conf.isRpcThriftCompressionEnabled()) + .setSelectorNumOfAsyncClientManager(conf.getSelectorNumOfClientManager()) + .setPrintLogWhenEncounterException(false) + .build()), + new ClientPoolProperty.Builder<SyncDataNodeInternalServiceClient>() + .build() + .getConfig()); + ClientManagerMetrics.getInstance() + .registerClientManager(this.getClass().getSimpleName(), clientPool); + return clientPool; + } + } + public static class SyncDataNodeMPPDataExchangeServiceClientPoolFactory implements IClientPoolFactory<TEndPoint, SyncDataNodeMPPDataExchangeServiceClient> { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncConfigNodeIServiceClient.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncConfigNodeIServiceClient.java index 99350ed32f7..3bcfa6495e3 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncConfigNodeIServiceClient.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncConfigNodeIServiceClient.java @@ -24,6 +24,8 @@ import org.apache.iotdb.commons.client.ClientManager; import org.apache.iotdb.commons.client.ThriftClient; import org.apache.iotdb.commons.client.factory.ThriftClientFactory; import org.apache.iotdb.commons.client.property.ThriftClientProperty; +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.confignode.rpc.thrift.IConfigNodeRPCService; import org.apache.iotdb.rpc.DeepCopyRpcTransportFactory; import org.apache.iotdb.rpc.TConfigurationConst; @@ -42,6 +44,7 @@ public class SyncConfigNodeIServiceClient extends IConfigNodeRPCService.Client private final boolean printLogWhenEncounterException; private final TEndPoint endpoint; private final ClientManager<TEndPoint, SyncConfigNodeIServiceClient> clientManager; + private static final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig(); public SyncConfigNodeIServiceClient( ThriftClientProperty property, @@ -52,12 +55,21 @@ public class SyncConfigNodeIServiceClient extends IConfigNodeRPCService.Client property .getProtocolFactory() .getProtocol( - DeepCopyRpcTransportFactory.INSTANCE.getTransport( - new TSocket( - TConfigurationConst.defaultTConfiguration, + commonConfig.isEnableSSL() + ? DeepCopyRpcTransportFactory.INSTANCE.getTransport( endPoint.getIp(), endPoint.getPort(), - property.getConnectionTimeoutMs())))); + property.getConnectionTimeoutMs(), + commonConfig.getTrustStorePath(), + commonConfig.getTrustStorePwd(), + commonConfig.getKeyStorePath(), + commonConfig.getKeyStorePwd()) + : DeepCopyRpcTransportFactory.INSTANCE.getTransport( + new TSocket( + TConfigurationConst.defaultTConfiguration, + endPoint.getIp(), + endPoint.getPort(), + property.getConnectionTimeoutMs())))); this.printLogWhenEncounterException = property.isPrintLogWhenEncounterException(); this.endpoint = endPoint; this.clientManager = clientManager; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncDataNodeInternalServiceClient.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncDataNodeInternalServiceClient.java index b82968d4f7d..4ec2fc1a46f 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncDataNodeInternalServiceClient.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncDataNodeInternalServiceClient.java @@ -24,6 +24,8 @@ import org.apache.iotdb.commons.client.ClientManager; import org.apache.iotdb.commons.client.ThriftClient; import org.apache.iotdb.commons.client.factory.ThriftClientFactory; import org.apache.iotdb.commons.client.property.ThriftClientProperty; +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.mpp.rpc.thrift.IDataNodeRPCService; import org.apache.iotdb.rpc.DeepCopyRpcTransportFactory; @@ -43,6 +45,7 @@ public class SyncDataNodeInternalServiceClient extends IDataNodeRPCService.Clien private final boolean printLogWhenEncounterException; private final TEndPoint endpoint; private final ClientManager<TEndPoint, SyncDataNodeInternalServiceClient> clientManager; + private static final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig(); public SyncDataNodeInternalServiceClient( ThriftClientProperty property, @@ -53,16 +56,27 @@ public class SyncDataNodeInternalServiceClient extends IDataNodeRPCService.Clien property .getProtocolFactory() .getProtocol( - DeepCopyRpcTransportFactory.INSTANCE.getTransport( - new TSocket( - TConfigurationConst.defaultTConfiguration, + commonConfig.isEnableSSL() + ? DeepCopyRpcTransportFactory.INSTANCE.getTransport( endpoint.getIp(), endpoint.getPort(), - property.getConnectionTimeoutMs())))); + property.getConnectionTimeoutMs(), + commonConfig.getTrustStorePath(), + commonConfig.getTrustStorePwd(), + commonConfig.getKeyStorePath(), + commonConfig.getKeyStorePwd()) + : DeepCopyRpcTransportFactory.INSTANCE.getTransport( + new TSocket( + TConfigurationConst.defaultTConfiguration, + endpoint.getIp(), + endpoint.getPort(), + property.getConnectionTimeoutMs())))); this.printLogWhenEncounterException = property.isPrintLogWhenEncounterException(); this.endpoint = endpoint; this.clientManager = clientManager; - getInputProtocol().getTransport().open(); + if (!getInputProtocol().getTransport().isOpen()) { + getInputProtocol().getTransport().open(); + } } public int getTimeout() throws SocketException { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncDataNodeMPPDataExchangeServiceClient.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncDataNodeMPPDataExchangeServiceClient.java index 592bb1ddccd..9b0a6136cb3 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncDataNodeMPPDataExchangeServiceClient.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncDataNodeMPPDataExchangeServiceClient.java @@ -24,6 +24,8 @@ import org.apache.iotdb.commons.client.ClientManager; import org.apache.iotdb.commons.client.ThriftClient; import org.apache.iotdb.commons.client.factory.ThriftClientFactory; import org.apache.iotdb.commons.client.property.ThriftClientProperty; +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.mpp.rpc.thrift.MPPDataExchangeService; import org.apache.iotdb.rpc.DeepCopyRpcTransportFactory; import org.apache.iotdb.rpc.TConfigurationConst; @@ -42,6 +44,7 @@ public class SyncDataNodeMPPDataExchangeServiceClient extends MPPDataExchangeSer private final boolean printLogWhenEncounterException; private final TEndPoint endpoint; private final ClientManager<TEndPoint, SyncDataNodeMPPDataExchangeServiceClient> clientManager; + private static final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig(); public SyncDataNodeMPPDataExchangeServiceClient( ThriftClientProperty property, @@ -52,12 +55,21 @@ public class SyncDataNodeMPPDataExchangeServiceClient extends MPPDataExchangeSer property .getProtocolFactory() .getProtocol( - DeepCopyRpcTransportFactory.INSTANCE.getTransport( - new TSocket( - TConfigurationConst.defaultTConfiguration, + commonConfig.isEnableSSL() + ? DeepCopyRpcTransportFactory.INSTANCE.getTransport( endpoint.getIp(), endpoint.getPort(), - property.getConnectionTimeoutMs())))); + property.getConnectionTimeoutMs(), + commonConfig.getTrustStorePath(), + commonConfig.getTrustStorePwd(), + commonConfig.getKeyStorePath(), + commonConfig.getKeyStorePwd()) + : DeepCopyRpcTransportFactory.INSTANCE.getTransport( + new TSocket( + TConfigurationConst.defaultTConfiguration, + endpoint.getIp(), + endpoint.getPort(), + property.getConnectionTimeoutMs())))); this.printLogWhenEncounterException = property.isPrintLogWhenEncounterException(); this.endpoint = endpoint; this.clientManager = clientManager; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncPipeConsensusServiceClient.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncPipeConsensusServiceClient.java index 6cc9897f23a..d79854d29ed 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncPipeConsensusServiceClient.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncPipeConsensusServiceClient.java @@ -24,6 +24,8 @@ import org.apache.iotdb.commons.client.ClientManager; import org.apache.iotdb.commons.client.ThriftClient; import org.apache.iotdb.commons.client.factory.ThriftClientFactory; import org.apache.iotdb.commons.client.property.ThriftClientProperty; +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.consensus.pipe.thrift.PipeConsensusIService; import org.apache.iotdb.rpc.DeepCopyRpcTransportFactory; import org.apache.iotdb.rpc.TConfigurationConst; @@ -42,6 +44,7 @@ public class SyncPipeConsensusServiceClient extends PipeConsensusIService.Client private final boolean printLogWhenEncounterException; private final TEndPoint endpoint; private final ClientManager<TEndPoint, SyncPipeConsensusServiceClient> clientManager; + private static final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig(); public SyncPipeConsensusServiceClient( ThriftClientProperty property, @@ -52,12 +55,21 @@ public class SyncPipeConsensusServiceClient extends PipeConsensusIService.Client property .getProtocolFactory() .getProtocol( - DeepCopyRpcTransportFactory.INSTANCE.getTransport( - new TSocket( - TConfigurationConst.defaultTConfiguration, + commonConfig.isEnableSSL() + ? DeepCopyRpcTransportFactory.INSTANCE.getTransport( endpoint.getIp(), endpoint.getPort(), - property.getConnectionTimeoutMs())))); + property.getConnectionTimeoutMs(), + commonConfig.getTrustStorePath(), + commonConfig.getTrustStorePwd(), + commonConfig.getKeyStorePath(), + commonConfig.getKeyStorePwd()) + : DeepCopyRpcTransportFactory.INSTANCE.getTransport( + new TSocket( + TConfigurationConst.defaultTConfiguration, + endpoint.getIp(), + endpoint.getPort(), + property.getConnectionTimeoutMs())))); this.printLogWhenEncounterException = property.isPrintLogWhenEncounterException(); this.endpoint = endpoint; this.clientManager = clientManager; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/AbstractThriftServiceThread.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/AbstractThriftServiceThread.java index 70ef38b09d0..6c1aed0b55d 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/AbstractThriftServiceThread.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/AbstractThriftServiceThread.java @@ -179,7 +179,9 @@ public abstract class AbstractThriftServiceThread extends Thread { TSSLTransportFactory.TSSLTransportParameters params = new TSSLTransportFactory.TSSLTransportParameters(); params.setKeyStore(keyStorePath, keyStorePwd); - params.setTrustStore(trustStorePath, trustStorePwd); + if (trustStorePath != null && !trustStorePath.isEmpty()) { + params.setTrustStore(trustStorePath, trustStorePwd); + } params.requireClientAuth(false); InetSocketAddress socketAddress = new InetSocketAddress(bindAddress, port); serverTransport =
