This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 06f4d81c41d [To dev/1.3] Adjust default data region num per node from
5 to to cpu core / 2 & IoTConsensusV2: Flush old leader when leader transfer &
Delay the execution of invalidateSchemaCache after leader change (#14963)
06f4d81c41d is described below
commit 06f4d81c41d0d8bae730dc5baad1990290e242c7
Author: Peng Junzhi <[email protected]>
AuthorDate: Thu Feb 27 11:32:28 2025 +0800
[To dev/1.3] Adjust default data region num per node from 5 to to cpu core
/ 2 & IoTConsensusV2: Flush old leader when leader transfer & Delay the
execution of invalidateSchemaCache after leader change (#14963)
* Adjust default data region num per node from 5 to to cpu core / 2 (#14348)
* finish
Signed-off-by: OneSizeFitQuorum <[email protected]>
* fix it & ut
Signed-off-by: OneSizeFitQuorum <[email protected]>
---------
Signed-off-by: OneSizeFitQuorum <[email protected]>
(cherry picked from commit f1714d05cdb6bfd007dc1efd2c543ea5d36fcf3a)
* merge
* merge
* spotless
---------
Co-authored-by: Potato <[email protected]>
---
.../it/cluster/IoTDBClusterNodeGetterIT.java | 6 +-
.../confignode/it/utils/ConfigNodeTestUtils.java | 4 +-
.../iotdb/confignode/conf/ConfigNodeConfig.java | 24 +++-
.../confignode/conf/ConfigNodeDescriptor.java | 25 ++--
.../iotdb/confignode/manager/ConfigManager.java | 9 ++
.../apache/iotdb/confignode/manager/IManager.java | 4 +
.../manager/load/balancer/RouteBalancer.java | 152 ++++++++++++++++-----
...artiteGraphReplicationRegionGroupAllocator.java | 7 +-
.../iotdb/confignode/manager/node/NodeManager.java | 8 ++
.../manager/schema/ClusterSchemaManager.java | 11 +-
.../confignode/persistence/node/NodeInfo.java | 9 --
.../region/AllocatorScatterWidthManualTest.java | 4 +-
.../GreedyCopySetRegionGroupAllocatorTest.java | 4 +-
.../iotdb/db/storageengine/StorageEngine.java | 19 ++-
.../conf/iotdb-system.properties.template | 9 +-
.../thrift-commons/src/main/thrift/common.thrift | 1 +
.../src/main/thrift/confignode.thrift | 4 +-
17 files changed, 210 insertions(+), 90 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBClusterNodeGetterIT.java
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBClusterNodeGetterIT.java
index d31e2217249..1c1afe917c0 100644
---
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBClusterNodeGetterIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBClusterNodeGetterIT.java
@@ -162,12 +162,10 @@ public class IoTDBClusterNodeGetterIT {
clusterParameters.getSchemaReplicationFactor());
Assert.assertEquals(
expectedParameters.getDataRegionPerDataNode(),
- clusterParameters.getDataRegionPerDataNode(),
- 0.01);
+ clusterParameters.getDataRegionPerDataNode());
Assert.assertEquals(
expectedParameters.getSchemaRegionPerDataNode(),
- clusterParameters.getSchemaRegionPerDataNode(),
- 0.01);
+ clusterParameters.getSchemaRegionPerDataNode());
Assert.assertEquals(
expectedParameters.getDiskSpaceWarningThreshold(),
clusterParameters.getDiskSpaceWarningThreshold(),
diff --git
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/utils/ConfigNodeTestUtils.java
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/utils/ConfigNodeTestUtils.java
index 76feab5efe3..5f9da5ac36a 100644
---
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/utils/ConfigNodeTestUtils.java
+++
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/utils/ConfigNodeTestUtils.java
@@ -323,8 +323,8 @@ public class ConfigNodeTestUtils {
clusterParameters.setTimePartitionInterval(604800000);
clusterParameters.setDataReplicationFactor(1);
clusterParameters.setSchemaReplicationFactor(1);
- clusterParameters.setDataRegionPerDataNode(5.0);
- clusterParameters.setSchemaRegionPerDataNode(1.0);
+ clusterParameters.setDataRegionPerDataNode(0);
+ clusterParameters.setSchemaRegionPerDataNode(1);
clusterParameters.setDiskSpaceWarningThreshold(0.01);
clusterParameters.setReadConsistencyLevel("strong");
clusterParameters.setTimestampPrecision("ms");
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java
index d116e4aec5b..ffb61da07e1 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java
@@ -93,7 +93,7 @@ public class ConfigNodeConfig {
private int defaultSchemaRegionGroupNumPerDatabase = 1;
/** The maximum number of SchemaRegions expected to be managed by each
DataNode. */
- private double schemaRegionPerDataNode = schemaReplicationFactor;
+ private int schemaRegionPerDataNode = 1;
/** The policy of extension DataRegionGroup for each Database. */
private RegionGroupExtensionPolicy dataRegionGroupExtensionPolicy =
@@ -106,8 +106,14 @@ public class ConfigNodeConfig {
*/
private int defaultDataRegionGroupNumPerDatabase = 2;
- /** The maximum number of DataRegions expected to be managed by each
DataNode. */
- private double dataRegionPerDataNode = 5.0;
+ /**
+ * The maximum number of DataRegions expected to be managed by each
DataNode. Set to 0 means that
+ * each dataNode automatically has the number of CPU cores / 2 regions.
+ */
+ private int dataRegionPerDataNode = 0;
+
+ /** each dataNode automatically has the number of CPU cores / 2 regions. */
+ private double dataRegionPerDataNodeProportion = 0.5;
/** RegionGroup allocate policy. */
private RegionBalancer.RegionGroupAllocatePolicy regionGroupAllocatePolicy =
@@ -481,11 +487,11 @@ public class ConfigNodeConfig {
this.defaultDataRegionGroupNumPerDatabase =
defaultDataRegionGroupNumPerDatabase;
}
- public double getSchemaRegionPerDataNode() {
+ public int getSchemaRegionPerDataNode() {
return schemaRegionPerDataNode;
}
- public void setSchemaRegionPerDataNode(double schemaRegionPerDataNode) {
+ public void setSchemaRegionPerDataNode(int schemaRegionPerDataNode) {
this.schemaRegionPerDataNode = schemaRegionPerDataNode;
}
@@ -497,14 +503,18 @@ public class ConfigNodeConfig {
this.dataRegionConsensusProtocolClass = dataRegionConsensusProtocolClass;
}
- public double getDataRegionPerDataNode() {
+ public int getDataRegionPerDataNode() {
return dataRegionPerDataNode;
}
- public void setDataRegionPerDataNode(double dataRegionPerDataNode) {
+ public void setDataRegionPerDataNode(int dataRegionPerDataNode) {
this.dataRegionPerDataNode = dataRegionPerDataNode;
}
+ public double getDataRegionPerDataNodeProportion() {
+ return dataRegionPerDataNodeProportion;
+ }
+
public RegionBalancer.RegionGroupAllocatePolicy
getRegionGroupAllocatePolicy() {
return regionGroupAllocatePolicy;
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java
index 918c1856a00..8369998f966 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java
@@ -238,12 +238,13 @@ public class ConfigNodeDescriptor {
.trim()));
conf.setSchemaRegionPerDataNode(
- Double.parseDouble(
- properties
- .getProperty(
- "schema_region_per_data_node",
- String.valueOf(conf.getSchemaRegionPerDataNode()))
- .trim()));
+ (int)
+ Double.parseDouble(
+ properties
+ .getProperty(
+ "schema_region_per_data_node",
+ String.valueOf(conf.getSchemaRegionPerDataNode()))
+ .trim()));
conf.setDataRegionGroupExtensionPolicy(
RegionGroupExtensionPolicy.parse(
@@ -258,11 +259,13 @@ public class ConfigNodeDescriptor {
String.valueOf(conf.getDefaultDataRegionGroupNumPerDatabase()).trim())));
conf.setDataRegionPerDataNode(
- Double.parseDouble(
- properties
- .getProperty(
- "data_region_per_data_node",
String.valueOf(conf.getDataRegionPerDataNode()))
- .trim()));
+ (int)
+ Double.parseDouble(
+ properties
+ .getProperty(
+ "data_region_per_data_node",
+ String.valueOf(conf.getDataRegionPerDataNode()))
+ .trim()));
try {
conf.setRegionAllocateStrategy(
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
index d9b5aa422ab..ebdf8ca9266 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
@@ -1564,6 +1564,15 @@ public class ConfigManager implements IManager {
: status;
}
+ @Override
+ public TSStatus flushOnSpecificDN(
+ final TFlushReq req, final Map<Integer, TDataNodeLocation>
dataNodeLocationMap) {
+ final TSStatus status = confirmLeader();
+ return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
+ ? RpcUtils.squashResponseStatusList(nodeManager.flushOnSpecificDN(req,
dataNodeLocationMap))
+ : status;
+ }
+
@Override
public TSStatus clearCache() {
TSStatus status = confirmLeader();
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
index ac826c761cd..6fd0a8b567c 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
@@ -146,6 +146,7 @@ import org.apache.iotdb.consensus.common.DataSet;
import org.apache.iotdb.rpc.TSStatusCode;
import java.util.List;
+import java.util.Map;
/**
* A subset of services provided by {@link ConfigManager}. For use internally
only, passed to
@@ -537,6 +538,9 @@ public interface IManager {
/** Flush on all DataNodes. */
TSStatus flush(TFlushReq req);
+ /** Flush on specific Datanode. */
+ TSStatus flushOnSpecificDN(TFlushReq req, Map<Integer, TDataNodeLocation>
dataNodeLocationMap);
+
/** Clear cache on all DataNodes. */
TSStatus clearCache();
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RouteBalancer.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RouteBalancer.java
index 8bfc3d418a1..144e5a24368 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RouteBalancer.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RouteBalancer.java
@@ -22,6 +22,7 @@ import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TFlushReq;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.cluster.NodeStatus;
@@ -56,15 +57,19 @@ import org.apache.tsfile.utils.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.BiConsumer;
import java.util.stream.Collectors;
/** The RouteBalancer guides the cluster RegionGroups' leader distribution and
routing priority. */
@@ -114,11 +119,16 @@ public class RouteBalancer implements
IClusterStatusSubscriber {
private static final long BALANCE_RATIS_LEADER_FAILED_INTERVAL_IN_NS = 20 *
1000L * 1000L * 1000L;
private final Map<TConsensusGroupId, Long> lastFailedTimeForLeaderBalance;
+ private final Map<Integer, List<String>> lastBalancedOldLeaderId2RegionMap;
+ private Map<TConsensusGroupId, Integer> lastDataRegion2OldLeaderMap;
+ private Set<TConsensusGroupId> lastBalancedDataRegionSet;
+
public RouteBalancer(IManager configManager) {
this.configManager = configManager;
this.priorityMapLock = new ReentrantReadWriteLock();
this.regionPriorityMap = new TreeMap<>();
this.lastFailedTimeForLeaderBalance = new TreeMap<>();
+ this.lastBalancedOldLeaderId2RegionMap = new ConcurrentHashMap<>();
switch (CONF.getLeaderDistributionPolicy()) {
case AbstractLeaderBalancer.GREEDY_POLICY:
@@ -178,20 +188,38 @@ public class RouteBalancer implements
IClusterStatusSubscriber {
return;
}
- if (newLeaderId != -1 &&
!newLeaderId.equals(currentLeaderMap.get(regionGroupId))) {
+ int oldLeaderId = currentLeaderMap.get(regionGroupId);
+ if (newLeaderId != -1 && !newLeaderId.equals(oldLeaderId)) {
LOGGER.info(
"[LeaderBalancer] Try to change the leader of Region: {} to
DataNode: {} ",
regionGroupId,
newLeaderId);
switch (consensusProtocolClass) {
- case ConsensusFactory.IOT_CONSENSUS_V2:
case ConsensusFactory.IOT_CONSENSUS:
case ConsensusFactory.SIMPLE_CONSENSUS:
- // For IoTConsensus or SimpleConsensus or PipeConsensus
protocol, change
+ // For IoTConsensus or SimpleConsensus protocol, change
// RegionRouteMap is enough
successTransferMap.put(
regionGroupId, new
ConsensusGroupHeartbeatSample(currentTime, newLeaderId));
break;
+ case ConsensusFactory.IOT_CONSENSUS_V2:
+ // For IoTConsensusV2 protocol, change RegionRouteMap and
execute flush on old
+ // region leader
+ successTransferMap.put(
+ regionGroupId, new
ConsensusGroupHeartbeatSample(currentTime, newLeaderId));
+ // Prepare data for flushOldLeader
+ lastBalancedOldLeaderId2RegionMap.compute(
+ oldLeaderId,
+ (k, v) -> {
+ if (v == null) {
+ List<String> value = new ArrayList<>();
+ value.add(String.valueOf(regionGroupId.getId()));
+ return value;
+ }
+ v.add(String.valueOf(regionGroupId.getId()));
+ return v;
+ });
+ break;
case ConsensusFactory.RATIS_CONSENSUS:
default:
// For ratis protocol, the ConfigNode-leader will send a
changeLeaderRequest to the
@@ -246,46 +274,97 @@ public class RouteBalancer implements
IClusterStatusSubscriber {
getLoadManager().forceUpdateConsensusGroupCache(successTransferMap);
- invalidateSchemaCacheOfOldLeaders(currentLeaderMap,
successTransferMap.keySet());
+ // Prepare data for invalidSchemaCacheOfOldLeaders
+ if (regionGroupType.equals(TConsensusGroupType.DataRegion)) {
+ lastBalancedDataRegionSet = successTransferMap.keySet();
+ lastDataRegion2OldLeaderMap = currentLeaderMap;
+ }
}
- private void invalidateSchemaCacheOfOldLeaders(
- final Map<TConsensusGroupId, Integer> oldLeaderMap,
- final Set<TConsensusGroupId> successTransferSet) {
- final DataNodeAsyncRequestContext<String, TSStatus>
invalidateSchemaCacheRequestHandler =
- new
DataNodeAsyncRequestContext<>(CnToDnAsyncRequestType.INVALIDATE_LAST_CACHE);
- final AtomicInteger requestIndex = new AtomicInteger(0);
- oldLeaderMap.entrySet().stream()
- .filter(entry -> TConsensusGroupType.DataRegion ==
entry.getKey().getType())
- .filter(entry -> successTransferSet.contains(entry.getKey()))
- .forEach(
- entry -> {
- // set target
- final Integer dataNodeId = entry.getValue();
- if (dataNodeId == -1) {
- return;
- }
- final TDataNodeLocation dataNodeLocation =
-
getNodeManager().getRegisteredDataNode(dataNodeId).getLocation();
- if (dataNodeLocation == null) {
- LOGGER.warn("DataNodeLocation is null, datanodeId {}",
dataNodeId);
- return;
- }
- invalidateSchemaCacheRequestHandler.putNodeLocation(
- requestIndex.get(), dataNodeLocation);
- // set req
- final TConsensusGroupId consensusGroupId = entry.getKey();
- final String database =
getPartitionManager().getRegionDatabase(consensusGroupId);
-
invalidateSchemaCacheRequestHandler.putRequest(requestIndex.get(), database);
- requestIndex.incrementAndGet();
- });
- CnToDnInternalServiceAsyncRequestManager.getInstance()
- .sendAsyncRequest(invalidateSchemaCacheRequestHandler);
+ private void invalidateSchemaCacheOfOldLeaders() {
+ BiConsumer<Map<TConsensusGroupId, Integer>, Set<TConsensusGroupId>>
consumer =
+ (oldLeaderMap, successTransferSet) -> {
+ final DataNodeAsyncRequestContext<String, TSStatus>
invalidateSchemaCacheRequestHandler =
+ new
DataNodeAsyncRequestContext<>(CnToDnAsyncRequestType.INVALIDATE_LAST_CACHE);
+ final AtomicInteger requestIndex = new AtomicInteger(0);
+ oldLeaderMap.entrySet().stream()
+ .filter(entry -> successTransferSet.contains(entry.getKey()))
+ .forEach(
+ entry -> {
+ // set target
+ final Integer dataNodeId = entry.getValue();
+ if (dataNodeId == -1) {
+ return;
+ }
+ final TDataNodeLocation dataNodeLocation =
+
getNodeManager().getRegisteredDataNode(dataNodeId).getLocation();
+ if (dataNodeLocation == null) {
+ LOGGER.warn("DataNodeLocation is null, datanodeId {}",
dataNodeId);
+ return;
+ }
+ invalidateSchemaCacheRequestHandler.putNodeLocation(
+ requestIndex.get(), dataNodeLocation);
+ // set req
+ final TConsensusGroupId consensusGroupId = entry.getKey();
+ final String database =
+
getPartitionManager().getRegionDatabase(consensusGroupId);
+
invalidateSchemaCacheRequestHandler.putRequest(requestIndex.get(), database);
+ requestIndex.incrementAndGet();
+ });
+ CnToDnInternalServiceAsyncRequestManager.getInstance()
+ .sendAsyncRequest(invalidateSchemaCacheRequestHandler);
+ };
+
+ if (IS_ENABLE_AUTO_LEADER_BALANCE_FOR_DATA_REGION) {
+ consumer.accept(lastDataRegion2OldLeaderMap, lastBalancedDataRegionSet);
+ }
+ }
+
+ private void flushOldLeaderIfIoTV2() {
+ if (!IS_ENABLE_AUTO_LEADER_BALANCE_FOR_DATA_REGION
+ || !Objects.equals(
+ DATA_REGION_CONSENSUS_PROTOCOL_CLASS,
ConsensusFactory.IOT_CONSENSUS_V2)) {
+ return;
+ }
+
+ BiConsumer<Integer, List<String>> consumer =
+ (oldLeaderId, regionGroupIds) -> {
+ TDataNodeConfiguration configuration =
+ getNodeManager().getRegisteredDataNode(oldLeaderId);
+ Map<Integer, TDataNodeLocation> oldLeaderDataNodeLocation = new
HashMap<>();
+ oldLeaderDataNodeLocation.put(
+ configuration.getLocation().dataNodeId,
configuration.getLocation());
+
+ TFlushReq flushReq = new TFlushReq();
+ flushReq.setRegionIds(regionGroupIds);
+ // Do our best to flush. If flush failed, never retry
+ TSStatus result = configManager.flushOnSpecificDN(flushReq,
oldLeaderDataNodeLocation);
+ if (result.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode())
{
+ LOGGER.info(
+ "[IoTConsensusV2 Leader Changed] Successfully flush old leader
{} for region {}",
+ oldLeaderId,
+ regionGroupIds);
+ } else {
+ LOGGER.info(
+ "[IoTConsensusV2 Leader Changed] Failed to flush old leader {}
for region {}",
+ oldLeaderId,
+ regionGroupIds);
+ }
+ };
+ lastBalancedOldLeaderId2RegionMap.forEach(consumer);
+ // after flush, clear map for next balance
+ lastBalancedOldLeaderId2RegionMap.clear();
+ }
+
+ private synchronized void handleBalanceAction() {
+ invalidateSchemaCacheOfOldLeaders();
+ flushOldLeaderIfIoTV2();
}
public synchronized void balanceRegionLeaderAndPriority() {
balanceRegionLeader();
balanceRegionPriority();
+ handleBalanceAction();
}
/** Balance cluster RegionGroup route priority through configured algorithm.
*/
@@ -468,5 +547,6 @@ public class RouteBalancer implements
IClusterStatusSubscriber {
public void
onConsensusGroupStatisticsChanged(ConsensusGroupStatisticsChangeEvent event) {
balanceRegionLeader();
balanceRegionPriority();
+ handleBalanceAction();
}
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/PartiteGraphReplicationRegionGroupAllocator.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/PartiteGraphReplicationRegionGroupAllocator.java
index b24acc1bd46..1205e0abc06 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/PartiteGraphReplicationRegionGroupAllocator.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/PartiteGraphReplicationRegionGroupAllocator.java
@@ -69,10 +69,9 @@ public class PartiteGraphReplicationRegionGroupAllocator
implements IRegionGroup
TConsensusGroupId consensusGroupId) {
this.regionPerDataNode =
- (int)
- (consensusGroupId.getType().equals(TConsensusGroupType.DataRegion)
- ?
ConfigNodeDescriptor.getInstance().getConf().getDataRegionPerDataNode()
- :
ConfigNodeDescriptor.getInstance().getConf().getSchemaRegionPerDataNode());
+ consensusGroupId.getType().equals(TConsensusGroupType.DataRegion)
+ ?
ConfigNodeDescriptor.getInstance().getConf().getDataRegionPerDataNode()
+ :
ConfigNodeDescriptor.getInstance().getConf().getSchemaRegionPerDataNode();
prepare(replicationFactor, availableDataNodeMap, allocatedRegionGroups);
// Select a set of optimal alpha nodes
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
index 9653ff6fa87..b39a18689f4 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
@@ -875,6 +875,14 @@ public class NodeManager {
return clientHandler.getResponseList();
}
+ public List<TSStatus> flushOnSpecificDN(
+ TFlushReq req, Map<Integer, TDataNodeLocation> dataNodeLocationMap) {
+ DataNodeAsyncRequestContext<TFlushReq, TSStatus> clientHandler =
+ new DataNodeAsyncRequestContext<>(CnToDnAsyncRequestType.FLUSH, req,
dataNodeLocationMap);
+
CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler);
+ return clientHandler.getResponseList();
+ }
+
public List<TSStatus> clearCache() {
Map<Integer, TDataNodeLocation> dataNodeLocationMap =
configManager.getNodeManager().getRegisteredDataNodeLocations();
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java
index 25b65e606cd..994fc83ffe7 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java
@@ -107,8 +107,8 @@ public class ClusterSchemaManager {
private static final Logger LOGGER =
LoggerFactory.getLogger(ClusterSchemaManager.class);
private static final ConfigNodeConfig CONF =
ConfigNodeDescriptor.getInstance().getConf();
- private static final double SCHEMA_REGION_PER_DATA_NODE =
CONF.getSchemaRegionPerDataNode();
- private static final double DATA_REGION_PER_DATA_NODE =
CONF.getDataRegionPerDataNode();
+ private static final int SCHEMA_REGION_PER_DATA_NODE =
CONF.getSchemaRegionPerDataNode();
+ private static final int DATA_REGION_PER_DATA_NODE =
CONF.getDataRegionPerDataNode();
private final IManager configManager;
private final ClusterSchemaInfo clusterSchemaInfo;
@@ -462,6 +462,7 @@ public class ClusterSchemaManager {
}
int dataNodeNum = getNodeManager().getRegisteredDataNodeCount();
+ int totalCpuCoreNum = getNodeManager().getDataNodeCpuCoreCount();
int databaseNum = databaseSchemaMap.size();
for (TDatabaseSchema databaseSchema : databaseSchemaMap.values()) {
@@ -515,8 +516,10 @@ public class ClusterSchemaManager {
int maxDataRegionGroupNum =
calcMaxRegionGroupNum(
databaseSchema.getMinDataRegionGroupNum(),
- DATA_REGION_PER_DATA_NODE,
- dataNodeNum,
+ DATA_REGION_PER_DATA_NODE == 0
+ ? CONF.getDataRegionPerDataNodeProportion()
+ : DATA_REGION_PER_DATA_NODE,
+ DATA_REGION_PER_DATA_NODE == 0 ? totalCpuCoreNum : dataNodeNum,
databaseNum,
databaseSchema.getDataReplicationFactor(),
allocatedDataRegionGroupCount);
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/node/NodeInfo.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/node/NodeInfo.java
index 54570321e9e..a0e0b219434 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/node/NodeInfo.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/node/NodeInfo.java
@@ -290,15 +290,6 @@ public class NodeInfo implements SnapshotProcessor {
return result;
}
- public int getDataNodeCpuCoreCount(int dataNodeId) {
- try {
- return registeredDataNodes.get(dataNodeId).getResource().getCpuCoreNum();
- } catch (Exception e) {
- LOGGER.warn("Get DataNode {} cpu core fail, will be treated as zero.",
dataNodeId, e);
- return 0;
- }
- }
-
/** Return the number of total cpu cores in online DataNodes. */
public int getDataNodeTotalCpuCoreCount() {
int result = 0;
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/region/AllocatorScatterWidthManualTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/region/AllocatorScatterWidthManualTest.java
index b159525268f..4a4a47dabd0 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/region/AllocatorScatterWidthManualTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/region/AllocatorScatterWidthManualTest.java
@@ -24,7 +24,6 @@ import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
-import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
import org.junit.Assert;
import org.junit.BeforeClass;
@@ -51,8 +50,7 @@ public class AllocatorScatterWidthManualTest {
private static final IRegionGroupAllocator ALLOCATOR = new
GreedyRegionGroupAllocator();
private static final int TEST_DATA_NODE_NUM = 50;
- private static final int DATA_REGION_PER_DATA_NODE =
- (int)
ConfigNodeDescriptor.getInstance().getConf().getDataRegionPerDataNode();
+ private static final int DATA_REGION_PER_DATA_NODE = 5;
private static final int DATA_REPLICATION_FACTOR = 3;
private static final Map<Integer, TDataNodeConfiguration>
AVAILABLE_DATA_NODE_MAP =
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/region/GreedyCopySetRegionGroupAllocatorTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/region/GreedyCopySetRegionGroupAllocatorTest.java
index 95a23ab3e64..b807b98986a 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/region/GreedyCopySetRegionGroupAllocatorTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/region/GreedyCopySetRegionGroupAllocatorTest.java
@@ -24,7 +24,6 @@ import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
-import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
import org.junit.Assert;
import org.junit.BeforeClass;
@@ -54,8 +53,7 @@ public class GreedyCopySetRegionGroupAllocatorTest {
private static final Random RANDOM = new Random();
private static final int TEST_DATABASE_NUM = 3;
private static final int TEST_DATA_NODE_NUM = 21;
- private static final int DATA_REGION_PER_DATA_NODE =
- (int)
ConfigNodeDescriptor.getInstance().getConf().getDataRegionPerDataNode();
+ private static final int DATA_REGION_PER_DATA_NODE = 5;
private static final Map<Integer, TDataNodeConfiguration>
AVAILABLE_DATA_NODE_MAP =
new HashMap<>();
private static final Map<Integer, Double> FREE_SPACE_MAP = new HashMap<>();
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
index 85fe54576b5..18d43931ba4 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
@@ -530,6 +530,21 @@ public class StorageEngine implements IService {
checkResults(tasks, "Failed to sync close processor.");
}
+ public void syncCloseProcessorsInRegion(List<String> dataRegionIds) {
+ List<Future<Void>> tasks = new ArrayList<>();
+ for (DataRegion dataRegion : dataRegionMap.values()) {
+ if (dataRegion != null &&
dataRegionIds.contains(dataRegion.getDataRegionId())) {
+ tasks.add(
+ cachedThreadPool.submit(
+ () -> {
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
+ return null;
+ }));
+ }
+ }
+ checkResults(tasks, "Failed to sync close processor.");
+ }
+
public void syncCloseProcessorsInDatabase(String databaseName, boolean
isSeq) {
List<Future<Void>> tasks = new ArrayList<>();
for (DataRegion dataRegion : dataRegionMap.values()) {
@@ -638,7 +653,9 @@ public class StorageEngine implements IService {
}
public void operateFlush(TFlushReq req) {
- if (req.storageGroups == null) {
+ if (req.getRegionIds() != null && !req.getRegionIds().isEmpty()) {
+
StorageEngine.getInstance().syncCloseProcessorsInRegion(req.getRegionIds());
+ } else if (req.storageGroups == null || req.storageGroups.isEmpty()) {
StorageEngine.getInstance().syncCloseAllProcessor();
WALManager.getInstance().syncDeleteOutdatedFilesInWALNodes();
} else {
diff --git
a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
index 1e4eb8ba65d..f09bfc10e9e 100644
---
a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
+++
b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
@@ -606,8 +606,8 @@ default_schema_region_group_num_per_database=1
# Notice: Since each Database requires at least one SchemaRegionGroup to
manage its schema,
# this parameter doesn't limit the upper bound of cluster SchemaRegions when
there are too many Databases.
# effectiveMode: restart
-# Datatype: Double
-schema_region_per_data_node=1.0
+# Datatype: Integer
+schema_region_per_data_node=1
# The policy of extension DataRegionGroup for each Database.
# These policies are currently supported:
@@ -627,11 +627,12 @@ default_data_region_group_num_per_database=2
# Only take effect when set data_region_group_extension_policy=AUTO.
# This parameter is the maximum number of DataRegions expected to be managed
by each DataNode.
+# Set to 0 means that each node automatically has the number of CPU cores / 2
regions
# Notice: Since each Database requires at least two DataRegionGroups to manage
its data,
# this parameter doesn't limit the upper bound of cluster DataRegions when
there are too many Databases.
# effectiveMode: restart
-# Datatype: Double
-data_region_per_data_node=5.0
+# Datatype: Integer
+data_region_per_data_node=0
# Whether to enable auto leader balance for Ratis consensus protocol.
# The ConfigNode-leader will balance the leader of Ratis-RegionGroups by
leader_distribution_policy if set true.
diff --git a/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
b/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
index 2b4b67cf5b9..aed1b5f88f5 100644
--- a/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
+++ b/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
@@ -119,6 +119,7 @@ enum TRegionMaintainTaskStatus {
struct TFlushReq {
1: optional string isSeq
2: optional list<string> storageGroups
+ 3: optional list<string> regionIds
}
struct TSettleReq {
diff --git a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
index 8183790a2bf..0adf5ade231 100644
--- a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
+++ b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
@@ -405,8 +405,8 @@ struct TClusterParameters {
6: required string configNodeConsensusProtocolClass
7: required i64 timePartitionInterval
8: required string readConsistencyLevel
- 9: required double schemaRegionPerDataNode
- 10: required double dataRegionPerDataNode
+ 9: required i32 schemaRegionPerDataNode
+ 10: required i32 dataRegionPerDataNode
11: required i32 seriesPartitionSlotNum
12: required string seriesPartitionExecutorClass
13: required double diskSpaceWarningThreshold
