You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by qi...@apache.org on 2022/06/14 02:11:09 UTC
[iotdb] 01/01: Revert "[IOTDB-3390] Serialize ConfigNode expansion (#6256)"
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch revert-6256-Serialize_ConfigNode_expansion
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 02c4f0464fba333e90c9178145f541e1fee5e080
Author: Jialin Qiao <qj...@mails.tsinghua.edu.cn>
AuthorDate: Tue Jun 14 10:11:02 2022 +0800
Revert "[IOTDB-3390] Serialize ConfigNode expansion (#6256)"
This reverts commit c742abeff67b805ad685f553502a9a5231cfd45c.
---
.../resources/conf/iotdb-confignode.properties | 6 -
...lientPool.java => AsyncDataNodeClientPool.java} | 12 +-
...ientPool.java => SyncConfigNodeClientPool.java} | 55 +++---
...ClientPool.java => SyncDataNodeClientPool.java} | 12 +-
.../iotdb/confignode/conf/ConfigNodeConf.java | 28 ++-
.../confignode/conf/ConfigNodeDescriptor.java | 9 -
.../confignode/conf/ConfigNodeStartupCheck.java | 100 +++--------
.../consensus/request/ConfigRequest.java | 6 +-
.../consensus/request/ConfigRequestType.java | 2 +-
...rConfigNodeReq.java => ApplyConfigNodeReq.java} | 12 +-
.../iotdb/confignode/manager/ConfigManager.java | 97 +++++++++--
.../iotdb/confignode/manager/ConsensusManager.java | 38 ++++-
.../apache/iotdb/confignode/manager/Manager.java | 8 +
.../iotdb/confignode/manager/NodeManager.java | 190 +++++++++------------
.../iotdb/confignode/manager/PartitionManager.java | 4 +-
.../confignode/manager/PermissionManager.java | 4 +-
.../iotdb/confignode/manager/UDFManager.java | 6 +-
.../iotdb/confignode/manager/load/LoadManager.java | 6 +-
.../iotdb/confignode/persistence/NodeInfo.java | 12 +-
.../executor/ConfigRequestExecutor.java | 6 +-
.../procedure/env/ConfigNodeProcedureEnv.java | 6 +-
.../iotdb/confignode/service/ConfigNode.java | 10 +-
.../thrift/ConfigNodeRPCServiceProcessor.java | 16 +-
.../consensus/request/ConfigRequestSerDeTest.java | 10 +-
.../iotdb/confignode/persistence/NodeInfoTest.java | 5 +-
.../apache/iotdb/db/auth/AuthorizerManager.java | 11 +-
.../iotdb/db/auth/ClusterAuthorityFetcher.java | 25 ++-
.../apache/iotdb/db/auth/IAuthorityFetcher.java | 6 +-
.../iotdb/db/auth/StandaloneAuthorityFetcher.java | 6 +-
...ConfigNodeClient.java => ConfigNodeClient.java} | 61 ++++---
.../iotdb/db/client/DataNodeClientPoolFactory.java | 10 +-
.../org/apache/iotdb/db/mpp/plan/Coordinator.java | 6 +-
.../mpp/plan/analyze/ClusterPartitionFetcher.java | 25 ++-
.../mpp/plan/execution/config/AuthorizerTask.java | 10 +-
.../mpp/plan/execution/config/ConfigExecution.java | 6 +-
.../execution/config/CountStorageGroupTask.java | 7 +-
.../plan/execution/config/CreateFunctionTask.java | 9 +-
.../execution/config/DeleteStorageGroupTask.java | 7 +-
.../plan/execution/config/DropFunctionTask.java | 9 +-
.../db/mpp/plan/execution/config/FlushTask.java | 7 +-
.../db/mpp/plan/execution/config/IConfigTask.java | 4 +-
.../plan/execution/config/SetStorageGroupTask.java | 8 +-
.../db/mpp/plan/execution/config/SetTTLTask.java | 8 +-
.../mpp/plan/execution/config/ShowClusterTask.java | 7 +-
.../plan/execution/config/ShowFunctionsTask.java | 4 +-
.../execution/config/ShowStorageGroupTask.java | 7 +-
.../db/mpp/plan/execution/config/ShowTTLTask.java | 7 +-
.../java/org/apache/iotdb/db/service/DataNode.java | 8 +-
.../db/mpp/execution/ConfigExecutionTest.java | 4 +-
.../java/org/apache/iotdb/rpc/TSStatusCode.java | 7 +-
.../src/main/thrift/confignode.thrift | 47 ++---
51 files changed, 499 insertions(+), 477 deletions(-)
diff --git a/confignode/src/assembly/resources/conf/iotdb-confignode.properties b/confignode/src/assembly/resources/conf/iotdb-confignode.properties
index fe80a8ef00..f067335003 100644
--- a/confignode/src/assembly/resources/conf/iotdb-confignode.properties
+++ b/confignode/src/assembly/resources/conf/iotdb-confignode.properties
@@ -47,16 +47,10 @@ consensus_port=22278
target_confignode=0.0.0.0:22277
-# Used to identify clusters
-# Modify this parameter if multiple IoTDB clusters need to be created.
-# Datatype: int
-# cluster_id=0
-
####################
### Consensus protocol configuration
####################
-
# All parameters in Consensus protocol configuration is unmodifiable after ConfigNode starts for the first time.
# And these parameters should be consistent within the ConfigNodeGroup.
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/client/AsyncConfigNodeToDataNodeClientPool.java b/confignode/src/main/java/org/apache/iotdb/confignode/client/AsyncDataNodeClientPool.java
similarity index 96%
rename from confignode/src/main/java/org/apache/iotdb/confignode/client/AsyncConfigNodeToDataNodeClientPool.java
rename to confignode/src/main/java/org/apache/iotdb/confignode/client/AsyncDataNodeClientPool.java
index 564ee074be..a38218ba04 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/client/AsyncConfigNodeToDataNodeClientPool.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/client/AsyncDataNodeClientPool.java
@@ -46,14 +46,13 @@ import java.util.TreeMap;
import java.util.concurrent.CountDownLatch;
/** Asynchronously send RPC requests to DataNodes. See mpp.thrift for more details. */
-public class AsyncConfigNodeToDataNodeClientPool {
+public class AsyncDataNodeClientPool {
- private static final Logger LOGGER =
- LoggerFactory.getLogger(AsyncConfigNodeToDataNodeClientPool.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(AsyncDataNodeClientPool.class);
private final IClientManager<TEndPoint, AsyncDataNodeInternalServiceClient> clientManager;
- private AsyncConfigNodeToDataNodeClientPool() {
+ private AsyncDataNodeClientPool() {
clientManager =
new IClientManager.Factory<TEndPoint, AsyncDataNodeInternalServiceClient>()
.createClientManager(
@@ -268,15 +267,14 @@ public class AsyncConfigNodeToDataNodeClientPool {
// TODO: Is the ClientPool must be a singleton?
private static class ClientPoolHolder {
- private static final AsyncConfigNodeToDataNodeClientPool INSTANCE =
- new AsyncConfigNodeToDataNodeClientPool();
+ private static final AsyncDataNodeClientPool INSTANCE = new AsyncDataNodeClientPool();
private ClientPoolHolder() {
// Empty constructor
}
}
- public static AsyncConfigNodeToDataNodeClientPool getInstance() {
+ public static AsyncDataNodeClientPool getInstance() {
return ClientPoolHolder.INSTANCE;
}
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/client/SyncConfigNodeToConfigNodeClientPool.java b/confignode/src/main/java/org/apache/iotdb/confignode/client/SyncConfigNodeClientPool.java
similarity index 65%
rename from confignode/src/main/java/org/apache/iotdb/confignode/client/SyncConfigNodeToConfigNodeClientPool.java
rename to confignode/src/main/java/org/apache/iotdb/confignode/client/SyncConfigNodeClientPool.java
index ab9827718b..484687eacd 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/client/SyncConfigNodeToConfigNodeClientPool.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/client/SyncConfigNodeClientPool.java
@@ -18,6 +18,7 @@
*/
package org.apache.iotdb.confignode.client;
+import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.client.IClientManager;
@@ -32,35 +33,23 @@ import org.slf4j.LoggerFactory;
import java.util.concurrent.TimeUnit;
-/**
- * The SyncConfigNodeClientPool synchronously send RPC requests to a specific ConfigNode. The retry
- * logic is included to avoid network failures like interruption and timeout. For more details about
- * the interfaces, see confignode.thrift file.
- */
-public class SyncConfigNodeToConfigNodeClientPool {
+/** Synchronously send RPC requests to ConfigNode. See confignode.thrift for more details. */
+public class SyncConfigNodeClientPool {
- private static final Logger LOGGER =
- LoggerFactory.getLogger(SyncConfigNodeToConfigNodeClientPool.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(SyncConfigNodeClientPool.class);
private static final int retryNum = 5;
private final IClientManager<TEndPoint, SyncConfigNodeIServiceClient> clientManager;
- private SyncConfigNodeToConfigNodeClientPool() {
- this.clientManager =
+ private SyncConfigNodeClientPool() {
+ clientManager =
new IClientManager.Factory<TEndPoint, SyncConfigNodeIServiceClient>()
.createClientManager(
new DataNodeClientPoolFactory.SyncConfigNodeIServiceClientPoolFactory());
}
- /**
- * Register when the ConfigNode isn't seed and is first startup.
- *
- * @param endPoint RPC target
- * @param req TConfigNodeRegisterReq
- * @return Complete TConfigNodeRegisterResp when the request arrives successfully,
- * ALL_RETRY_FAILED otherwise.
- */
+ /** Only use registerConfigNode when the ConfigNode is first startup. */
public TConfigNodeRegisterResp registerConfigNode(
TEndPoint endPoint, TConfigNodeRegisterReq req) {
// TODO: Unified retry logic
@@ -69,19 +58,34 @@ public class SyncConfigNodeToConfigNodeClientPool {
return client.registerConfigNode(req);
} catch (Exception e) {
LOGGER.warn("Register ConfigNode failed, retrying...", e);
- doRetryWait(retry);
+ doRetryWait();
}
}
- LOGGER.error("Register ConfigNode failed because network failure.");
+ LOGGER.error("Register ConfigNode failed");
return new TConfigNodeRegisterResp()
.setStatus(
- new TSStatus(TSStatusCode.ALL_RETRY_FAILEDD.getStatusCode())
+ new TSStatus(TSStatusCode.ALL_RETRY_FAILED.getStatusCode())
.setMessage("All retry failed."));
}
- private void doRetryWait(int retryNum) {
+ public TSStatus applyConfigNode(TEndPoint endPoint, TConfigNodeLocation configNodeLocation) {
+ // TODO: Unified retry logic
+ for (int retry = 0; retry < retryNum; retry++) {
+ try (SyncConfigNodeIServiceClient client = clientManager.borrowClient(endPoint)) {
+ return client.applyConfigNode(configNodeLocation);
+ } catch (Exception e) {
+ LOGGER.warn("Apply ConfigNode failed, retrying...", e);
+ doRetryWait();
+ }
+ }
+ LOGGER.error("Apply ConfigNode failed");
+ return new TSStatus(TSStatusCode.ALL_RETRY_FAILED.getStatusCode())
+ .setMessage("All retry failed.");
+ }
+
+ private void doRetryWait() {
try {
- TimeUnit.MILLISECONDS.sleep((long) Math.pow(2, retryNum) * 100);
+ TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) {
LOGGER.error("Retry wait failed.", e);
}
@@ -90,15 +94,14 @@ public class SyncConfigNodeToConfigNodeClientPool {
// TODO: Is the ClientPool must be a singleton?
private static class SyncConfigNodeClientPoolHolder {
- private static final SyncConfigNodeToConfigNodeClientPool INSTANCE =
- new SyncConfigNodeToConfigNodeClientPool();
+ private static final SyncConfigNodeClientPool INSTANCE = new SyncConfigNodeClientPool();
private SyncConfigNodeClientPoolHolder() {
// Empty constructor
}
}
- public static SyncConfigNodeToConfigNodeClientPool getInstance() {
+ public static SyncConfigNodeClientPool getInstance() {
return SyncConfigNodeClientPoolHolder.INSTANCE;
}
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/client/SyncConfigNodeToDataNodeClientPool.java b/confignode/src/main/java/org/apache/iotdb/confignode/client/SyncDataNodeClientPool.java
similarity index 94%
rename from confignode/src/main/java/org/apache/iotdb/confignode/client/SyncConfigNodeToDataNodeClientPool.java
rename to confignode/src/main/java/org/apache/iotdb/confignode/client/SyncDataNodeClientPool.java
index 8f73e587af..90f17ce679 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/client/SyncConfigNodeToDataNodeClientPool.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/client/SyncDataNodeClientPool.java
@@ -42,14 +42,13 @@ import java.util.Map;
import java.util.Set;
/** Asynchronously send RPC requests to DataNodes. See mpp.thrift for more details. */
-public class SyncConfigNodeToDataNodeClientPool {
+public class SyncDataNodeClientPool {
- private static final Logger LOGGER =
- LoggerFactory.getLogger(SyncConfigNodeToDataNodeClientPool.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(SyncDataNodeClientPool.class);
private final IClientManager<TEndPoint, SyncDataNodeInternalServiceClient> clientManager;
- private SyncConfigNodeToDataNodeClientPool() {
+ private SyncDataNodeClientPool() {
clientManager =
new IClientManager.Factory<TEndPoint, SyncDataNodeInternalServiceClient>()
.createClientManager(
@@ -157,15 +156,14 @@ public class SyncConfigNodeToDataNodeClientPool {
// TODO: Is the ClientPool must be a singleton?
private static class ClientPoolHolder {
- private static final SyncConfigNodeToDataNodeClientPool INSTANCE =
- new SyncConfigNodeToDataNodeClientPool();
+ private static final SyncDataNodeClientPool INSTANCE = new SyncDataNodeClientPool();
private ClientPoolHolder() {
// Empty constructor
}
}
- public static SyncConfigNodeToDataNodeClientPool getInstance() {
+ public static SyncDataNodeClientPool getInstance() {
return ClientPoolHolder.INSTANCE;
}
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConf.java b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConf.java
index df734f16e5..1d601a7b03 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConf.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConf.java
@@ -19,14 +19,13 @@
package org.apache.iotdb.confignode.conf;
import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
-import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
-import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.rpc.RpcUtils;
import java.io.File;
+import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
@@ -44,12 +43,14 @@ public class ConfigNodeConf {
/** Used for connecting to the ConfigNodeGroup */
private TEndPoint targetConfigNode = new TEndPoint("0.0.0.0", 22277);
- /** Used to identify clusters */
- private TConsensusGroupId partitionRegionId =
- new TConsensusGroupId(TConsensusGroupType.PartitionRegion, 0);
+ /** Mark if the ConfigNode needs to apply */
+ private boolean needApply = false;
+
+ // TODO: Read from iotdb-confignode.properties
+ private int partitionRegionId = 0;
/** Used for building the PartitionRegion */
- private volatile List<TConfigNodeLocation> configNodeList;
+ private List<TConfigNodeLocation> configNodeList = new ArrayList<>();
/** Thrift socket and connection timeout between nodes */
private int connectionTimeoutInMS = (int) TimeUnit.SECONDS.toMillis(20);
@@ -193,6 +194,14 @@ public class ConfigNodeConf {
this.consensusPort = consensusPort;
}
+ public boolean isNeedApply() {
+ return needApply;
+ }
+
+ public void setNeedApply(boolean needApply) {
+ this.needApply = needApply;
+ }
+
public TEndPoint getTargetConfigNode() {
return targetConfigNode;
}
@@ -201,11 +210,11 @@ public class ConfigNodeConf {
this.targetConfigNode = targetConfigNode;
}
- public TConsensusGroupId getPartitionRegionId() {
+ public int getPartitionRegionId() {
return partitionRegionId;
}
- public void setPartitionRegionId(TConsensusGroupId partitionRegionId) {
+ public void setPartitionRegionId(int partitionRegionId) {
this.partitionRegionId = partitionRegionId;
}
@@ -289,8 +298,9 @@ public class ConfigNodeConf {
return connectionTimeoutInMS;
}
- public void setConnectionTimeoutInMS(int connectionTimeoutInMS) {
+ public ConfigNodeConf setConnectionTimeoutInMS(int connectionTimeoutInMS) {
this.connectionTimeoutInMS = connectionTimeoutInMS;
+ return this;
}
public void setSelectorNumOfClientManager(int selectorNumOfClientManager) {
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java
index 7b40b05725..515669ff19 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java
@@ -18,8 +18,6 @@
*/
package org.apache.iotdb.confignode.conf;
-import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
-import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.exception.BadNodeUrlException;
import org.apache.iotdb.commons.utils.NodeUrlUtils;
@@ -122,13 +120,6 @@ public class ConfigNodeDescriptor {
conf.setTargetConfigNode(NodeUrlUtils.parseTEndPointUrl(targetConfigNode));
}
- int clusterId =
- Integer.parseInt(
- properties.getProperty(
- "cluster_id", String.valueOf(conf.getPartitionRegionId().getId())));
- conf.setPartitionRegionId(
- new TConsensusGroupId(TConsensusGroupType.PartitionRegion, clusterId));
-
conf.setSeriesPartitionSlotNum(
Integer.parseInt(
properties.getProperty(
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheck.java b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheck.java
index 25e546eef9..2c0c1c8f0a 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheck.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheck.java
@@ -25,10 +25,9 @@ import org.apache.iotdb.commons.exception.BadNodeUrlException;
import org.apache.iotdb.commons.exception.ConfigurationException;
import org.apache.iotdb.commons.exception.StartupException;
import org.apache.iotdb.commons.utils.NodeUrlUtils;
-import org.apache.iotdb.confignode.client.SyncConfigNodeToConfigNodeClientPool;
+import org.apache.iotdb.confignode.client.SyncConfigNodeClientPool;
import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeRegisterReq;
import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeRegisterResp;
-import org.apache.iotdb.confignode.rpc.thrift.TGlobalConfig;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -40,11 +39,8 @@ import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Collections;
-import java.util.List;
import java.util.Objects;
import java.util.Properties;
-import java.util.Random;
-import java.util.concurrent.TimeUnit;
/**
* ConfigNodeStartupCheck checks the cluster status and parameters in iotdb-confignode.properties
@@ -66,12 +62,14 @@ public class ConfigNodeStartupCheck {
}
public void startUpCheck() throws StartupException, IOException, ConfigurationException {
- checkConfigConflict();
+ checkGlobalConfig();
if (isFirstStart()) {
// Do initialization when first start
if (!isSeedConfigNode()) {
// Register when the current ConfigNode isn't Seed-ConfigNode
registerConfigNode();
+ // Apply after constructing PartitionRegion
+ conf.setNeedApply(true);
}
// Persistence the unchangeable parameters
writeSystemProperties();
@@ -82,8 +80,8 @@ public class ConfigNodeStartupCheck {
}
}
- /** Check whether the parameters conflict */
- private void checkConfigConflict() throws ConfigurationException {
+ /** Check whether the global configuration of the cluster is correct */
+ private void checkGlobalConfig() throws ConfigurationException {
// When the ConfigNode consensus protocol is set to StandAlone,
// the target_configNode needs to point to itself
if (conf.getConfigNodeConsensusProtocolClass().equals(ConsensusFactory.StandAloneConsensus)
@@ -159,29 +157,23 @@ public class ConfigNodeStartupCheck {
}
/**
- * Check if the current ConfigNode is SeedConfigNode.
+ * Check if the current ConfigNode is SeedConfigNode. If true, do the SeedConfigNode configuration
+ * as well.
*
- * @return True if the target_confignode points to itself, false otherwise.
+ * @return True if the target_confignode points to itself
*/
private boolean isSeedConfigNode() {
boolean result =
conf.getRpcAddress().equals(conf.getTargetConfigNode().getIp())
&& conf.getRpcPort() == conf.getTargetConfigNode().getPort();
-
if (result) {
- // If the current ConfigNode is seed, set configNodeList to itself
+ // TODO: Set PartitionRegionId from iotdb-confignode.properties
conf.setConfigNodeList(
Collections.singletonList(
new TConfigNodeLocation(
new TEndPoint(conf.getRpcAddress(), conf.getRpcPort()),
new TEndPoint(conf.getRpcAddress(), conf.getConsensusPort()))));
- } else {
- // If the current ConfigNode isn't seed, set configNodeList to targetConfigNode
- conf.setConfigNodeList(
- Collections.singletonList(
- new TConfigNodeLocation(conf.getTargetConfigNode(), new TEndPoint())));
}
-
return result;
}
@@ -189,74 +181,26 @@ public class ConfigNodeStartupCheck {
private void registerConfigNode() throws StartupException {
TConfigNodeRegisterReq req =
new TConfigNodeRegisterReq(
- conf.getPartitionRegionId(),
new TConfigNodeLocation(
new TEndPoint(conf.getRpcAddress(), conf.getRpcPort()),
new TEndPoint(conf.getRpcAddress(), conf.getConsensusPort())),
- new TGlobalConfig(
- conf.getDataRegionConsensusProtocolClass(),
- conf.getSchemaRegionConsensusProtocolClass(),
- conf.getSeriesPartitionSlotNum(),
- conf.getSeriesPartitionExecutorClass(),
- conf.getTimePartitionInterval()),
+ conf.getDataRegionConsensusProtocolClass(),
+ conf.getSchemaRegionConsensusProtocolClass(),
+ conf.getSeriesPartitionSlotNum(),
+ conf.getSeriesPartitionExecutorClass(),
CommonDescriptor.getInstance().getConfig().getDefaultTTL(),
+ conf.getTimePartitionInterval(),
conf.getSchemaReplicationFactor(),
conf.getDataReplicationFactor());
- long timeout = 100;
- TEndPoint leader = null;
- Random random = new Random();
- for (int retry = 0; retry < 10; retry++) {
- TEndPoint target = null;
- TConfigNodeRegisterResp resp;
- if (leader == null) {
- // Pick a random ConfigNode and send request
- List<TConfigNodeLocation> configNodeList = conf.getConfigNodeList();
- while (target == null
- || target.equals(new TEndPoint(conf.getRpcAddress(), conf.getRpcPort()))) {
- target = configNodeList.get(random.nextInt(configNodeList.size())).getInternalEndPoint();
- }
- resp = SyncConfigNodeToConfigNodeClientPool.getInstance().registerConfigNode(target, req);
- } else {
- // Send request to leader
- resp = SyncConfigNodeToConfigNodeClientPool.getInstance().registerConfigNode(leader, req);
- }
-
- // Update configNodeList
+ TConfigNodeRegisterResp resp =
+ SyncConfigNodeClientPool.getInstance().registerConfigNode(conf.getTargetConfigNode(), req);
+ if (resp.getStatus().getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ conf.setPartitionRegionId(resp.getPartitionRegionId().getId());
conf.setConfigNodeList(resp.getConfigNodeList());
-
- if (resp.getStatus().getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- return;
- } else if (resp.getStatus().getCode()
- == TSStatusCode.INCONSISTENT_SYSTEM_CONFIG.getStatusCode()) {
- // Log the inconsistent parameters
- LOGGER.error(resp.getStatus().getMessage());
- break;
- } else if (resp.getStatus().getCode() == TSStatusCode.NEED_REDIRECTION.getStatusCode()) {
- // Try to cache leader
- if (resp.getStatus().isSetRedirectNode()) {
- leader = resp.getStatus().getRedirectNode();
- }
- } else if (resp.getStatus().getCode() == TSStatusCode.NEED_RETRY.getStatusCode()) {
- // In this case, we're already locate the leader
- leader = target;
- try {
- TimeUnit.MILLISECONDS.sleep(timeout);
- timeout *= 2;
- } catch (InterruptedException e) {
- // Just ignore and retry
- }
- } else {
- // There are network failure
- try {
- TimeUnit.MILLISECONDS.sleep(100);
- } catch (InterruptedException e) {
- // Just ignore and retry
- }
- }
+ } else {
+ throw new StartupException("Register ConfigNode failed!");
}
-
- throw new StartupException("Register ConfigNode failed!");
}
/**
@@ -316,7 +260,7 @@ public class ConfigNodeStartupCheck {
}
}
- /** Ensure that system parameters are consistent with each startup except the first one */
+ /** Ensure that special parameters are consistent with each startup except the first one */
private void checkSystemProperties()
throws ConfigurationException, IOException, StartupException {
boolean needReWrite = false;
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigRequest.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigRequest.java
index 7f9992120a..f1c2e896c8 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigRequest.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigRequest.java
@@ -27,6 +27,7 @@ import org.apache.iotdb.confignode.consensus.request.read.GetOrCreateDataPartiti
import org.apache.iotdb.confignode.consensus.request.read.GetOrCreateSchemaPartitionReq;
import org.apache.iotdb.confignode.consensus.request.read.GetSchemaPartitionReq;
import org.apache.iotdb.confignode.consensus.request.read.GetStorageGroupReq;
+import org.apache.iotdb.confignode.consensus.request.write.ApplyConfigNodeReq;
import org.apache.iotdb.confignode.consensus.request.write.CreateDataPartitionReq;
import org.apache.iotdb.confignode.consensus.request.write.CreateFunctionReq;
import org.apache.iotdb.confignode.consensus.request.write.CreateRegionsReq;
@@ -36,7 +37,6 @@ import org.apache.iotdb.confignode.consensus.request.write.DeleteRegionsReq;
import org.apache.iotdb.confignode.consensus.request.write.DeleteStorageGroupReq;
import org.apache.iotdb.confignode.consensus.request.write.DropFunctionReq;
import org.apache.iotdb.confignode.consensus.request.write.PreDeleteStorageGroupReq;
-import org.apache.iotdb.confignode.consensus.request.write.RegisterConfigNodeReq;
import org.apache.iotdb.confignode.consensus.request.write.RegisterDataNodeReq;
import org.apache.iotdb.confignode.consensus.request.write.SetDataReplicationFactorReq;
import org.apache.iotdb.confignode.consensus.request.write.SetSchemaReplicationFactorReq;
@@ -177,8 +177,8 @@ public abstract class ConfigRequest implements IConsensusRequest {
case UpdateUser:
req = new AuthorReq(type);
break;
- case RegisterConfigNode:
- req = new RegisterConfigNodeReq();
+ case ApplyConfigNode:
+ req = new ApplyConfigNodeReq();
break;
case CreateFunction:
req = new CreateFunctionReq();
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigRequestType.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigRequestType.java
index 35d08f1c0b..a8f122e412 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigRequestType.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigRequestType.java
@@ -58,7 +58,7 @@ public enum ConfigRequestType {
ListRolePrivilege,
ListUserRoles,
ListRoleUsers,
- RegisterConfigNode,
+ ApplyConfigNode,
CreateFunction,
DropFunction,
GetNodePathsPartition;
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/RegisterConfigNodeReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/ApplyConfigNodeReq.java
similarity index 85%
rename from confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/RegisterConfigNodeReq.java
rename to confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/ApplyConfigNodeReq.java
index 8761129f92..00e55b6369 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/RegisterConfigNodeReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/ApplyConfigNodeReq.java
@@ -29,15 +29,15 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
-public class RegisterConfigNodeReq extends ConfigRequest {
+public class ApplyConfigNodeReq extends ConfigRequest {
private TConfigNodeLocation configNodeLocation;
- public RegisterConfigNodeReq() {
- super(ConfigRequestType.RegisterConfigNode);
+ public ApplyConfigNodeReq() {
+ super(ConfigRequestType.ApplyConfigNode);
}
- public RegisterConfigNodeReq(TConfigNodeLocation configNodeLocation) {
+ public ApplyConfigNodeReq(TConfigNodeLocation configNodeLocation) {
this();
this.configNodeLocation = configNodeLocation;
}
@@ -48,7 +48,7 @@ public class RegisterConfigNodeReq extends ConfigRequest {
@Override
protected void serializeImpl(DataOutputStream stream) throws IOException {
- ReadWriteIOUtils.write(ConfigRequestType.RegisterConfigNode.ordinal(), stream);
+ ReadWriteIOUtils.write(ConfigRequestType.ApplyConfigNode.ordinal(), stream);
ThriftConfigNodeSerDeUtils.serializeTConfigNodeLocation(configNodeLocation, stream);
}
@@ -62,7 +62,7 @@ public class RegisterConfigNodeReq extends ConfigRequest {
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
- RegisterConfigNodeReq that = (RegisterConfigNodeReq) o;
+ ApplyConfigNodeReq that = (ApplyConfigNodeReq) o;
return configNodeLocation.equals(that.configNodeLocation);
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
index 047b93224c..e81c72d76c 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
@@ -21,9 +21,11 @@ package org.apache.iotdb.confignode.manager;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.utils.AuthUtils;
import org.apache.iotdb.commons.utils.PathUtils;
+import org.apache.iotdb.confignode.conf.ConfigNodeConf;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
import org.apache.iotdb.confignode.consensus.request.auth.AuthorReq;
import org.apache.iotdb.confignode.consensus.request.read.CountStorageGroupReq;
@@ -34,6 +36,7 @@ import org.apache.iotdb.confignode.consensus.request.read.GetOrCreateDataPartiti
import org.apache.iotdb.confignode.consensus.request.read.GetOrCreateSchemaPartitionReq;
import org.apache.iotdb.confignode.consensus.request.read.GetSchemaPartitionReq;
import org.apache.iotdb.confignode.consensus.request.read.GetStorageGroupReq;
+import org.apache.iotdb.confignode.consensus.request.write.ApplyConfigNodeReq;
import org.apache.iotdb.confignode.consensus.request.write.RegisterDataNodeReq;
import org.apache.iotdb.confignode.consensus.request.write.SetDataReplicationFactorReq;
import org.apache.iotdb.confignode.consensus.request.write.SetSchemaReplicationFactorReq;
@@ -436,6 +439,16 @@ public class ConfigManager implements Manager {
}
}
+ private TSStatus confirmLeader() {
+ if (getConsensusManager().isLeader()) {
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ } else {
+ return new TSStatus(TSStatusCode.NEED_REDIRECTION.getStatusCode())
+ .setMessage(
+ "The current ConfigNode is not leader. And ConfigNodeGroup is in leader election. Please redirect with a random ConfigNode.");
+ }
+ }
+
@Override
public NodeManager getNodeManager() {
return nodeManager;
@@ -510,12 +523,76 @@ public class ConfigManager implements Manager {
@Override
public TConfigNodeRegisterResp registerConfigNode(TConfigNodeRegisterReq req) {
- TSStatus status = confirmLeader();
- if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- return nodeManager.registerConfigNode(req);
- } else {
- return new TConfigNodeRegisterResp(status, getNodeManager().getOnlineConfigNodes());
+ // Check global configuration
+ ConfigNodeConf conf = ConfigNodeDescriptor.getInstance().getConf();
+ TConfigNodeRegisterResp errorResp = new TConfigNodeRegisterResp();
+ errorResp.setStatus(new TSStatus(TSStatusCode.ERROR_GLOBAL_CONFIG.getStatusCode()));
+ if (!req.getDataRegionConsensusProtocolClass()
+ .equals(conf.getDataRegionConsensusProtocolClass())) {
+ errorResp
+ .getStatus()
+ .setMessage(
+ "Reject register, please ensure that the data_region_consensus_protocol_class "
+ + "are consistent.");
+ return errorResp;
+ }
+ if (!req.getSchemaRegionConsensusProtocolClass()
+ .equals(conf.getSchemaRegionConsensusProtocolClass())) {
+ errorResp
+ .getStatus()
+ .setMessage(
+ "Reject register, please ensure that the schema_region_consensus_protocol_class "
+ + "are consistent.");
+ return errorResp;
+ }
+ if (req.getSeriesPartitionSlotNum() != conf.getSeriesPartitionSlotNum()) {
+ errorResp
+ .getStatus()
+ .setMessage(
+ "Reject register, please ensure that the series_partition_slot_num are consistent.");
+ return errorResp;
+ }
+ if (!req.getSeriesPartitionExecutorClass().equals(conf.getSeriesPartitionExecutorClass())) {
+ errorResp
+ .getStatus()
+ .setMessage(
+ "Reject register, please ensure that the series_partition_executor_class are consistent.");
+ return errorResp;
+ }
+ if (req.getDefaultTTL() != CommonDescriptor.getInstance().getConfig().getDefaultTTL()) {
+ errorResp
+ .getStatus()
+ .setMessage("Reject register, please ensure that the default_ttl are consistent.");
+ return errorResp;
}
+ if (req.getTimePartitionInterval() != conf.getTimePartitionInterval()) {
+ errorResp
+ .getStatus()
+ .setMessage(
+ "Reject register, please ensure that the time_partition_interval are consistent.");
+ return errorResp;
+ }
+ if (req.getSchemaReplicationFactor() != conf.getSchemaReplicationFactor()) {
+ errorResp
+ .getStatus()
+ .setMessage(
+ "Reject register, please ensure that the schema_replication_factor are consistent.");
+ return errorResp;
+ }
+ if (req.getDataReplicationFactor() != conf.getDataReplicationFactor()) {
+ errorResp
+ .getStatus()
+ .setMessage(
+ "Reject register, please ensure that the data_replication_factor are consistent.");
+ return errorResp;
+ }
+
+ return nodeManager.registerConfigNode(req);
+ }
+
+ @Override
+ public TSStatus applyConfigNode(ApplyConfigNodeReq applyConfigNodeReq) {
+ return nodeManager.applyConfigNode(applyConfigNodeReq);
}
@Override
@@ -543,16 +620,6 @@ public class ConfigManager implements Manager {
return procedureManager;
}
- private TSStatus confirmLeader() {
- if (getConsensusManager().isLeader()) {
- return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
- } else {
- return new TSStatus(TSStatusCode.NEED_REDIRECTION.getStatusCode())
- .setMessage(
- "The current ConfigNode is not leader. And ConfigNodeGroup is in leader election. Please redirect with a random ConfigNode.");
- }
- }
-
/**
* @param storageGroups the storage groups to check
* @return List of PartialPath the storage groups that not exist
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConsensusManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConsensusManager.java
index 12cb1a12b6..d46e6e66b3 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConsensusManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConsensusManager.java
@@ -20,12 +20,15 @@ package org.apache.iotdb.confignode.manager;
import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
+import org.apache.iotdb.commons.consensus.PartitionRegionId;
import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.confignode.client.SyncConfigNodeClientPool;
import org.apache.iotdb.confignode.conf.ConfigNodeConf;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
-import org.apache.iotdb.confignode.consensus.request.write.RegisterConfigNodeReq;
+import org.apache.iotdb.confignode.consensus.request.write.ApplyConfigNodeReq;
import org.apache.iotdb.confignode.consensus.statemachine.PartitionRegionStateMachine;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.consensus.IConsensus;
@@ -33,6 +36,7 @@ import org.apache.iotdb.consensus.common.Peer;
import org.apache.iotdb.consensus.common.response.ConsensusReadResponse;
import org.apache.iotdb.consensus.common.response.ConsensusWriteResponse;
import org.apache.iotdb.consensus.config.ConsensusConfig;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,7 +44,6 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
-import java.util.concurrent.TimeUnit;
/** ConsensusManager maintains consensus class, request will redirect to consensus layer */
public class ConsensusManager {
@@ -65,7 +68,7 @@ public class ConsensusManager {
long maxWaitTime = 1000 * 60; // milliseconds, which is 60s
try {
while (!consensusImpl.isLeader(consensusGroupId)) {
- TimeUnit.MILLISECONDS.sleep(100);
+ Thread.sleep(100);
long elapsed = System.currentTimeMillis() - startTime;
if (elapsed > maxWaitTime) {
return;
@@ -78,8 +81,8 @@ public class ConsensusManager {
/** Build ConfigNodeGroup ConsensusLayer */
private void setConsensusLayer(PartitionRegionStateMachine stateMachine) throws IOException {
- consensusGroupId =
- ConsensusGroupId.Factory.createFromTConsensusGroupId(conf.getPartitionRegionId());
+ // There is only one ConfigNodeGroup
+ consensusGroupId = new PartitionRegionId(conf.getPartitionRegionId());
// Consensus local implement
consensusImpl =
@@ -105,22 +108,37 @@ public class ConsensusManager {
peerList.add(new Peer(consensusGroupId, configNodeLocation.getConsensusEndPoint()));
}
consensusImpl.addConsensusGroup(consensusGroupId, peerList);
+
+ // Apply ConfigNode if necessary
+ if (conf.isNeedApply()) {
+ TSStatus status =
+ SyncConfigNodeClientPool.getInstance()
+ .applyConfigNode(
+ conf.getTargetConfigNode(),
+ new TConfigNodeLocation(
+ new TEndPoint(conf.getRpcAddress(), conf.getRpcPort()),
+ new TEndPoint(conf.getRpcAddress(), conf.getConsensusPort())));
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ LOGGER.error(status.getMessage());
+ throw new IOException("Apply ConfigNode failed:");
+ }
+ }
}
/**
* Apply new ConfigNode Peer into PartitionRegion
*
- * @param registerConfigNodeReq ApplyConfigNodeReq
+ * @param applyConfigNodeReq ApplyConfigNodeReq
* @return True if successfully addPeer. False if another ConfigNode is being added to the
* PartitionRegion
*/
- public boolean addConfigNodePeer(RegisterConfigNodeReq registerConfigNodeReq) {
+ public boolean addConfigNodePeer(ApplyConfigNodeReq applyConfigNodeReq) {
return consensusImpl
.addPeer(
consensusGroupId,
new Peer(
consensusGroupId,
- registerConfigNodeReq.getConfigNodeLocation().getConsensusEndPoint()))
+ applyConfigNodeReq.getConfigNodeLocation().getConsensusEndPoint()))
.isSuccess();
}
@@ -137,4 +155,8 @@ public class ConsensusManager {
public boolean isLeader() {
return consensusImpl.isLeader(consensusGroupId);
}
+
+ public ConsensusGroupId getConsensusGroupId() {
+ return consensusGroupId;
+ }
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/Manager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/Manager.java
index 50a931bb59..df6b1593d5 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/Manager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/Manager.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.confignode.consensus.request.read.GetDataNodeInfoReq;
import org.apache.iotdb.confignode.consensus.request.read.GetDataPartitionReq;
import org.apache.iotdb.confignode.consensus.request.read.GetOrCreateDataPartitionReq;
import org.apache.iotdb.confignode.consensus.request.read.GetStorageGroupReq;
+import org.apache.iotdb.confignode.consensus.request.write.ApplyConfigNodeReq;
import org.apache.iotdb.confignode.consensus.request.write.RegisterDataNodeReq;
import org.apache.iotdb.confignode.consensus.request.write.SetDataReplicationFactorReq;
import org.apache.iotdb.confignode.consensus.request.write.SetSchemaReplicationFactorReq;
@@ -209,6 +210,13 @@ public interface Manager {
*/
TConfigNodeRegisterResp registerConfigNode(TConfigNodeRegisterReq req);
+ /**
+ * Apply ConfigNode when it is first startup
+ *
+ * @return status
+ */
+ TSStatus applyConfigNode(ApplyConfigNodeReq applyConfigNodeReq);
+
TSStatus createFunction(String udfName, String className, List<String> uris);
TSStatus dropFunction(String udfName);
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/NodeManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/NodeManager.java
index 813da92e2e..e4cfb0c44e 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/NodeManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/NodeManager.java
@@ -20,13 +20,12 @@ package org.apache.iotdb.confignode.manager;
import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TDataNodeInfo;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
-import org.apache.iotdb.commons.conf.CommonDescriptor;
-import org.apache.iotdb.confignode.client.AsyncConfigNodeToDataNodeClientPool;
-import org.apache.iotdb.confignode.conf.ConfigNodeConf;
+import org.apache.iotdb.confignode.client.AsyncDataNodeClientPool;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
import org.apache.iotdb.confignode.consensus.request.read.GetDataNodeInfoReq;
-import org.apache.iotdb.confignode.consensus.request.write.RegisterConfigNodeReq;
+import org.apache.iotdb.confignode.consensus.request.write.ApplyConfigNodeReq;
import org.apache.iotdb.confignode.consensus.request.write.RegisterDataNodeReq;
import org.apache.iotdb.confignode.consensus.response.DataNodeConfigurationResp;
import org.apache.iotdb.confignode.consensus.response.DataNodeInfosResp;
@@ -42,7 +41,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
-import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.CopyOnWriteArrayList;
/** NodeManager manages cluster node addition and removal requests */
public class NodeManager {
@@ -50,15 +49,14 @@ public class NodeManager {
private static final Logger LOGGER = LoggerFactory.getLogger(NodeManager.class);
private final Manager configManager;
-
private final NodeInfo nodeInfo;
- private final ReentrantLock registerConfigNodeLock;
+ /** TODO:do some operate after add node or remove node */
+ private final List<ChangeServerListener> listeners = new CopyOnWriteArrayList<>();
public NodeManager(Manager configManager, NodeInfo nodeInfo) {
this.configManager = configManager;
this.nodeInfo = nodeInfo;
- this.registerConfigNodeLock = new ReentrantLock();
}
private void setGlobalConfig(DataNodeConfigurationResp dataSet) {
@@ -89,7 +87,7 @@ public class NodeManager {
if (nodeInfo.isOnlineDataNode(req.getInfo().getLocation())) {
// Reset client
- AsyncConfigNodeToDataNodeClientPool.getInstance()
+ AsyncDataNodeClientPool.getInstance()
.resetClient(req.getInfo().getLocation().getInternalEndPoint());
TSStatus status = new TSStatus(TSStatusCode.DATANODE_ALREADY_REGISTERED.getStatusCode());
@@ -155,127 +153,97 @@ public class NodeManager {
* @return TConfigNodeRegisterResp with PartitionRegionId and online ConfigNodes
*/
public TConfigNodeRegisterResp registerConfigNode(TConfigNodeRegisterReq req) {
- // Check system configuration
- TSStatus status = checkSystemConfig(req);
- if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- return new TConfigNodeRegisterResp(status, getOnlineConfigNodes());
- }
+ TConfigNodeRegisterResp resp = new TConfigNodeRegisterResp();
+
+ resp.setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+
+ // Return PartitionRegionId
+ resp.setPartitionRegionId(
+ getConsensusManager().getConsensusGroupId().convertToTConsensusGroupId());
- // If there is any error in the following process,
- // the new ConfigNode must retry and the online ConfigNodes should remain the same
- TConfigNodeRegisterResp errorResp =
- new TConfigNodeRegisterResp(
- new TSStatus(TSStatusCode.NEED_RETRY.getStatusCode()), getOnlineConfigNodes());
- if (registerConfigNodeLock.tryLock()) {
- // We use a ReentrantLock here so that the register ConfigNode process is serialized,
- // and the concurrent register request will be rejected directly.
- try {
- // Skip register process if it's done before
- if (!getOnlineConfigNodes().contains(req.getConfigNodeLocation())) {
- // Do register
- RegisterConfigNodeReq registerConfigNodeReq =
- new RegisterConfigNodeReq(req.getConfigNodeLocation());
- if (!getConsensusManager().addConfigNodePeer(registerConfigNodeReq)) {
- LOGGER.error("Error when addPeer: {}", registerConfigNodeReq.getConfigNodeLocation());
- return errorResp;
- }
- if (getConsensusManager().write(registerConfigNodeReq).getStatus().getCode()
- != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- LOGGER.error(
- "Error when cache ConfigNode: {}", registerConfigNodeReq.getConfigNodeLocation());
- return errorResp;
- }
- }
-
- // Return SUCCESS_STATUS and new configNodeList
- return new TConfigNodeRegisterResp(
- new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()), getOnlineConfigNodes());
- } finally {
- registerConfigNodeLock.unlock();
- }
+ // Return online ConfigNodes
+ resp.setConfigNodeList(nodeInfo.getOnlineConfigNodes());
+ resp.getConfigNodeList().add(req.getConfigNodeLocation());
+
+ return resp;
+ }
+
+ public TSStatus applyConfigNode(ApplyConfigNodeReq applyConfigNodeReq) {
+ if (getConsensusManager().addConfigNodePeer(applyConfigNodeReq)) {
+ return getConsensusManager().write(applyConfigNodeReq).getStatus();
} else {
- return errorResp;
+ return new TSStatus(TSStatusCode.APPLY_CONFIGNODE_FAILED.getStatusCode())
+ .setMessage("Apply ConfigNode failed because there is another ConfigNode being applied.");
}
}
- /**
- * Ensure the system parameters are consistent.
- *
- * @return SUCCESS_STATUS when consistent, INCONSISTENT_SYSTEM_CONFIG otherwise.
- */
- private TSStatus checkSystemConfig(TConfigNodeRegisterReq req) {
- ConfigNodeConf conf = ConfigNodeDescriptor.getInstance().getConf();
- TSStatus errorStatus = new TSStatus(TSStatusCode.INCONSISTENT_SYSTEM_CONFIG.getStatusCode());
-
- if (!req.getPartitionRegionId().equals(conf.getPartitionRegionId())) {
- errorStatus.setMessage(
- "Reject register, please ensure that the cluster_id " + "are consistent.");
- return errorStatus;
- }
+ public List<TConfigNodeLocation> getOnlineConfigNodes() {
+ return nodeInfo.getOnlineConfigNodes();
+ }
- if (!req.getGlobalConfig()
- .getDataRegionConsensusProtocolClass()
- .equals(conf.getDataRegionConsensusProtocolClass())) {
- errorStatus.setMessage(
- "Reject register, please ensure that the data_region_consensus_protocol_class "
- + "are consistent.");
- return errorStatus;
- }
+ private ConsensusManager getConsensusManager() {
+ return configManager.getConsensusManager();
+ }
- if (!req.getGlobalConfig()
- .getSchemaRegionConsensusProtocolClass()
- .equals(conf.getSchemaRegionConsensusProtocolClass())) {
- errorStatus.setMessage(
- "Reject register, please ensure that the schema_region_consensus_protocol_class "
- + "are consistent.");
- return errorStatus;
- }
+ public void registerListener(final ChangeServerListener serverListener) {
+ listeners.add(serverListener);
+ }
- if (req.getGlobalConfig().getSeriesPartitionSlotNum() != conf.getSeriesPartitionSlotNum()) {
- errorStatus.setMessage(
- "Reject register, please ensure that the series_partition_slot_num are consistent.");
- return errorStatus;
- }
+ public boolean unregisterListener(final ChangeServerListener serverListener) {
+ return listeners.remove(serverListener);
+ }
- if (!req.getGlobalConfig()
- .getSeriesPartitionExecutorClass()
- .equals(conf.getSeriesPartitionExecutorClass())) {
- errorStatus.setMessage(
- "Reject register, please ensure that the series_partition_executor_class are consistent.");
- return errorStatus;
- }
+ /** TODO: wait data node register, wait */
+ public void waitForDataNodes() {
+ listeners.stream().forEach(serverListener -> serverListener.waiting());
+ }
- if (req.getGlobalConfig().getTimePartitionInterval() != conf.getTimePartitionInterval()) {
- errorStatus.setMessage(
- "Reject register, please ensure that the time_partition_interval are consistent.");
- return errorStatus;
- }
+ private class ServerStartListenerThread extends Thread implements ChangeServerListener {
+ private boolean changed = false;
- if (req.getDefaultTTL() != CommonDescriptor.getInstance().getConfig().getDefaultTTL()) {
- errorStatus.setMessage("Reject register, please ensure that the default_ttl are consistent.");
- return errorStatus;
+ ServerStartListenerThread() {
+ setDaemon(true);
}
- if (req.getSchemaReplicationFactor() != conf.getSchemaReplicationFactor()) {
- errorStatus.setMessage(
- "Reject register, please ensure that the schema_replication_factor are consistent.");
- return errorStatus;
+ @Override
+ public void addDataNode(TDataNodeLocation DataNodeInfo) {
+ serverChanged();
}
- if (req.getDataReplicationFactor() != conf.getDataReplicationFactor()) {
- errorStatus.setMessage(
- "Reject register, please ensure that the data_replication_factor are consistent.");
- return errorStatus;
+ @Override
+ public void removeDataNode(TDataNodeLocation dataNodeInfo) {
+ serverChanged();
}
- return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
- }
+ private synchronized void serverChanged() {
+ changed = true;
+ this.notify();
+ }
- public List<TConfigNodeLocation> getOnlineConfigNodes() {
- return nodeInfo.getOnlineConfigNodes();
+ @Override
+ public void run() {
+ while (!configManager.isStopped()) {}
+ }
}
- private ConsensusManager getConsensusManager() {
- return configManager.getConsensusManager();
+ /** TODO: For listener for add or remove data node */
+ public interface ChangeServerListener {
+
+ /** Started waiting on DataNode to check */
+ default void waiting() {};
+
+ /**
+ * The server has joined the cluster
+ *
+ * @param dataNodeInfo datanode info
+ */
+ void addDataNode(final TDataNodeLocation dataNodeInfo);
+
+ /**
+ * remove data node
+ *
+ * @param dataNodeInfo data node info
+ */
+ void removeDataNode(final TDataNodeLocation dataNodeInfo);
}
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/PartitionManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/PartitionManager.java
index 785f8df6b6..767d186693 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/PartitionManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/PartitionManager.java
@@ -29,7 +29,7 @@ import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
import org.apache.iotdb.commons.partition.DataPartitionTable;
import org.apache.iotdb.commons.partition.SchemaPartitionTable;
import org.apache.iotdb.commons.partition.executor.SeriesPartitionExecutor;
-import org.apache.iotdb.confignode.client.SyncConfigNodeToDataNodeClientPool;
+import org.apache.iotdb.confignode.client.SyncDataNodeClientPool;
import org.apache.iotdb.confignode.conf.ConfigNodeConf;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
import org.apache.iotdb.confignode.consensus.request.read.GetDataPartitionReq;
@@ -443,7 +443,7 @@ public class PartitionManager {
deletedRegionSet.stream()
.map(TRegionReplicaSet::getRegionId)
.collect(Collectors.toList()));
- SyncConfigNodeToDataNodeClientPool.getInstance().deleteRegions(deletedRegionSet);
+ SyncDataNodeClientPool.getInstance().deleteRegions(deletedRegionSet);
}
}
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/PermissionManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/PermissionManager.java
index aaa708e7fe..0d682a570f 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/PermissionManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/PermissionManager.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.confignode.manager;
import org.apache.iotdb.common.rpc.thrift.TDataNodeInfo;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
-import org.apache.iotdb.confignode.client.SyncConfigNodeToDataNodeClientPool;
+import org.apache.iotdb.confignode.client.SyncDataNodeClientPool;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
import org.apache.iotdb.confignode.consensus.request.auth.AuthorReq;
import org.apache.iotdb.confignode.consensus.response.PermissionInfoResp;
@@ -105,7 +105,7 @@ public class PermissionManager {
req.setRoleName(roleName);
for (TDataNodeInfo dataNodeInfo : allDataNodes) {
status =
- SyncConfigNodeToDataNodeClientPool.getInstance()
+ SyncDataNodeClientPool.getInstance()
.invalidatePermissionCache(dataNodeInfo.getLocation().getInternalEndPoint(), req);
if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
return status;
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/UDFManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/UDFManager.java
index 7c0d9168e0..f1e861cad2 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/UDFManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/UDFManager.java
@@ -22,7 +22,7 @@ package org.apache.iotdb.confignode.manager;
import org.apache.iotdb.common.rpc.thrift.TDataNodeInfo;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
-import org.apache.iotdb.confignode.client.AsyncConfigNodeToDataNodeClientPool;
+import org.apache.iotdb.confignode.client.AsyncDataNodeClientPool;
import org.apache.iotdb.confignode.client.handlers.FunctionManagementHandler;
import org.apache.iotdb.confignode.consensus.request.write.CreateFunctionReq;
import org.apache.iotdb.confignode.consensus.request.write.DropFunctionReq;
@@ -89,7 +89,7 @@ public class UDFManager {
for (TDataNodeInfo dataNodeInfo : onlineDataNodes) {
final TEndPoint endPoint = dataNodeInfo.getLocation().getInternalEndPoint();
- AsyncConfigNodeToDataNodeClientPool.getInstance()
+ AsyncDataNodeClientPool.getInstance()
.createFunction(
endPoint,
request,
@@ -133,7 +133,7 @@ public class UDFManager {
for (TDataNodeInfo dataNodeInfo : onlineDataNodes) {
final TEndPoint endPoint = dataNodeInfo.getLocation().getInternalEndPoint();
- AsyncConfigNodeToDataNodeClientPool.getInstance()
+ AsyncDataNodeClientPool.getInstance()
.dropFunction(
endPoint,
request,
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java
index df83aa7361..857c188bcc 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java
@@ -27,7 +27,7 @@ import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.partition.DataPartitionTable;
import org.apache.iotdb.commons.partition.SchemaPartitionTable;
-import org.apache.iotdb.confignode.client.AsyncConfigNodeToDataNodeClientPool;
+import org.apache.iotdb.confignode.client.AsyncDataNodeClientPool;
import org.apache.iotdb.confignode.client.handlers.HeartbeatHandler;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
import org.apache.iotdb.confignode.consensus.request.write.CreateRegionsReq;
@@ -108,7 +108,7 @@ public class LoadManager implements Runnable {
storageGroup,
getClusterSchemaManager().getStorageGroupSchemaByName(storageGroup).getTTL());
}
- AsyncConfigNodeToDataNodeClientPool.getInstance().createRegions(createRegionsReq, ttlMap);
+ AsyncDataNodeClientPool.getInstance().createRegions(createRegionsReq, ttlMap);
}
/**
@@ -155,7 +155,7 @@ public class LoadManager implements Runnable {
for (TDataNodeInfo dataNodeInfo : onlineDataNodes) {
HeartbeatHandler handler =
new HeartbeatHandler(dataNodeInfo.getLocation(), heartbeatCache);
- AsyncConfigNodeToDataNodeClientPool.getInstance()
+ AsyncDataNodeClientPool.getInstance()
.getHeartBeat(
dataNodeInfo.getLocation().getInternalEndPoint(), genHeartbeatReq(), handler);
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/NodeInfo.java b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/NodeInfo.java
index 1caa6df824..bb8d20966b 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/NodeInfo.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/NodeInfo.java
@@ -28,7 +28,7 @@ import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.confignode.conf.ConfigNodeConstant;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
import org.apache.iotdb.confignode.consensus.request.read.GetDataNodeInfoReq;
-import org.apache.iotdb.confignode.consensus.request.write.RegisterConfigNodeReq;
+import org.apache.iotdb.confignode.consensus.request.write.ApplyConfigNodeReq;
import org.apache.iotdb.confignode.consensus.request.write.RegisterDataNodeReq;
import org.apache.iotdb.confignode.consensus.response.DataNodeInfosResp;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -235,23 +235,23 @@ public class NodeInfo implements SnapshotProcessor {
/**
* Update ConfigNodeList both in memory and confignode-system.properties file
*
- * @param registerConfigNodeReq RegisterConfigNodeReq
+ * @param applyConfigNodeReq ApplyConfigNodeReq
* @return APPLY_CONFIGNODE_FAILED if update online ConfigNode failed.
*/
- public TSStatus registerConfigNode(RegisterConfigNodeReq registerConfigNodeReq) {
+ public TSStatus updateConfigNodeList(ApplyConfigNodeReq applyConfigNodeReq) {
TSStatus status = new TSStatus();
configNodeInfoReadWriteLock.writeLock().lock();
try {
- onlineConfigNodes.add(registerConfigNodeReq.getConfigNodeLocation());
+ onlineConfigNodes.add(applyConfigNodeReq.getConfigNodeLocation());
storeConfigNode();
LOGGER.info(
"Successfully apply ConfigNode: {}. Current ConfigNodeGroup: {}",
- registerConfigNodeReq.getConfigNodeLocation(),
+ applyConfigNodeReq.getConfigNodeLocation(),
onlineConfigNodes);
status.setCode(TSStatusCode.SUCCESS_STATUS.getStatusCode());
} catch (IOException e) {
LOGGER.error("Update online ConfigNode failed.", e);
- status.setCode(TSStatusCode.REGISTER_CONFIGNODE_FAILED.getStatusCode());
+ status.setCode(TSStatusCode.APPLY_CONFIGNODE_FAILED.getStatusCode());
status.setMessage(
"Apply new ConfigNode failed because current ConfigNode can't store ConfigNode information.");
} finally {
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigRequestExecutor.java b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigRequestExecutor.java
index e251b879b8..f990370a80 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigRequestExecutor.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigRequestExecutor.java
@@ -30,6 +30,7 @@ import org.apache.iotdb.confignode.consensus.request.read.GetDataPartitionReq;
import org.apache.iotdb.confignode.consensus.request.read.GetNodePathsPartitionReq;
import org.apache.iotdb.confignode.consensus.request.read.GetSchemaPartitionReq;
import org.apache.iotdb.confignode.consensus.request.read.GetStorageGroupReq;
+import org.apache.iotdb.confignode.consensus.request.write.ApplyConfigNodeReq;
import org.apache.iotdb.confignode.consensus.request.write.CreateDataPartitionReq;
import org.apache.iotdb.confignode.consensus.request.write.CreateFunctionReq;
import org.apache.iotdb.confignode.consensus.request.write.CreateRegionsReq;
@@ -38,7 +39,6 @@ import org.apache.iotdb.confignode.consensus.request.write.DeleteProcedureReq;
import org.apache.iotdb.confignode.consensus.request.write.DeleteStorageGroupReq;
import org.apache.iotdb.confignode.consensus.request.write.DropFunctionReq;
import org.apache.iotdb.confignode.consensus.request.write.PreDeleteStorageGroupReq;
-import org.apache.iotdb.confignode.consensus.request.write.RegisterConfigNodeReq;
import org.apache.iotdb.confignode.consensus.request.write.RegisterDataNodeReq;
import org.apache.iotdb.confignode.consensus.request.write.SetDataReplicationFactorReq;
import org.apache.iotdb.confignode.consensus.request.write.SetSchemaReplicationFactorReq;
@@ -181,8 +181,8 @@ public class ConfigRequestExecutor {
case RevokeRoleFromUser:
case UpdateUser:
return authorInfo.authorNonQuery((AuthorReq) req);
- case RegisterConfigNode:
- return nodeInfo.registerConfigNode((RegisterConfigNodeReq) req);
+ case ApplyConfigNode:
+ return nodeInfo.updateConfigNodeList((ApplyConfigNodeReq) req);
case CreateFunction:
return udfInfo.createFunction((CreateFunctionReq) req);
case DropFunction:
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
index db4a6e4736..bcddd90aa4 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.confignode.procedure.env;
import org.apache.iotdb.common.rpc.thrift.TDataNodeInfo;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
-import org.apache.iotdb.confignode.client.SyncConfigNodeToDataNodeClientPool;
+import org.apache.iotdb.confignode.client.SyncDataNodeClientPool;
import org.apache.iotdb.confignode.consensus.request.write.DeleteStorageGroupReq;
import org.apache.iotdb.confignode.consensus.request.write.PreDeleteStorageGroupReq;
import org.apache.iotdb.confignode.manager.ConfigManager;
@@ -100,11 +100,11 @@ public class ConfigNodeProcedureEnv {
invalidateCacheReq.setFullPath(storageGroupName);
for (TDataNodeInfo dataNodeInfo : allDataNodes) {
final TSStatus invalidateSchemaStatus =
- SyncConfigNodeToDataNodeClientPool.getInstance()
+ SyncDataNodeClientPool.getInstance()
.invalidateSchemaCache(
dataNodeInfo.getLocation().getInternalEndPoint(), invalidateCacheReq);
final TSStatus invalidatePartitionStatus =
- SyncConfigNodeToDataNodeClientPool.getInstance()
+ SyncDataNodeClientPool.getInstance()
.invalidatePartitionCache(
dataNodeInfo.getLocation().getInternalEndPoint(), invalidateCacheReq);
if (!verifySucceed(invalidatePartitionStatus, invalidateSchemaStatus)) {
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/service/ConfigNode.java b/confignode/src/main/java/org/apache/iotdb/confignode/service/ConfigNode.java
index 62a5e04fbc..753d45fac7 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/service/ConfigNode.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/service/ConfigNode.java
@@ -116,14 +116,8 @@ public class ConfigNode implements ConfigNodeMBean {
return;
}
- ConfigNodeConf conf = ConfigNodeDescriptor.getInstance().getConf();
- if (conf.getRpcAddress().equals(conf.getTargetConfigNode().getIp())
- && conf.getRpcPort() == conf.getTargetConfigNode().getPort()) {
- LOGGER.info("{} has successfully started as a seed", ConfigNodeConstant.GLOBAL_NAME);
- } else {
- LOGGER.info(
- "{} has successfully started and joined the cluster.", ConfigNodeConstant.GLOBAL_NAME);
- }
+ LOGGER.info(
+ "{} has successfully started and joined the cluster.", ConfigNodeConstant.GLOBAL_NAME);
}
public void deactivate() throws IOException {
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java b/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
index 2448d86b02..aabffbed77 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
@@ -27,7 +27,7 @@ import org.apache.iotdb.commons.auth.AuthException;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.utils.TestOnly;
-import org.apache.iotdb.confignode.client.SyncConfigNodeToDataNodeClientPool;
+import org.apache.iotdb.confignode.client.SyncDataNodeClientPool;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
import org.apache.iotdb.confignode.consensus.request.auth.AuthorReq;
@@ -36,6 +36,7 @@ import org.apache.iotdb.confignode.consensus.request.read.GetDataNodeInfoReq;
import org.apache.iotdb.confignode.consensus.request.read.GetDataPartitionReq;
import org.apache.iotdb.confignode.consensus.request.read.GetOrCreateDataPartitionReq;
import org.apache.iotdb.confignode.consensus.request.read.GetStorageGroupReq;
+import org.apache.iotdb.confignode.consensus.request.write.ApplyConfigNodeReq;
import org.apache.iotdb.confignode.consensus.request.write.RegisterDataNodeReq;
import org.apache.iotdb.confignode.consensus.request.write.SetDataReplicationFactorReq;
import org.apache.iotdb.confignode.consensus.request.write.SetSchemaReplicationFactorReq;
@@ -388,6 +389,17 @@ public class ConfigNodeRPCServiceProcessor implements ConfigIService.Iface {
return resp;
}
+ @Override
+ public TSStatus applyConfigNode(TConfigNodeLocation configNodeLocation) throws TException {
+ ApplyConfigNodeReq applyConfigNodeReq = new ApplyConfigNodeReq(configNodeLocation);
+ TSStatus status = configManager.applyConfigNode(applyConfigNodeReq);
+
+ // Print log to record the ConfigNode that performs the ApplyConfigNodeRequest
+ LOGGER.info("Execute ApplyConfigNodeRequest {} with result {}", configNodeLocation, status);
+
+ return status;
+ }
+
@Override
public TSStatus createFunction(TCreateFunctionReq req) {
return configManager.createFunction(req.getUdfName(), req.getClassName(), req.getUris());
@@ -417,7 +429,7 @@ public class ConfigNodeRPCServiceProcessor implements ConfigIService.Iface {
TSStatus tsStatus = new TSStatus();
for (TDataNodeInfo dataNodeInfo : onlineDataNodes) {
tsStatus =
- SyncConfigNodeToDataNodeClientPool.getInstance()
+ SyncDataNodeClientPool.getInstance()
.flush(dataNodeInfo.getLocation().getInternalEndPoint(), req);
if (tsStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
return tsStatus;
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigRequestSerDeTest.java b/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigRequestSerDeTest.java
index 65ace14b4a..5d327923a6 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigRequestSerDeTest.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigRequestSerDeTest.java
@@ -40,13 +40,13 @@ import org.apache.iotdb.confignode.consensus.request.read.GetOrCreateDataPartiti
import org.apache.iotdb.confignode.consensus.request.read.GetOrCreateSchemaPartitionReq;
import org.apache.iotdb.confignode.consensus.request.read.GetSchemaPartitionReq;
import org.apache.iotdb.confignode.consensus.request.read.GetStorageGroupReq;
+import org.apache.iotdb.confignode.consensus.request.write.ApplyConfigNodeReq;
import org.apache.iotdb.confignode.consensus.request.write.CreateDataPartitionReq;
import org.apache.iotdb.confignode.consensus.request.write.CreateRegionsReq;
import org.apache.iotdb.confignode.consensus.request.write.CreateSchemaPartitionReq;
import org.apache.iotdb.confignode.consensus.request.write.DeleteProcedureReq;
import org.apache.iotdb.confignode.consensus.request.write.DeleteRegionsReq;
import org.apache.iotdb.confignode.consensus.request.write.DeleteStorageGroupReq;
-import org.apache.iotdb.confignode.consensus.request.write.RegisterConfigNodeReq;
import org.apache.iotdb.confignode.consensus.request.write.RegisterDataNodeReq;
import org.apache.iotdb.confignode.consensus.request.write.SetDataReplicationFactorReq;
import org.apache.iotdb.confignode.consensus.request.write.SetSchemaReplicationFactorReq;
@@ -472,12 +472,12 @@ public class ConfigRequestSerDeTest {
@Test
public void registerConfigNodeReqTest() throws IOException {
- RegisterConfigNodeReq req0 =
- new RegisterConfigNodeReq(
+ ApplyConfigNodeReq req0 =
+ new ApplyConfigNodeReq(
new TConfigNodeLocation(
new TEndPoint("0.0.0.0", 22277), new TEndPoint("0.0.0.0", 22278)));
- RegisterConfigNodeReq req1 =
- (RegisterConfigNodeReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
+ ApplyConfigNodeReq req1 =
+ (ApplyConfigNodeReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
}
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/persistence/NodeInfoTest.java b/confignode/src/test/java/org/apache/iotdb/confignode/persistence/NodeInfoTest.java
index 5758c5b942..d03cc6d5e7 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/persistence/NodeInfoTest.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/persistence/NodeInfoTest.java
@@ -21,7 +21,6 @@ package org.apache.iotdb.confignode.persistence;
import org.apache.iotdb.common.rpc.thrift.TDataNodeInfo;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
-import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
import org.apache.iotdb.confignode.consensus.request.write.RegisterDataNodeReq;
import org.apache.commons.io.FileUtils;
@@ -33,7 +32,6 @@ import org.junit.Test;
import java.io.File;
import java.io.IOException;
-import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -47,7 +45,6 @@ public class NodeInfoTest {
@BeforeClass
public static void setup() {
- ConfigNodeDescriptor.getInstance().getConf().setConfigNodeList(new ArrayList<>());
nodeInfo = new NodeInfo();
if (!snapshotDir.exists()) {
snapshotDir.mkdirs();
@@ -74,7 +71,7 @@ public class NodeInfoTest {
nodeInfo.registerDataNode(registerDataNodeReq);
Set<TDataNodeLocation> drainingDataNodes_before = new HashSet<>();
- // parameter i is used as flag in generateTDataNodeLocation
+ // parameter i is used to be flag in generateTDataNodeLocation
for (int i = 3; i < 8; i++) {
drainingDataNodes_before.add(generateTDataNodeLocation(i));
}
diff --git a/server/src/main/java/org/apache/iotdb/db/auth/AuthorizerManager.java b/server/src/main/java/org/apache/iotdb/db/auth/AuthorizerManager.java
index 567af977be..5d42da83a5 100644
--- a/server/src/main/java/org/apache/iotdb/db/auth/AuthorizerManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/auth/AuthorizerManager.java
@@ -26,7 +26,7 @@ import org.apache.iotdb.commons.auth.authorizer.IAuthorizer;
import org.apache.iotdb.commons.auth.entity.Role;
import org.apache.iotdb.commons.auth.entity.User;
import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerReq;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.mpp.common.header.ColumnHeader;
import org.apache.iotdb.db.mpp.common.header.DatasetHeader;
@@ -389,21 +389,20 @@ public class AuthorizerManager implements IAuthorizer {
}
public SettableFuture<ConfigTaskResult> queryPermission(
- TAuthorizerReq authorizerReq, DataNodeToConfigNodeClient dataNodeToConfigNodeClient)
- throws TException {
+ TAuthorizerReq authorizerReq, ConfigNodeClient configNodeClient) throws TException {
authReadWriteLock.readLock().lock();
try {
- return authorityFetcher.queryPermission(authorizerReq, dataNodeToConfigNodeClient);
+ return authorityFetcher.queryPermission(authorizerReq, configNodeClient);
} finally {
authReadWriteLock.readLock().unlock();
}
}
public SettableFuture<ConfigTaskResult> operatePermission(
- TAuthorizerReq authorizerReq, DataNodeToConfigNodeClient dataNodeToConfigNodeClient) {
+ TAuthorizerReq authorizerReq, ConfigNodeClient configNodeClient) {
authReadWriteLock.writeLock().lock();
try {
- return authorityFetcher.operatePermission(authorizerReq, dataNodeToConfigNodeClient);
+ return authorityFetcher.operatePermission(authorizerReq, configNodeClient);
} finally {
authReadWriteLock.writeLock().unlock();
}
diff --git a/server/src/main/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcher.java b/server/src/main/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcher.java
index 83d2c46b49..0da77fde91 100644
--- a/server/src/main/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcher.java
@@ -32,9 +32,9 @@ import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerResp;
import org.apache.iotdb.confignode.rpc.thrift.TCheckUserPrivilegesReq;
import org.apache.iotdb.confignode.rpc.thrift.TLoginReq;
import org.apache.iotdb.confignode.rpc.thrift.TPermissionInfoResp;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.client.ConfigNodeInfo;
import org.apache.iotdb.db.client.DataNodeClientPoolFactory;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
import org.apache.iotdb.db.qp.logical.sys.AuthorOperator;
@@ -75,10 +75,9 @@ public class ClusterAuthorityFetcher implements IAuthorityFetcher {
.expireAfterAccess(conf.getConfig().getAuthorCacheExpireTime(), TimeUnit.MINUTES)
.build();
- private static final IClientManager<PartitionRegionId, DataNodeToConfigNodeClient>
- configNodeClientManager =
- new IClientManager.Factory<PartitionRegionId, DataNodeToConfigNodeClient>()
- .createClientManager(new DataNodeClientPoolFactory.ConfigNodeClientPoolFactory());
+ private static final IClientManager<PartitionRegionId, ConfigNodeClient> configNodeClientManager =
+ new IClientManager.Factory<PartitionRegionId, ConfigNodeClient>()
+ .createClientManager(new DataNodeClientPoolFactory.ConfigNodeClientPoolFactory());
private static final class ClusterAuthorityFetcherHolder {
private static final ClusterAuthorityFetcher INSTANCE = new ClusterAuthorityFetcher();
@@ -131,11 +130,11 @@ public class ClusterAuthorityFetcher implements IAuthorityFetcher {
@Override
public SettableFuture<ConfigTaskResult> operatePermission(
- TAuthorizerReq authorizerReq, DataNodeToConfigNodeClient dataNodeToConfigNodeClient) {
+ TAuthorizerReq authorizerReq, ConfigNodeClient configNodeClient) {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
try {
// Send request to some API server
- TSStatus tsStatus = dataNodeToConfigNodeClient.operatePermission(authorizerReq);
+ TSStatus tsStatus = configNodeClient.operatePermission(authorizerReq);
// Get response or throw exception
if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
logger.error(
@@ -159,12 +158,12 @@ public class ClusterAuthorityFetcher implements IAuthorityFetcher {
@Override
public SettableFuture<ConfigTaskResult> queryPermission(
- TAuthorizerReq authorizerReq, DataNodeToConfigNodeClient dataNodeToConfigNodeClient) {
+ TAuthorizerReq authorizerReq, ConfigNodeClient configNodeClient) {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
TAuthorizerResp authorizerResp = new TAuthorizerResp();
try {
// Send request to some API server
- authorizerResp = dataNodeToConfigNodeClient.queryPermission(authorizerReq);
+ authorizerResp = configNodeClient.queryPermission(authorizerReq);
// Get response or throw exception
if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != authorizerResp.getStatus().getCode()) {
logger.error(
@@ -199,10 +198,10 @@ public class ClusterAuthorityFetcher implements IAuthorityFetcher {
} else {
TLoginReq req = new TLoginReq(username, password);
TPermissionInfoResp status = null;
- try (DataNodeToConfigNodeClient dataNodeToConfigNodeClient =
+ try (ConfigNodeClient configNodeClient =
configNodeClientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
// Send request to some API server
- status = dataNodeToConfigNodeClient.login(req);
+ status = configNodeClient.login(req);
} catch (TException | IOException e) {
logger.error("Failed to connect to config node.");
status = new TPermissionInfoResp();
@@ -226,10 +225,10 @@ public class ClusterAuthorityFetcher implements IAuthorityFetcher {
public TSStatus checkPath(String username, List<String> allPath, int permission) {
TCheckUserPrivilegesReq req = new TCheckUserPrivilegesReq(username, allPath, permission);
TPermissionInfoResp permissionInfoResp;
- try (DataNodeToConfigNodeClient dataNodeToConfigNodeClient =
+ try (ConfigNodeClient configNodeClient =
configNodeClientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
// Send request to some API server
- permissionInfoResp = dataNodeToConfigNodeClient.checkUserPrivileges(req);
+ permissionInfoResp = configNodeClient.checkUserPrivileges(req);
} catch (TException | IOException e) {
logger.error("Failed to connect to config node.");
permissionInfoResp = new TPermissionInfoResp();
diff --git a/server/src/main/java/org/apache/iotdb/db/auth/IAuthorityFetcher.java b/server/src/main/java/org/apache/iotdb/db/auth/IAuthorityFetcher.java
index d9294948d1..fc618d1e1a 100644
--- a/server/src/main/java/org/apache/iotdb/db/auth/IAuthorityFetcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/auth/IAuthorityFetcher.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.db.auth;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerReq;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
import com.google.common.util.concurrent.SettableFuture;
@@ -35,8 +35,8 @@ public interface IAuthorityFetcher {
TSStatus checkUserPrivileges(String username, List<String> allPath, int permission);
SettableFuture<ConfigTaskResult> operatePermission(
- TAuthorizerReq authorizerReq, DataNodeToConfigNodeClient dataNodeToConfigNodeClient);
+ TAuthorizerReq authorizerReq, ConfigNodeClient configNodeClient);
SettableFuture<ConfigTaskResult> queryPermission(
- TAuthorizerReq authorizerReq, DataNodeToConfigNodeClient dataNodeToConfigNodeClient);
+ TAuthorizerReq authorizerReq, ConfigNodeClient configNodeClient);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/auth/StandaloneAuthorityFetcher.java b/server/src/main/java/org/apache/iotdb/db/auth/StandaloneAuthorityFetcher.java
index fa4df8a74f..d12d5c8329 100644
--- a/server/src/main/java/org/apache/iotdb/db/auth/StandaloneAuthorityFetcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/auth/StandaloneAuthorityFetcher.java
@@ -23,7 +23,7 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.auth.AuthException;
import org.apache.iotdb.commons.utils.AuthUtils;
import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerReq;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.localconfignode.LocalConfigNode;
import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
import org.apache.iotdb.rpc.RpcUtils;
@@ -103,7 +103,7 @@ public class StandaloneAuthorityFetcher implements IAuthorityFetcher {
@Override
public SettableFuture<ConfigTaskResult> operatePermission(
- TAuthorizerReq authorizerReq, DataNodeToConfigNodeClient dataNodeToConfigNodeClient) {
+ TAuthorizerReq authorizerReq, ConfigNodeClient configNodeClient) {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
boolean status = true;
try {
@@ -120,7 +120,7 @@ public class StandaloneAuthorityFetcher implements IAuthorityFetcher {
@Override
public SettableFuture<ConfigTaskResult> queryPermission(
- TAuthorizerReq authorizerReq, DataNodeToConfigNodeClient dataNodeToConfigNodeClient) {
+ TAuthorizerReq authorizerReq, ConfigNodeClient configNodeClient) {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
Map<String, List<String>> authorizerResp = new HashMap<>();
try {
diff --git a/server/src/main/java/org/apache/iotdb/db/client/DataNodeToConfigNodeClient.java b/server/src/main/java/org/apache/iotdb/db/client/ConfigNodeClient.java
similarity index 91%
rename from server/src/main/java/org/apache/iotdb/db/client/DataNodeToConfigNodeClient.java
rename to server/src/main/java/org/apache/iotdb/db/client/ConfigNodeClient.java
index 85e78289f6..f7e42bc075 100644
--- a/server/src/main/java/org/apache/iotdb/db/client/DataNodeToConfigNodeClient.java
+++ b/server/src/main/java/org/apache/iotdb/db/client/ConfigNodeClient.java
@@ -78,9 +78,8 @@ import java.lang.reflect.Constructor;
import java.util.ArrayList;
import java.util.List;
-public class DataNodeToConfigNodeClient
- implements ConfigIService.Iface, SyncThriftClient, AutoCloseable {
- private static final Logger logger = LoggerFactory.getLogger(DataNodeToConfigNodeClient.class);
+public class ConfigNodeClient implements ConfigIService.Iface, SyncThriftClient, AutoCloseable {
+ private static final Logger logger = LoggerFactory.getLogger(ConfigNodeClient.class);
private static final int RETRY_NUM = 5;
@@ -99,13 +98,13 @@ public class DataNodeToConfigNodeClient
private int cursor = 0;
- ClientManager<PartitionRegionId, DataNodeToConfigNodeClient> clientManager;
+ ClientManager<PartitionRegionId, ConfigNodeClient> clientManager;
PartitionRegionId partitionRegionId = ConfigNodeInfo.partitionRegionId;
TProtocolFactory protocolFactory;
- public DataNodeToConfigNodeClient() throws TException {
+ public ConfigNodeClient() throws TException {
// Read config nodes from configuration
configNodes = ConfigNodeInfo.getInstance().getLatestConfigNodes();
protocolFactory =
@@ -116,10 +115,10 @@ public class DataNodeToConfigNodeClient
init();
}
- public DataNodeToConfigNodeClient(
+ public ConfigNodeClient(
TProtocolFactory protocolFactory,
long connectionTimeout,
- ClientManager<PartitionRegionId, DataNodeToConfigNodeClient> clientManager)
+ ClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
throws TException {
configNodes = ConfigNodeInfo.getInstance().getLatestConfigNodes();
this.protocolFactory = protocolFactory;
@@ -579,7 +578,34 @@ public class DataNodeToConfigNodeClient
@Override
public TConfigNodeRegisterResp registerConfigNode(TConfigNodeRegisterReq req) throws TException {
- throw new RuntimeException("Cannot call registerConfigNode in DataNode!");
+ for (int i = 0; i < RETRY_NUM; i++) {
+ try {
+ TConfigNodeRegisterResp resp = client.registerConfigNode(req);
+ if (!updateConfigNodeLeader(resp.status)) {
+ return resp;
+ }
+ } catch (TException e) {
+ configLeader = null;
+ }
+ reconnect();
+ }
+ throw new TException(MSG_RECONNECTION_FAIL);
+ }
+
+ @Override
+ public TSStatus applyConfigNode(TConfigNodeLocation configNodeLocation) throws TException {
+ for (int i = 0; i < RETRY_NUM; i++) {
+ try {
+ TSStatus status = client.applyConfigNode(configNodeLocation);
+ if (!updateConfigNodeLeader(status)) {
+ return status;
+ }
+ } catch (TException e) {
+ configLeader = null;
+ }
+ reconnect();
+ }
+ throw new TException(MSG_RECONNECTION_FAIL);
}
@Override
@@ -630,31 +656,29 @@ public class DataNodeToConfigNodeClient
throw new TException(MSG_RECONNECTION_FAIL);
}
- public static class Factory
- extends BaseClientFactory<PartitionRegionId, DataNodeToConfigNodeClient> {
+ public static class Factory extends BaseClientFactory<PartitionRegionId, ConfigNodeClient> {
public Factory(
- ClientManager<PartitionRegionId, DataNodeToConfigNodeClient> clientManager,
+ ClientManager<PartitionRegionId, ConfigNodeClient> clientManager,
ClientFactoryProperty clientFactoryProperty) {
super(clientManager, clientFactoryProperty);
}
@Override
public void destroyObject(
- PartitionRegionId partitionRegionId,
- PooledObject<DataNodeToConfigNodeClient> pooledObject) {
+ PartitionRegionId partitionRegionId, PooledObject<ConfigNodeClient> pooledObject) {
pooledObject.getObject().invalidate();
}
@Override
- public PooledObject<DataNodeToConfigNodeClient> makeObject(PartitionRegionId partitionRegionId)
+ public PooledObject<ConfigNodeClient> makeObject(PartitionRegionId partitionRegionId)
throws Exception {
- Constructor<DataNodeToConfigNodeClient> constructor =
- DataNodeToConfigNodeClient.class.getConstructor(
+ Constructor<ConfigNodeClient> constructor =
+ ConfigNodeClient.class.getConstructor(
TProtocolFactory.class, long.class, clientManager.getClass());
return new DefaultPooledObject<>(
SyncThriftClientWithErrorHandler.newErrorHandler(
- DataNodeToConfigNodeClient.class,
+ ConfigNodeClient.class,
constructor,
clientFactoryProperty.getProtocolFactory(),
clientFactoryProperty.getConnectionTimeoutMs(),
@@ -663,8 +687,7 @@ public class DataNodeToConfigNodeClient
@Override
public boolean validateObject(
- PartitionRegionId partitionRegionId,
- PooledObject<DataNodeToConfigNodeClient> pooledObject) {
+ PartitionRegionId partitionRegionId, PooledObject<ConfigNodeClient> pooledObject) {
return pooledObject.getObject() != null && pooledObject.getObject().getTransport().isOpen();
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/client/DataNodeClientPoolFactory.java b/server/src/main/java/org/apache/iotdb/db/client/DataNodeClientPoolFactory.java
index 7a54f11900..551d021bff 100644
--- a/server/src/main/java/org/apache/iotdb/db/client/DataNodeClientPoolFactory.java
+++ b/server/src/main/java/org/apache/iotdb/db/client/DataNodeClientPoolFactory.java
@@ -148,19 +148,19 @@ public class DataNodeClientPoolFactory {
}
public static class ConfigNodeClientPoolFactory
- implements IClientPoolFactory<PartitionRegionId, DataNodeToConfigNodeClient> {
+ implements IClientPoolFactory<PartitionRegionId, ConfigNodeClient> {
@Override
- public KeyedObjectPool<PartitionRegionId, DataNodeToConfigNodeClient> createClientPool(
- ClientManager<PartitionRegionId, DataNodeToConfigNodeClient> manager) {
+ public KeyedObjectPool<PartitionRegionId, ConfigNodeClient> createClientPool(
+ ClientManager<PartitionRegionId, ConfigNodeClient> manager) {
return new GenericKeyedObjectPool<>(
- new DataNodeToConfigNodeClient.Factory(
+ new ConfigNodeClient.Factory(
manager,
new ClientFactoryProperty.Builder()
.setConnectionTimeoutMs(conf.getConnectionTimeoutInMS())
.setRpcThriftCompressionEnabled(conf.isRpcThriftCompressionEnable())
.setSelectorNumOfAsyncClientManager(conf.getSelectorNumOfClientManager())
.build()),
- new ClientPoolProperty.Builder<DataNodeToConfigNodeClient>().build().getConfig());
+ new ClientPoolProperty.Builder<ConfigNodeClient>().build().getConfig());
}
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/Coordinator.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/Coordinator.java
index b2da85795a..9b79b1a00e 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/Coordinator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/Coordinator.java
@@ -23,8 +23,8 @@ import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.consensus.PartitionRegionId;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.client.DataNodeClientPoolFactory;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
import org.apache.iotdb.db.mpp.common.MPPQueryContext;
import org.apache.iotdb.db.mpp.common.QueryId;
import org.apache.iotdb.db.mpp.common.SessionInfo;
@@ -68,9 +68,9 @@ public class Coordinator {
.createClientManager(
new DataNodeClientPoolFactory.SyncDataNodeInternalServiceClientPoolFactory());
- private static final IClientManager<PartitionRegionId, DataNodeToConfigNodeClient>
+ private static final IClientManager<PartitionRegionId, ConfigNodeClient>
CONFIG_NODE_CLIENT_MANAGER =
- new IClientManager.Factory<PartitionRegionId, DataNodeToConfigNodeClient>()
+ new IClientManager.Factory<PartitionRegionId, ConfigNodeClient>()
.createClientManager(new DataNodeClientPoolFactory.ConfigNodeClientPoolFactory());
private final ExecutorService executor;
private final ExecutorService writeOperationExecutor;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterPartitionFetcher.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterPartitionFetcher.java
index 29e0c6f526..5da5985d80 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterPartitionFetcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterPartitionFetcher.java
@@ -40,9 +40,9 @@ import org.apache.iotdb.confignode.rpc.thrift.TSchemaPartitionResp;
import org.apache.iotdb.confignode.rpc.thrift.TSetStorageGroupReq;
import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchemaResp;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.client.ConfigNodeInfo;
import org.apache.iotdb.db.client.DataNodeClientPoolFactory;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.sql.StatementAnalyzeException;
@@ -80,10 +80,9 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
private PartitionCache partitionCache;
- private final IClientManager<PartitionRegionId, DataNodeToConfigNodeClient>
- configNodeClientManager =
- new IClientManager.Factory<PartitionRegionId, DataNodeToConfigNodeClient>()
- .createClientManager(new DataNodeClientPoolFactory.ConfigNodeClientPoolFactory());
+ private final IClientManager<PartitionRegionId, ConfigNodeClient> configNodeClientManager =
+ new IClientManager.Factory<PartitionRegionId, ConfigNodeClient>()
+ .createClientManager(new DataNodeClientPoolFactory.ConfigNodeClientPoolFactory());
private static final class ClusterPartitionFetcherHolder {
private static final ClusterPartitionFetcher INSTANCE = new ClusterPartitionFetcher();
@@ -106,7 +105,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
@Override
public SchemaPartition getSchemaPartition(PathPatternTree patternTree) {
- try (DataNodeToConfigNodeClient client =
+ try (ConfigNodeClient client =
configNodeClientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
patternTree.constructTree();
List<String> devicePaths = patternTree.findAllDevicePaths();
@@ -130,7 +129,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
@Override
public SchemaPartition getOrCreateSchemaPartition(PathPatternTree patternTree) {
- try (DataNodeToConfigNodeClient client =
+ try (ConfigNodeClient client =
configNodeClientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
patternTree.constructTree();
List<String> devicePaths = patternTree.findAllDevicePaths();
@@ -155,7 +154,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
@Override
public SchemaNodeManagementPartition getSchemaNodeManagementPartitionWithLevel(
PathPatternTree patternTree, Integer level) {
- try (DataNodeToConfigNodeClient client =
+ try (ConfigNodeClient client =
configNodeClientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
patternTree.constructTree();
TSchemaNodeManagementResp schemaNodeManagementResp =
@@ -172,7 +171,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
@Override
public DataPartition getDataPartition(
Map<String, List<DataPartitionQueryParam>> sgNameToQueryParamsMap) {
- try (DataNodeToConfigNodeClient client =
+ try (ConfigNodeClient client =
configNodeClientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
TDataPartitionResp dataPartitionResp =
client.getDataPartition(constructDataPartitionReq(sgNameToQueryParamsMap));
@@ -188,7 +187,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
@Override
public DataPartition getDataPartition(List<DataPartitionQueryParam> dataPartitionQueryParams) {
- try (DataNodeToConfigNodeClient client =
+ try (ConfigNodeClient client =
configNodeClientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
Map<String, List<DataPartitionQueryParam>> splitDataPartitionQueryParams =
splitDataPartitionQueryParam(dataPartitionQueryParams, false);
@@ -213,7 +212,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
public DataPartition getOrCreateDataPartition(
Map<String, List<DataPartitionQueryParam>> sgNameToQueryParamsMap) {
// Do not use data partition cache
- try (DataNodeToConfigNodeClient client =
+ try (ConfigNodeClient client =
configNodeClientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
TDataPartitionResp dataPartitionResp =
client.getOrCreateDataPartition(constructDataPartitionReq(sgNameToQueryParamsMap));
@@ -230,7 +229,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
@Override
public DataPartition getOrCreateDataPartition(
List<DataPartitionQueryParam> dataPartitionQueryParams) {
- try (DataNodeToConfigNodeClient client =
+ try (ConfigNodeClient client =
configNodeClientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
Map<String, List<DataPartitionQueryParam>> splitDataPartitionQueryParams =
splitDataPartitionQueryParam(dataPartitionQueryParams, true);
@@ -275,7 +274,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
boolean firstTryResult = partitionCache.getStorageGroup(devicePaths, deviceToStorageGroup);
if (!firstTryResult) {
List<String> storageGroupPathPattern = ROOT_PATH;
- try (DataNodeToConfigNodeClient client =
+ try (ConfigNodeClient client =
configNodeClientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
TStorageGroupSchemaResp storageGroupSchemaResp =
client.getMatchedStorageGroupSchemas(storageGroupPathPattern);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/AuthorizerTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/AuthorizerTask.java
index 8452503b79..375ffe0600 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/AuthorizerTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/AuthorizerTask.java
@@ -24,8 +24,8 @@ import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.consensus.PartitionRegionId;
import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerReq;
import org.apache.iotdb.db.auth.AuthorizerManager;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.mpp.plan.analyze.QueryType;
@@ -55,7 +55,7 @@ public class AuthorizerTask implements IConfigTask {
@Override
public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, DataNodeToConfigNodeClient> clientManager) {
+ IClientManager<PartitionRegionId, ConfigNodeClient> clientManager) {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
try {
// Construct request using statement
@@ -72,12 +72,12 @@ public class AuthorizerTask implements IConfigTask {
: authorStatement.getNodeName().getFullPath());
// Send request to some API server
if (config.isClusterMode()) {
- try (DataNodeToConfigNodeClient dataNodeToConfigNodeClient =
+ try (ConfigNodeClient configNodeClient =
clientManager.borrowClient(ConfigNodeInfo.partitionRegionId); ) {
if (authorStatement.getQueryType() == QueryType.WRITE) {
- future = authorizerManager.operatePermission(req, dataNodeToConfigNodeClient);
+ future = authorizerManager.operatePermission(req, configNodeClient);
} else {
- future = authorizerManager.queryPermission(req, dataNodeToConfigNodeClient);
+ future = authorizerManager.queryPermission(req, configNodeClient);
}
}
} else {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ConfigExecution.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ConfigExecution.java
index dd760e59b4..b014a11926 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ConfigExecution.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ConfigExecution.java
@@ -22,7 +22,7 @@ package org.apache.iotdb.db.mpp.plan.execution.config;
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.consensus.PartitionRegionId;
import org.apache.iotdb.commons.utils.TestOnly;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.mpp.common.MPPQueryContext;
import org.apache.iotdb.db.mpp.common.header.DatasetHeader;
import org.apache.iotdb.db.mpp.execution.QueryStateMachine;
@@ -57,13 +57,13 @@ public class ConfigExecution implements IQueryExecution {
private boolean resultSetConsumed;
private final IConfigTask task;
- private IClientManager<PartitionRegionId, DataNodeToConfigNodeClient> clientManager;
+ private IClientManager<PartitionRegionId, ConfigNodeClient> clientManager;
public ConfigExecution(
MPPQueryContext context,
Statement statement,
ExecutorService executor,
- IClientManager<PartitionRegionId, DataNodeToConfigNodeClient> clientManager) {
+ IClientManager<PartitionRegionId, ConfigNodeClient> clientManager) {
this.context = context;
this.statement = statement;
this.executor = executor;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/CountStorageGroupTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/CountStorageGroupTask.java
index 0206cfac45..5d8fa105a7 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/CountStorageGroupTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/CountStorageGroupTask.java
@@ -24,8 +24,8 @@ import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.consensus.PartitionRegionId;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.confignode.rpc.thrift.TCountStorageGroupResp;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.localconfignode.LocalConfigNode;
@@ -60,15 +60,14 @@ public class CountStorageGroupTask implements IConfigTask {
@Override
public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, DataNodeToConfigNodeClient> clientManager)
+ IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
throws InterruptedException {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
int storageGroupNum = 0;
if (config.isClusterMode()) {
List<String> storageGroupPathPattern =
Arrays.asList(countStorageGroupStatement.getPartialPath().getNodes());
- try (DataNodeToConfigNodeClient client =
- clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ try (ConfigNodeClient client = clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
TCountStorageGroupResp resp = client.countMatchedStorageGroups(storageGroupPathPattern);
storageGroupNum = resp.getCount();
} catch (TException | IOException e) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/CreateFunctionTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/CreateFunctionTask.java
index 553a9e85a2..56b95836bf 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/CreateFunctionTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/CreateFunctionTask.java
@@ -25,8 +25,8 @@ import org.apache.iotdb.commons.consensus.PartitionRegionId;
import org.apache.iotdb.commons.udf.service.UDFExecutableManager;
import org.apache.iotdb.commons.udf.service.UDFRegistrationService;
import org.apache.iotdb.confignode.rpc.thrift.TCreateFunctionReq;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateFunctionStatement;
@@ -62,7 +62,7 @@ public class CreateFunctionTask implements IConfigTask {
@Override
public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, DataNodeToConfigNodeClient> clientManager)
+ IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
throws InterruptedException {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
if (CONFIG.isClusterMode()) {
@@ -74,10 +74,9 @@ public class CreateFunctionTask implements IConfigTask {
}
private void executeCluster(
- IClientManager<PartitionRegionId, DataNodeToConfigNodeClient> clientManager,
+ IClientManager<PartitionRegionId, ConfigNodeClient> clientManager,
SettableFuture<ConfigTaskResult> future) {
- try (DataNodeToConfigNodeClient client =
- clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ try (ConfigNodeClient client = clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
final TSStatus executionStatus =
client.createFunction(new TCreateFunctionReq(udfName, className, uris));
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/DeleteStorageGroupTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/DeleteStorageGroupTask.java
index 0f719c9410..dceda3cb95 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/DeleteStorageGroupTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/DeleteStorageGroupTask.java
@@ -26,8 +26,8 @@ import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteStorageGroupsReq;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.localconfignode.LocalConfigNode;
@@ -58,13 +58,12 @@ public class DeleteStorageGroupTask implements IConfigTask {
@Override
public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, DataNodeToConfigNodeClient> clientManager) {
+ IClientManager<PartitionRegionId, ConfigNodeClient> clientManager) {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
if (config.isClusterMode()) {
TDeleteStorageGroupsReq req =
new TDeleteStorageGroupsReq(deleteStorageGroupStatement.getPrefixPath());
- try (DataNodeToConfigNodeClient client =
- clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ try (ConfigNodeClient client = clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
TSStatus tsStatus = client.deleteStorageGroups(req);
if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
LOGGER.error(
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/DropFunctionTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/DropFunctionTask.java
index 17c67910b9..6f67983284 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/DropFunctionTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/DropFunctionTask.java
@@ -24,8 +24,8 @@ import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.consensus.PartitionRegionId;
import org.apache.iotdb.commons.udf.service.UDFRegistrationService;
import org.apache.iotdb.confignode.rpc.thrift.TDropFunctionReq;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DropFunctionStatement;
@@ -53,7 +53,7 @@ public class DropFunctionTask implements IConfigTask {
@Override
public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, DataNodeToConfigNodeClient> clientManager)
+ IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
throws InterruptedException {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
if (CONFIG.isClusterMode()) {
@@ -65,10 +65,9 @@ public class DropFunctionTask implements IConfigTask {
}
private void executeCluster(
- IClientManager<PartitionRegionId, DataNodeToConfigNodeClient> clientManager,
+ IClientManager<PartitionRegionId, ConfigNodeClient> clientManager,
SettableFuture<ConfigTaskResult> future) {
- try (DataNodeToConfigNodeClient client =
- clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ try (ConfigNodeClient client = clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
final TSStatus executionStatus = client.dropFunction(new TDropFunctionReq(udfName));
if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != executionStatus.getCode()) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/FlushTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/FlushTask.java
index adb28f3a7b..231a967ce0 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/FlushTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/FlushTask.java
@@ -24,8 +24,8 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.consensus.PartitionRegionId;
import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.localconfignode.LocalConfigNode;
@@ -55,7 +55,7 @@ public class FlushTask implements IConfigTask {
@Override
public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, DataNodeToConfigNodeClient> clientManager)
+ IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
throws InterruptedException {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
TSStatus tsStatus = new TSStatus();
@@ -77,8 +77,7 @@ public class FlushTask implements IConfigTask {
tFlushReq.setDataNodeId(-1);
}
if (config.isClusterMode()) {
- try (DataNodeToConfigNodeClient client =
- clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ try (ConfigNodeClient client = clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
// Send request to some API server
tsStatus = client.flush(tFlushReq);
// Get response or throw exception
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/IConfigTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/IConfigTask.java
index acf8c0bc01..f7fbd4164e 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/IConfigTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/IConfigTask.java
@@ -21,12 +21,12 @@ package org.apache.iotdb.db.mpp.plan.execution.config;
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.consensus.PartitionRegionId;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import com.google.common.util.concurrent.ListenableFuture;
public interface IConfigTask {
ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, DataNodeToConfigNodeClient> clientManager)
+ IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
throws InterruptedException;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/SetStorageGroupTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/SetStorageGroupTask.java
index cdc546824b..ef2c251366 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/SetStorageGroupTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/SetStorageGroupTask.java
@@ -24,8 +24,8 @@ import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.consensus.PartitionRegionId;
import org.apache.iotdb.confignode.rpc.thrift.TSetStorageGroupReq;
import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.localconfignode.LocalConfigNode;
@@ -54,17 +54,17 @@ public class SetStorageGroupTask implements IConfigTask {
@Override
public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, DataNodeToConfigNodeClient> clientManager) {
+ IClientManager<PartitionRegionId, ConfigNodeClient> clientManager) {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
// TODO:(this judgement needs to be integrated in a high level framework)
if (config.isClusterMode()) {
// Construct request using statement
TStorageGroupSchema storageGroupSchema = constructStorageGroupSchema();
TSetStorageGroupReq req = new TSetStorageGroupReq(storageGroupSchema);
- try (DataNodeToConfigNodeClient dataNodeToConfigNodeClient =
+ try (ConfigNodeClient configNodeClient =
clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
// Send request to some API server
- TSStatus tsStatus = dataNodeToConfigNodeClient.setStorageGroup(req);
+ TSStatus tsStatus = configNodeClient.setStorageGroup(req);
// Get response or throw exception
if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
LOGGER.error(
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/SetTTLTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/SetTTLTask.java
index 57dc3335b9..241634f074 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/SetTTLTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/SetTTLTask.java
@@ -24,8 +24,8 @@ import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.consensus.PartitionRegionId;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.confignode.rpc.thrift.TSetTTLReq;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.localconfignode.LocalConfigNode;
@@ -56,16 +56,16 @@ public class SetTTLTask implements IConfigTask {
@Override
public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, DataNodeToConfigNodeClient> clientManager)
+ IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
throws InterruptedException {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
if (config.isClusterMode()) {
TSetTTLReq setTTLReq =
new TSetTTLReq(statement.getStorageGroupPath().getFullPath(), statement.getTTL());
- try (DataNodeToConfigNodeClient dataNodeToConfigNodeClient =
+ try (ConfigNodeClient configNodeClient =
clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
// Send request to some API server
- TSStatus tsStatus = dataNodeToConfigNodeClient.setTTL(setTTLReq);
+ TSStatus tsStatus = configNodeClient.setTTL(setTTLReq);
// Get response or throw exception
if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
LOGGER.error(
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowClusterTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowClusterTask.java
index 65d8fb4042..1be47b1c9d 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowClusterTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowClusterTask.java
@@ -22,8 +22,8 @@ package org.apache.iotdb.db.mpp.plan.execution.config;
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.consensus.PartitionRegionId;
import org.apache.iotdb.confignode.rpc.thrift.TClusterNodeInfos;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.mpp.common.header.DatasetHeader;
@@ -56,14 +56,13 @@ public class ShowClusterTask implements IConfigTask {
@Override
public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, DataNodeToConfigNodeClient> clientManager)
+ IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
throws InterruptedException {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
TClusterNodeInfos clusterNodeInfos = new TClusterNodeInfos();
if (config.isClusterMode()) {
- try (DataNodeToConfigNodeClient client =
- clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ try (ConfigNodeClient client = clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
clusterNodeInfos = client.getAllClusterNodeInfos();
} catch (TException | IOException e) {
LOGGER.error("Failed to connect to config node.");
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowFunctionsTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowFunctionsTask.java
index c0dd203755..c51b1d59c3 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowFunctionsTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowFunctionsTask.java
@@ -25,7 +25,7 @@ import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.udf.builtin.BuiltinAggregationFunction;
import org.apache.iotdb.commons.udf.service.UDFRegistrationInformation;
import org.apache.iotdb.commons.udf.service.UDFRegistrationService;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.mpp.common.header.HeaderConstant;
import org.apache.iotdb.db.query.dataset.ListDataSet;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -58,7 +58,7 @@ public class ShowFunctionsTask implements IConfigTask {
@Override
public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, DataNodeToConfigNodeClient> clientManager)
+ IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
throws InterruptedException {
final SettableFuture<ConfigTaskResult> future = SettableFuture.create();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowStorageGroupTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowStorageGroupTask.java
index bd9d8e597c..79894f4824 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowStorageGroupTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowStorageGroupTask.java
@@ -25,8 +25,8 @@ import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchemaResp;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.localconfignode.LocalConfigNode;
@@ -63,15 +63,14 @@ public class ShowStorageGroupTask implements IConfigTask {
@Override
public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, DataNodeToConfigNodeClient> clientManager)
+ IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
throws InterruptedException {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
Map<String, TStorageGroupSchema> storageGroupSchemaMap = new HashMap<>();
if (config.isClusterMode()) {
List<String> storageGroupPathPattern =
Arrays.asList(showStorageGroupStatement.getPathPattern().getNodes());
- try (DataNodeToConfigNodeClient client =
- clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ try (ConfigNodeClient client = clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
TStorageGroupSchemaResp resp =
client.getMatchedStorageGroupSchemas(storageGroupPathPattern);
storageGroupSchemaMap = resp.getStorageGroupSchemaMap();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowTTLTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowTTLTask.java
index c80fe7a9fc..02f348d970 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowTTLTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowTTLTask.java
@@ -25,8 +25,8 @@ import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchemaResp;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.localconfignode.LocalConfigNode;
@@ -62,14 +62,13 @@ public class ShowTTLTask implements IConfigTask {
@Override
public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, DataNodeToConfigNodeClient> clientManager)
+ IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
throws InterruptedException {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
List<PartialPath> storageGroupPaths = showTTLStatement.getPaths();
Map<String, Long> storageGroupToTTL = new HashMap<>();
if (config.isClusterMode()) {
- try (DataNodeToConfigNodeClient client =
- clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ try (ConfigNodeClient client = clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
if (showTTLStatement.isAll()) {
List<String> allStorageGroupPathPattern = Arrays.asList("root", "**");
TStorageGroupSchemaResp resp =
diff --git a/server/src/main/java/org/apache/iotdb/db/service/DataNode.java b/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
index 7415e23a7a..629672dbe8 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
@@ -34,8 +34,8 @@ import org.apache.iotdb.commons.udf.service.UDFExecutableManager;
import org.apache.iotdb.commons.udf.service.UDFRegistrationService;
import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRegisterReq;
import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRegisterResp;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.conf.IoTDBStartCheck;
@@ -168,8 +168,7 @@ public class DataNode implements DataNodeMBean {
.updateConfigNodeList(IoTDBDescriptor.getInstance().getConfig().getConfigNodeList());
while (retry > 0) {
logger.info("start joining the cluster.");
- try (DataNodeToConfigNodeClient dataNodeToConfigNodeClient =
- new DataNodeToConfigNodeClient()) {
+ try (ConfigNodeClient configNodeClient = new ConfigNodeClient()) {
IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
// Set DataNodeLocation
@@ -193,8 +192,7 @@ public class DataNode implements DataNodeMBean {
TDataNodeRegisterReq req = new TDataNodeRegisterReq();
req.setDataNodeInfo(info);
- TDataNodeRegisterResp dataNodeRegisterResp =
- dataNodeToConfigNodeClient.registerDataNode(req);
+ TDataNodeRegisterResp dataNodeRegisterResp = configNodeClient.registerDataNode(req);
// store config node lists from resp
List<TEndPoint> configNodeList = new ArrayList<>();
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/ConfigExecutionTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/ConfigExecutionTest.java
index ee730d0ad3..0741a6b4fa 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/ConfigExecutionTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/ConfigExecutionTest.java
@@ -22,7 +22,7 @@ package org.apache.iotdb.db.mpp.execution;
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.consensus.PartitionRegionId;
-import org.apache.iotdb.db.client.DataNodeToConfigNodeClient;
+import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.mpp.common.MPPQueryContext;
import org.apache.iotdb.db.mpp.common.QueryId;
import org.apache.iotdb.db.mpp.common.header.ColumnHeader;
@@ -115,7 +115,7 @@ public class ConfigExecutionTest {
@Override
public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, DataNodeToConfigNodeClient> clientManager)
+ IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
throws InterruptedException {
return result;
}
diff --git a/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java b/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
index 4d8ea75587..90a36ede1d 100644
--- a/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
+++ b/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
@@ -117,11 +117,10 @@ public enum TSStatusCode {
NO_CONNECTION(706),
NEED_REDIRECTION(707),
PARSE_LOG_ERROR(708),
- NEED_RETRY(709),
+ ALL_RETRY_FAILED(709),
MIGRATE_REGION_ERROR(710),
CREATE_REGION_ERROR(711),
DELETE_REGION_ERROR(712),
- ALL_RETRY_FAILEDD(713),
// configuration
CONFIG_ERROR(800),
@@ -131,8 +130,8 @@ public enum TSStatusCode {
SET_STORAGE_GROUP_FAILED(902),
STORAGE_GROUP_ALREADY_EXISTS(903),
NOT_ENOUGH_DATA_NODE(904),
- INCONSISTENT_SYSTEM_CONFIG(905),
- REGISTER_CONFIGNODE_FAILED(906);
+ ERROR_GLOBAL_CONFIG(905),
+ APPLY_CONFIGNODE_FAILED(906);
private int statusCode;
diff --git a/thrift-confignode/src/main/thrift/confignode.thrift b/thrift-confignode/src/main/thrift/confignode.thrift
index 3f16b09684..f1bd0e0818 100644
--- a/thrift-confignode/src/main/thrift/confignode.thrift
+++ b/thrift-confignode/src/main/thrift/confignode.thrift
@@ -105,7 +105,7 @@ struct TStorageGroupSchema {
7: optional i32 maximumDataRegionCount
}
-// SchemaPartition
+// Schema
struct TSchemaPartitionReq {
1: required binary pathPatternTree
}
@@ -116,19 +116,8 @@ struct TSchemaPartitionResp {
2: optional map<string, map<common.TSeriesPartitionSlot, common.TRegionReplicaSet>> schemaRegionMap
}
-// DataPartition
-struct TDataPartitionReq {
- // map<StorageGroupName, map<TSeriesPartitionSlot, list<TTimePartitionSlot>>>
- 1: required map<string, map<common.TSeriesPartitionSlot, list<common.TTimePartitionSlot>>> partitionSlotsMap
-}
-
-struct TDataPartitionResp {
- 1: required common.TSStatus status
- // map<StorageGroupName, map<TSeriesPartitionSlot, map<TTimePartitionSlot, list<TRegionReplicaSet>>>>
- 2: optional map<string, map<common.TSeriesPartitionSlot, map<common.TTimePartitionSlot, list<common.TRegionReplicaSet>>>> dataPartitionMap
-}
-
// Node Management
+
struct TSchemaNodeManagementReq {
1: required binary pathPatternTree
2: optional i32 level
@@ -141,6 +130,18 @@ struct TSchemaNodeManagementResp {
3: optional set<string> matchedNode
}
+// Data
+struct TDataPartitionReq {
+ // map<StorageGroupName, map<TSeriesPartitionSlot, list<TTimePartitionSlot>>>
+ 1: required map<string, map<common.TSeriesPartitionSlot, list<common.TTimePartitionSlot>>> partitionSlotsMap
+}
+
+struct TDataPartitionResp {
+ 1: required common.TSStatus status
+ // map<StorageGroupName, map<TSeriesPartitionSlot, map<TTimePartitionSlot, list<TRegionReplicaSet>>>>
+ 2: optional map<string, map<common.TSeriesPartitionSlot, map<common.TTimePartitionSlot, list<common.TRegionReplicaSet>>>> dataPartitionMap
+}
+
// Authorize
struct TAuthorizerReq {
1: required i32 authorType
@@ -188,17 +189,21 @@ struct TCheckUserPrivilegesReq {
// ConfigNode
struct TConfigNodeRegisterReq {
- 1: required common.TConsensusGroupId partitionRegionId
- 2: required common.TConfigNodeLocation configNodeLocation
- 3: required TGlobalConfig globalConfig
- 4: required i64 defaultTTL
- 5: required i32 schemaReplicationFactor
- 6: required i32 dataReplicationFactor
+ 1: required common.TConfigNodeLocation configNodeLocation
+ 2: required string dataRegionConsensusProtocolClass
+ 3: required string schemaRegionConsensusProtocolClass
+ 4: required i32 seriesPartitionSlotNum
+ 5: required string seriesPartitionExecutorClass
+ 6: required i64 defaultTTL
+ 7: required i64 timePartitionInterval
+ 8: required i32 schemaReplicationFactor
+ 9: required i32 dataReplicationFactor
}
struct TConfigNodeRegisterResp {
1: required common.TSStatus status
- 2: required list<common.TConfigNodeLocation> configNodeList
+ 2: optional common.TConsensusGroupId partitionRegionId
+ 3: optional list<common.TConfigNodeLocation> configNodeList
}
// Show cluster
@@ -280,6 +285,8 @@ service ConfigIService {
TConfigNodeRegisterResp registerConfigNode(TConfigNodeRegisterReq req)
+ common.TSStatus applyConfigNode(common.TConfigNodeLocation configNodeLocation)
+
/* UDF */
common.TSStatus createFunction(TCreateFunctionReq req)