This is an automated email from the ASF dual-hosted git repository. yongzao pushed a commit to branch addlogicalclockinterface in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 9bec1206ef20856475ce65b7785cd084d228f566 Author: YongzaoDan <[email protected]> AuthorDate: Fri Apr 12 22:15:30 2024 +0800 Ready for test --- .../heartbeat/ConfigNodeHeartbeatHandler.java | 14 +- .../heartbeat/DataNodeHeartbeatHandler.java | 66 +++-- .../iotdb/confignode/manager/load/LoadManager.java | 64 ++--- .../manager/load/balancer/RouteBalancer.java | 300 +++++++++++---------- .../manager/load/cache/AbstractLoadCache.java | 2 + .../confignode/manager/load/cache/LoadCache.java | 82 +----- .../load/cache/consensus/ConsensusCache.java | 77 +----- .../manager/load/cache/node/BaseNodeCache.java | 4 +- .../manager/load/cache/region/RegionCache.java | 46 +--- .../manager/load/service/EventService.java | 30 --- .../manager/load/service/HeartbeatService.java | 5 +- .../manager/load/service/StatisticsService.java | 69 +---- .../load/subscriber/IClusterStatusSubscriber.java | 8 + ...scriber.java => NodeStatisticsChangeEvent.java} | 22 +- .../RegionGroupStatisticsChangeEvent.java} | 33 +-- .../runtime/PipeLeaderChangeHandler.java | 12 + .../runtime/PipeRuntimeCoordinator.java | 12 + .../procedure/env/ConfigNodeProcedureEnv.java | 16 +- .../procedure/env/RegionMaintainHandler.java | 14 +- .../impl/schema/DeleteDatabaseProcedure.java | 3 - .../router/priority/GreedyPriorityTest.java | 2 +- .../priority/LeaderPriorityBalancerTest.java | 2 +- .../manager/load/cache/ConsensusCacheTest.java | 9 +- .../manager/load/cache/NodeCacheTest.java | 27 +- .../manager/load/cache/RegionGroupCacheTest.java | 50 +--- .../persistence/node/NodeStatisticsTest.java | 66 ----- .../statistics/RegionGroupStatisticsTest.java | 76 ------ .../partition/statistics/RegionStatisticsTest.java | 66 ----- 28 files changed, 370 insertions(+), 807 deletions(-) diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/heartbeat/ConfigNodeHeartbeatHandler.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/heartbeat/ConfigNodeHeartbeatHandler.java index 3e99f181f7a..9f2fbcabd99 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/heartbeat/ConfigNodeHeartbeatHandler.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/heartbeat/ConfigNodeHeartbeatHandler.java @@ -22,7 +22,7 @@ package org.apache.iotdb.confignode.client.async.handlers.heartbeat; import org.apache.iotdb.commons.client.ThriftClient; import org.apache.iotdb.commons.cluster.NodeStatus; import org.apache.iotdb.commons.cluster.NodeType; -import org.apache.iotdb.confignode.manager.load.cache.LoadCache; +import org.apache.iotdb.confignode.manager.load.LoadManager; import org.apache.iotdb.confignode.manager.load.cache.node.NodeHeartbeatSample; import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeHeartbeatResp; @@ -31,22 +31,24 @@ import org.apache.thrift.async.AsyncMethodCallback; public class ConfigNodeHeartbeatHandler implements AsyncMethodCallback<TConfigNodeHeartbeatResp> { private final int nodeId; - private final LoadCache loadCache; + private final LoadManager loadManager; - public ConfigNodeHeartbeatHandler(int nodeId, LoadCache loadCache) { + public ConfigNodeHeartbeatHandler(int nodeId, LoadManager loadManager) { this.nodeId = nodeId; - this.loadCache = loadCache; + this.loadManager = loadManager; } @Override public void onComplete(TConfigNodeHeartbeatResp resp) { - loadCache.cacheConfigNodeHeartbeatSample(nodeId, new NodeHeartbeatSample(resp)); + loadManager + .getLoadCache() + .cacheConfigNodeHeartbeatSample(nodeId, new NodeHeartbeatSample(resp)); } @Override public void onError(Exception e) { if (ThriftClient.isConnectionBroken(e)) { - loadCache.forceUpdateNodeCache( + loadManager.forceUpdateNodeCache( NodeType.ConfigNode, nodeId, new NodeHeartbeatSample(NodeStatus.Unknown)); } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/heartbeat/DataNodeHeartbeatHandler.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/heartbeat/DataNodeHeartbeatHandler.java index a29f11b89f0..d65cc68d658 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/heartbeat/DataNodeHeartbeatHandler.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/heartbeat/DataNodeHeartbeatHandler.java @@ -19,15 +19,19 @@ package org.apache.iotdb.confignode.client.async.handlers.heartbeat; +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType; import org.apache.iotdb.commons.client.ThriftClient; import org.apache.iotdb.commons.cluster.NodeStatus; import org.apache.iotdb.commons.cluster.NodeType; import org.apache.iotdb.commons.cluster.RegionStatus; -import org.apache.iotdb.confignode.manager.load.cache.LoadCache; +import org.apache.iotdb.confignode.conf.ConfigNodeConfig; +import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; +import org.apache.iotdb.confignode.manager.load.LoadManager; import org.apache.iotdb.confignode.manager.load.cache.consensus.ConsensusHeartbeatSample; import org.apache.iotdb.confignode.manager.load.cache.node.NodeHeartbeatSample; import org.apache.iotdb.confignode.manager.load.cache.region.RegionHeartbeatSample; import org.apache.iotdb.confignode.manager.pipe.coordinator.runtime.PipeRuntimeCoordinator; +import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.mpp.rpc.thrift.TDataNodeHeartbeatResp; import org.apache.thrift.async.AsyncMethodCallback; @@ -37,9 +41,15 @@ import java.util.function.Consumer; public class DataNodeHeartbeatHandler implements AsyncMethodCallback<TDataNodeHeartbeatResp> { + private static final ConfigNodeConfig CONF = ConfigNodeDescriptor.getInstance().getConf(); + private static final boolean SCHEMA_REGION_SHOULD_CACHE_CONSENSUS_SAMPLE = + ConsensusFactory.RATIS_CONSENSUS.equals(CONF.getSchemaRegionConsensusProtocolClass()); + private static final boolean DATA_REGION_SHOULD_CACHE_CONSENSUS_SAMPLE = + ConsensusFactory.RATIS_CONSENSUS.equals(CONF.getDataRegionConsensusProtocolClass()); + private final int nodeId; - private final LoadCache loadCache; + private final LoadManager loadManager; private final Map<Integer, Long> deviceNum; private final Map<Integer, Long> timeSeriesNum; @@ -52,7 +62,7 @@ public class DataNodeHeartbeatHandler implements AsyncMethodCallback<TDataNodeHe public DataNodeHeartbeatHandler( int nodeId, - LoadCache loadCache, + LoadManager loadManager, Map<Integer, Long> deviceNum, Map<Integer, Long> timeSeriesNum, Map<Integer, Long> regionDisk, @@ -61,7 +71,7 @@ public class DataNodeHeartbeatHandler implements AsyncMethodCallback<TDataNodeHe PipeRuntimeCoordinator pipeRuntimeCoordinator) { this.nodeId = nodeId; - this.loadCache = loadCache; + this.loadManager = loadManager; this.deviceNum = deviceNum; this.timeSeriesNum = timeSeriesNum; this.regionDisk = regionDisk; @@ -73,27 +83,37 @@ public class DataNodeHeartbeatHandler implements AsyncMethodCallback<TDataNodeHe @Override public void onComplete(TDataNodeHeartbeatResp heartbeatResp) { // Update NodeCache - loadCache.cacheDataNodeHeartbeatSample(nodeId, new NodeHeartbeatSample(heartbeatResp)); + loadManager + .getLoadCache() + .cacheDataNodeHeartbeatSample(nodeId, new NodeHeartbeatSample(heartbeatResp)); heartbeatResp .getJudgedLeaders() .forEach( (regionGroupId, isLeader) -> { // Update RegionGroupCache - loadCache.cacheRegionHeartbeatSample( - regionGroupId, - nodeId, - new RegionHeartbeatSample( - heartbeatResp.getHeartbeatTimestamp(), - // Region will inherit DataNode's status - RegionStatus.parse(heartbeatResp.getStatus()))); - - if (Boolean.TRUE.equals(isLeader)) { - // Update RegionRouteCache - loadCache.cacheConsensusSample( - regionGroupId, - new ConsensusHeartbeatSample( - heartbeatResp.getConsensusLogicalTimeMap().get(regionGroupId), nodeId)); + loadManager + .getLoadCache() + .cacheRegionHeartbeatSample( + regionGroupId, + nodeId, + new RegionHeartbeatSample( + heartbeatResp.getHeartbeatTimestamp(), + // Region will inherit DataNode's status + RegionStatus.parse(heartbeatResp.getStatus()))); + + if (((TConsensusGroupType.SchemaRegion.equals(regionGroupId.getType()) + && SCHEMA_REGION_SHOULD_CACHE_CONSENSUS_SAMPLE) + || (TConsensusGroupType.DataRegion.equals(regionGroupId.getType()) + && DATA_REGION_SHOULD_CACHE_CONSENSUS_SAMPLE)) + && Boolean.TRUE.equals(isLeader)) { + // Update ConsensusCache when necessary + loadManager + .getLoadCache() + .cacheConsensusSample( + regionGroupId, + new ConsensusHeartbeatSample( + heartbeatResp.getConsensusLogicalTimeMap().get(regionGroupId), nodeId)); } }); @@ -112,15 +132,17 @@ public class DataNodeHeartbeatHandler implements AsyncMethodCallback<TDataNodeHe pipeRuntimeCoordinator.parseHeartbeat(nodeId, heartbeatResp.getPipeMetaList()); } if (heartbeatResp.isSetConfirmedConfigNodeEndPoints()) { - loadCache.updateConfirmedConfigNodeEndPoints( - nodeId, heartbeatResp.getConfirmedConfigNodeEndPoints()); + loadManager + .getLoadCache() + .updateConfirmedConfigNodeEndPoints( + nodeId, heartbeatResp.getConfirmedConfigNodeEndPoints()); } } @Override public void onError(Exception e) { if (ThriftClient.isConnectionBroken(e)) { - loadCache.forceUpdateNodeCache( + loadManager.forceUpdateNodeCache( NodeType.DataNode, nodeId, new NodeHeartbeatSample(NodeStatus.Unknown)); } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java index 90bffe05687..b24159dbf2b 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java @@ -37,6 +37,7 @@ import org.apache.iotdb.confignode.manager.load.balancer.PartitionBalancer; import org.apache.iotdb.confignode.manager.load.balancer.RegionBalancer; import org.apache.iotdb.confignode.manager.load.balancer.RouteBalancer; import org.apache.iotdb.confignode.manager.load.cache.LoadCache; +import org.apache.iotdb.confignode.manager.load.cache.consensus.ConsensusHeartbeatSample; import org.apache.iotdb.confignode.manager.load.cache.node.NodeHeartbeatSample; import org.apache.iotdb.confignode.manager.load.cache.region.RegionHeartbeatSample; import org.apache.iotdb.confignode.manager.load.service.EventService; @@ -44,7 +45,6 @@ import org.apache.iotdb.confignode.manager.load.service.HeartbeatService; import org.apache.iotdb.confignode.manager.load.service.StatisticsService; import org.apache.iotdb.confignode.manager.partition.RegionGroupStatus; import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList; -import org.apache.iotdb.tsfile.utils.Pair; import java.util.List; import java.util.Map; @@ -80,7 +80,7 @@ public class LoadManager { this.loadCache = new LoadCache(); this.heartbeatService = new HeartbeatService(configManager, loadCache); - this.statisticsService = new StatisticsService(configManager, routeBalancer, loadCache); + this.statisticsService = new StatisticsService(loadCache); this.eventService = new EventService(configManager, loadCache); } @@ -132,10 +132,6 @@ public class LoadManager { partitionBalancer.reBalanceDataPartitionPolicy(database); } - public void broadcastLatestRegionRouteMap() { - statisticsService.broadcastLatestRegionRouteMap(); - } - public void startLoadServices() { loadCache.initHeartbeatCache(configManager); heartbeatService.startHeartbeatService(); @@ -327,14 +323,16 @@ public class LoadManager { /** * Force update the specified RegionGroups' cache. * - * @param regionGroupId Specified RegionGroupId - * @param heartbeatSampleMap Specified RegionHeartbeatSampleMap + * @param heartbeatSampleMap Map<RegionGroupId, Map<DataNodeId, RegionHeartbeatSample>> */ public void forceUpdateRegionGroupCache( - Map<TConsensusGroupId, Map<Integer, RegionHeartbeatSample>> heartbeatSampleMap) { + Map<TConsensusGroupId, Map<Integer, RegionHeartbeatSample>> heartbeatSampleMap) { heartbeatSampleMap.forEach( - (regionGroupId, regionHeartbeatSampleMap) -> regionHeartbeatSampleMap.forEach((dataNodeId, regionHeartbeatSample) -> - loadCache.cacheRegionHeartbeatSample(regionGroupId, dataNodeId, regionHeartbeatSample))); + (regionGroupId, regionHeartbeatSampleMap) -> + regionHeartbeatSampleMap.forEach( + (dataNodeId, regionHeartbeatSample) -> + loadCache.cacheRegionHeartbeatSample( + regionGroupId, dataNodeId, regionHeartbeatSample))); loadCache.updateRegionGroupStatistics(); eventService.checkAndBroadcastRegionGroupStatisticsChangeEventIfNecessary(); } @@ -359,7 +357,7 @@ public class LoadManager { * @return Map<RegionGroupId, RegionPriority>. */ public Map<TConsensusGroupId, TRegionReplicaSet> getRegionPriorityMap() { - return loadCache.getRegionPriorityMap(); + return routeBalancer.getRegionPriorityMap(); } /** @@ -391,42 +389,18 @@ public class LoadManager { } /** - * Force update the specified RegionGroup's leader. - * - * @param regionGroupId Specified RegionGroupId - * @param leaderId Leader DataNodeId - */ - public void forceUpdateRegionLeader(TConsensusGroupId regionGroupId, int leaderId) { - loadCache.forceUpdateRegionLeader(regionGroupId, leaderId); - } - - /** - * Force update the specified RegionGroup's priority. + * Force update the specified Consensus' cache. * - * @param regionGroupId Specified RegionGroupId - * @param regionPriority Region route priority + * @param heartbeatSampleMap Map<RegionGroupId, ConsensusHeartbeatSample> */ - public void forceUpdateRegionPriority( - TConsensusGroupId regionGroupId, TRegionReplicaSet regionPriority) { - loadCache.forceUpdateRegionPriority(regionGroupId, regionPriority); + public void forceUpdateConsensusCache( + Map<TConsensusGroupId, ConsensusHeartbeatSample> heartbeatSampleMap) { + heartbeatSampleMap.forEach(loadCache::cacheConsensusSample); + loadCache.updateConsensusStatistics(); + eventService.checkAndBroadcastConsensusStatisticsChangeEventIfNecessary(); } - /** - * Remove the specified RegionGroup's route cache. - * - * @param regionGroupId Specified RegionGroupId - */ - public void removeRegionRouteCache(TConsensusGroupId regionGroupId) { - loadCache.removeRegionRouteCache(regionGroupId); - } - - /** Force balance the region leader and broadcast RouteChangeEvent if necessary. */ - public void forceBalanceRegionLeader() { - Map<TConsensusGroupId, Pair<Integer, Integer>> differentRegionLeaderMap = - routeBalancer.balanceRegionLeader(); - Map<TConsensusGroupId, Pair<TRegionReplicaSet, TRegionReplicaSet>> differentRegionPriorityMap = - routeBalancer.balanceRegionPriority(); - statisticsService.broadcastRouteChangeEventIfNecessary( - differentRegionLeaderMap, differentRegionPriorityMap); + public LoadCache getLoadCache() { + return loadCache; } } 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 67b1c342b29..543c9a21a4c 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 @@ -38,10 +38,16 @@ import org.apache.iotdb.confignode.manager.load.balancer.router.leader.MinCostFl import org.apache.iotdb.confignode.manager.load.balancer.router.priority.GreedyPriorityBalancer; import org.apache.iotdb.confignode.manager.load.balancer.router.priority.IPriorityBalancer; import org.apache.iotdb.confignode.manager.load.balancer.router.priority.LeaderPriorityBalancer; +import org.apache.iotdb.confignode.manager.load.cache.consensus.ConsensusHeartbeatSample; +import org.apache.iotdb.confignode.manager.load.subscriber.ConsensusStatisticsChangeEvent; +import org.apache.iotdb.confignode.manager.load.subscriber.IClusterStatusSubscriber; +import org.apache.iotdb.confignode.manager.load.subscriber.NodeStatisticsChangeEvent; +import org.apache.iotdb.confignode.manager.load.subscriber.RegionGroupStatisticsChangeEvent; import org.apache.iotdb.confignode.manager.node.NodeManager; import org.apache.iotdb.confignode.manager.partition.PartitionManager; import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.mpp.rpc.thrift.TRegionLeaderChangeReq; +import org.apache.iotdb.mpp.rpc.thrift.TRegionRouteReq; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.tsfile.utils.Pair; @@ -49,28 +55,25 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; +import java.util.Objects; +import java.util.TreeMap; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; /** * The RouteBalancer will maintain cluster RegionRouteMap, which contains: * - * <p>1. regionLeaderMap, record the leader for each RegionGroup - * * <p>2. regionPriorityMap, record the priority for read/write requests in each RegionGroup */ -public class RouteBalancer { +public class RouteBalancer implements IClusterStatusSubscriber { private static final Logger LOGGER = LoggerFactory.getLogger(RouteBalancer.class); - private static final ConfigNodeConfig CONF = ConfigNodeDescriptor.getInstance().getConf(); - private static final String SCHEMA_REGION_CONSENSUS_PROTOCOL_CLASS = CONF.getSchemaRegionConsensusProtocolClass(); private static final String DATA_REGION_CONSENSUS_PROTOCOL_CLASS = CONF.getDataRegionConsensusProtocolClass(); - private static final boolean IS_ENABLE_AUTO_LEADER_BALANCE_FOR_DATA_REGION = (CONF.isEnableAutoLeaderBalanceForRatisConsensus() && ConsensusFactory.RATIS_CONSENSUS.equals(DATA_REGION_CONSENSUS_PROTOCOL_CLASS)) @@ -86,22 +89,25 @@ public class RouteBalancer { // The simple consensus protocol will always automatically designate itself as the leader || ConsensusFactory.SIMPLE_CONSENSUS.equals(SCHEMA_REGION_CONSENSUS_PROTOCOL_CLASS); - // The interval of retrying to balance ratis leader after the last failed time - private static final long BALANCE_RATIS_LEADER_FAILED_INTERVAL = 60 * 1000L; - private final IManager configManager; - /** RegionRouteMap */ - // For generating optimal RegionLeaderMap + // For generating optimal Region leader distribution private final ILeaderBalancer leaderBalancer; - // For generating optimal RegionPriorityMap + // For generating optimal cluster Region routing priority private final IPriorityBalancer priorityRouter; + // Map<RegionGroupId, Region priority> + // The client requests are preferentially routed to the Region with the lowest index in the + // TRegionReplicaSet + private final Map<TConsensusGroupId, TRegionReplicaSet> regionPriorityMap; - private long lastFailedTimeForBalanceRatisSchemaLeader = 0; - private long lastFailedTimeForBalanceRatisDataLeader = 0; + // The interval of retrying to balance ratis leader after the last failed time + private static final long BALANCE_RATIS_LEADER_FAILED_INTERVAL = 60 * 1000L; + private final Map<TConsensusGroupId, Long> lastFailedTimeForLeaderBalance; public RouteBalancer(IManager configManager) { this.configManager = configManager; + this.regionPriorityMap = new TreeMap<>(); + this.lastFailedTimeForLeaderBalance = new TreeMap<>(); switch (CONF.getLeaderDistributionPolicy()) { case ILeaderBalancer.GREEDY_POLICY: @@ -124,36 +130,23 @@ public class RouteBalancer { } } - /** - * Balance cluster RegionGroup leader distribution through configured algorithm TODO: @YongzaoDan, - * increase scheduling delay - * - * @return Map<RegionGroupId, Pair<old leader index, new leader index>> - */ - public synchronized Map<TConsensusGroupId, Pair<Integer, Integer>> balanceRegionLeader() { - Map<TConsensusGroupId, Pair<Integer, Integer>> differentRegionLeaderMap = - new ConcurrentHashMap<>(); - if (IS_ENABLE_AUTO_LEADER_BALANCE_FOR_SCHEMA_REGION - && (!ConsensusFactory.RATIS_CONSENSUS.equals(SCHEMA_REGION_CONSENSUS_PROTOCOL_CLASS) - || System.currentTimeMillis() - lastFailedTimeForBalanceRatisSchemaLeader - > BALANCE_RATIS_LEADER_FAILED_INTERVAL)) { - differentRegionLeaderMap.putAll(balanceRegionLeader(TConsensusGroupType.SchemaRegion)); + private synchronized void balanceClusterRoutingPolicy() { + balanceRegionLeader(); + balanceRegionPriority(); + } + + /** Balance cluster RegionGroup leader distribution through configured algorithm. */ + private void balanceRegionLeader() { + if (IS_ENABLE_AUTO_LEADER_BALANCE_FOR_SCHEMA_REGION) { + balanceRegionLeader(TConsensusGroupType.SchemaRegion, SCHEMA_REGION_CONSENSUS_PROTOCOL_CLASS); } - if (IS_ENABLE_AUTO_LEADER_BALANCE_FOR_DATA_REGION - && (!ConsensusFactory.RATIS_CONSENSUS.equals(DATA_REGION_CONSENSUS_PROTOCOL_CLASS) - || System.currentTimeMillis() - lastFailedTimeForBalanceRatisDataLeader - > BALANCE_RATIS_LEADER_FAILED_INTERVAL)) { - differentRegionLeaderMap.putAll(balanceRegionLeader(TConsensusGroupType.DataRegion)); + if (IS_ENABLE_AUTO_LEADER_BALANCE_FOR_DATA_REGION) { + balanceRegionLeader(TConsensusGroupType.DataRegion, DATA_REGION_CONSENSUS_PROTOCOL_CLASS); } - - return differentRegionLeaderMap; } - private Map<TConsensusGroupId, Pair<Integer, Integer>> balanceRegionLeader( - TConsensusGroupType regionGroupType) { - Map<TConsensusGroupId, Pair<Integer, Integer>> differentRegionLeaderMap = - new ConcurrentHashMap<>(); - + private void balanceRegionLeader( + TConsensusGroupType regionGroupType, String consensusProtocolClass) { // Collect the latest data and generate the optimal leader distribution Map<TConsensusGroupId, Integer> currentLeaderMap = getLoadManager().getRegionLeaderMap(); Map<TConsensusGroupId, Integer> optimalLeaderMap = @@ -170,35 +163,59 @@ public class RouteBalancer { .collect(Collectors.toSet())); // Transfer leader to the optimal distribution + long currentTime = System.nanoTime(); AtomicInteger requestId = new AtomicInteger(0); AsyncClientHandler<TRegionLeaderChangeReq, TSStatus> clientHandler = new AsyncClientHandler<>(DataNodeRequestType.CHANGE_REGION_LEADER); + Map<TConsensusGroupId, ConsensusHeartbeatSample> successTransferMap = new TreeMap<>(); optimalLeaderMap.forEach( (regionGroupId, newLeaderId) -> { + if (ConsensusFactory.RATIS_CONSENSUS.equals(consensusProtocolClass) + && currentTime - lastFailedTimeForLeaderBalance.getOrDefault(regionGroupId, 0L) + > BALANCE_RATIS_LEADER_FAILED_INTERVAL) { + return; + } + if (newLeaderId != -1 && !newLeaderId.equals(currentLeaderMap.get(regionGroupId))) { - String consensusProtocolClass; - switch (regionGroupId.getType()) { - case SchemaRegion: - consensusProtocolClass = SCHEMA_REGION_CONSENSUS_PROTOCOL_CLASS; - break; - case DataRegion: - default: - consensusProtocolClass = DATA_REGION_CONSENSUS_PROTOCOL_CLASS; - break; - } LOGGER.info( "[LeaderBalancer] Try to change the leader of Region: {} to DataNode: {} ", regionGroupId, newLeaderId); - changeRegionLeader( - consensusProtocolClass, - requestId, - clientHandler, - regionGroupId, - getNodeManager().getRegisteredDataNode(newLeaderId).getLocation(), - regionGroupType); - differentRegionLeaderMap.put( - regionGroupId, new Pair<>(currentLeaderMap.get(regionGroupId), newLeaderId)); + switch (consensusProtocolClass) { + case ConsensusFactory.IOT_CONSENSUS: + case ConsensusFactory.SIMPLE_CONSENSUS: + // For IoTConsensus or SimpleConsensus protocol, change RegionRouteMap is enough + successTransferMap.put( + regionGroupId, new ConsensusHeartbeatSample(currentTime, newLeaderId)); + break; + case ConsensusFactory.RATIS_CONSENSUS: + default: + // For ratis protocol, the ConfigNode-leader will send a changeLeaderRequest to the + // new + // leader. + // And the RegionRouteMap will be updated by Cluster-Heartbeat-Service later if + // change + // leader success. + // Force update region leader for ratis consensus when replication factor is 1. + if (TConsensusGroupType.SchemaRegion.equals(regionGroupType) + && CONF.getSchemaReplicationFactor() == 1) { + successTransferMap.put( + regionGroupId, new ConsensusHeartbeatSample(0, newLeaderId)); + } else if (TConsensusGroupType.DataRegion.equals(regionGroupType) + && CONF.getDataReplicationFactor() == 1) { + successTransferMap.put( + regionGroupId, new ConsensusHeartbeatSample(0, newLeaderId)); + } else { + TDataNodeLocation newLeader = + getNodeManager().getRegisteredDataNode(newLeaderId).getLocation(); + TRegionLeaderChangeReq regionLeaderChangeReq = + new TRegionLeaderChangeReq(regionGroupId, newLeader); + int requestIndex = requestId.getAndIncrement(); + clientHandler.putRequest(requestIndex, regionLeaderChangeReq); + clientHandler.putDataNodeLocation(requestIndex, newLeader); + } + break; + } } }); if (requestId.get() > 0) { @@ -207,20 +224,13 @@ public class RouteBalancer { for (int i = 0; i < requestId.get(); i++) { if (clientHandler.getResponseMap().get(i).getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - getLoadManager() - .forceUpdateRegionLeader( - clientHandler.getRequest(i).getRegionId(), - clientHandler.getRequest(i).getNewLeaderNode().getDataNodeId()); + successTransferMap.put( + clientHandler.getRequest(i).getRegionId(), + new ConsensusHeartbeatSample( + currentTime, clientHandler.getRequest(i).getNewLeaderNode().getDataNodeId())); } else { - differentRegionLeaderMap.remove(clientHandler.getRequest(i).getRegionId()); - if (TConsensusGroupType.SchemaRegion.equals(regionGroupType) - && ConsensusFactory.RATIS_CONSENSUS.equals(SCHEMA_REGION_CONSENSUS_PROTOCOL_CLASS)) { - lastFailedTimeForBalanceRatisSchemaLeader = System.currentTimeMillis(); - } - if (TConsensusGroupType.DataRegion.equals(regionGroupType) - && ConsensusFactory.RATIS_CONSENSUS.equals(DATA_REGION_CONSENSUS_PROTOCOL_CLASS)) { - lastFailedTimeForBalanceRatisDataLeader = System.currentTimeMillis(); - } + lastFailedTimeForLeaderBalance.put( + clientHandler.getRequest(i).getRegionId(), currentTime); LOGGER.error( "[LeaderBalancer] Failed to change the leader of Region: {} to DataNode: {}", clientHandler.getRequest(i).getRegionId(), @@ -228,58 +238,11 @@ public class RouteBalancer { } } } - return differentRegionLeaderMap; + getLoadManager().forceUpdateConsensusCache(successTransferMap); } - private void changeRegionLeader( - String consensusProtocolClass, - AtomicInteger requestId, - AsyncClientHandler<TRegionLeaderChangeReq, TSStatus> clientHandler, - TConsensusGroupId regionGroupId, - TDataNodeLocation newLeader, - TConsensusGroupType regionGroupType) { - switch (consensusProtocolClass) { - case ConsensusFactory.IOT_CONSENSUS: - case ConsensusFactory.SIMPLE_CONSENSUS: - // For IoTConsensus or SimpleConsensus protocol, change RegionRouteMap is enough. - // And the result will be broadcast by Cluster-LoadStatistics-Service soon. - getLoadManager().forceUpdateRegionLeader(regionGroupId, newLeader.getDataNodeId()); - break; - case ConsensusFactory.RATIS_CONSENSUS: - default: - // For ratis protocol, the ConfigNode-leader will send a changeLeaderRequest to the new - // leader. - // And the RegionRouteMap will be updated by Cluster-Heartbeat-Service later if change - // leader success. - // Force update region leader for ratis consensus when replication factor is 1. - if (TConsensusGroupType.SchemaRegion.equals(regionGroupType) - && CONF.getSchemaReplicationFactor() == 1) { - getLoadManager().forceUpdateRegionLeader(regionGroupId, newLeader.getDataNodeId()); - } else if (TConsensusGroupType.DataRegion.equals(regionGroupType) - && CONF.getDataReplicationFactor() == 1) { - getLoadManager().forceUpdateRegionLeader(regionGroupId, newLeader.getDataNodeId()); - } else { - TRegionLeaderChangeReq regionLeaderChangeReq = - new TRegionLeaderChangeReq(regionGroupId, newLeader); - int requestIndex = requestId.getAndIncrement(); - clientHandler.putRequest(requestIndex, regionLeaderChangeReq); - clientHandler.putDataNodeLocation(requestIndex, newLeader); - } - break; - } - } - - /** - * Balance cluster RegionGroup route priority through configured algorithm TODO: @YongzaoDan, - * increase scheduling delay - * - * @return Map<RegionGroupId, Pair<old route priority, new route priority>> - */ - public synchronized Map<TConsensusGroupId, Pair<TRegionReplicaSet, TRegionReplicaSet>> - balanceRegionPriority() { - - Map<TConsensusGroupId, TRegionReplicaSet> currentPriorityMap = - getLoadManager().getRegionPriorityMap(); + /** Balance cluster RegionGroup route priority through configured algorithm. */ + private void balanceRegionPriority() { Map<TConsensusGroupId, Integer> regionLeaderMap = getLoadManager().getRegionLeaderMap(); Map<Integer, Long> dataNodeLoadScoreMap = getLoadManager().getAllDataNodeLoadScores(); @@ -296,20 +259,72 @@ public class RouteBalancer { regionLeaderMap, dataNodeLoadScoreMap)); - Map<TConsensusGroupId, Pair<TRegionReplicaSet, TRegionReplicaSet>> differentRegionPriorityMap = - new ConcurrentHashMap<>(); - for (Map.Entry<TConsensusGroupId, TRegionReplicaSet> regionPriorityEntry : - optimalRegionPriorityMap.entrySet()) { - TConsensusGroupId regionGroupId = regionPriorityEntry.getKey(); - TRegionReplicaSet optimalRegionPriority = regionPriorityEntry.getValue(); - if (!optimalRegionPriority.equals(currentPriorityMap.get(regionGroupId))) { - differentRegionPriorityMap.put( - regionGroupId, - new Pair<>(currentPriorityMap.get(regionGroupId), optimalRegionPriority)); - getLoadManager().forceUpdateRegionPriority(regionGroupId, optimalRegionPriority); + AtomicBoolean needBroadcast = new AtomicBoolean(false); + Map<TConsensusGroupId, Pair<TRegionReplicaSet, TRegionReplicaSet>> differentPriorityMap = + new TreeMap<>(); + optimalRegionPriorityMap.forEach( + (regionGroupId, optimalRegionPriority) -> { + TRegionReplicaSet currentRegionPriority = regionPriorityMap.get(regionGroupId); + if (!optimalRegionPriority.equals(currentRegionPriority)) { + differentPriorityMap.put( + regionGroupId, new Pair<>(currentRegionPriority, optimalRegionPriority)); + needBroadcast.set(true); + } + }); + if (needBroadcast.get()) { + broadcastLatestRegionPriorityMap(); + recordRegionPriorityMap(differentPriorityMap); + } + } + + private void broadcastLatestRegionPriorityMap() { + // Broadcast the RegionRouteMap to all DataNodes except the unknown ones + Map<Integer, TDataNodeLocation> dataNodeLocationMap = + getNodeManager() + .filterDataNodeThroughStatus( + NodeStatus.Running, NodeStatus.Removing, NodeStatus.ReadOnly) + .stream() + .map(TDataNodeConfiguration::getLocation) + .collect(Collectors.toMap(TDataNodeLocation::getDataNodeId, location -> location)); + + long broadcastTime = System.currentTimeMillis(); + AsyncClientHandler<TRegionRouteReq, TSStatus> clientHandler = + new AsyncClientHandler<>( + DataNodeRequestType.UPDATE_REGION_ROUTE_MAP, + new TRegionRouteReq(broadcastTime, regionPriorityMap), + dataNodeLocationMap); + AsyncDataNodeClientPool.getInstance().sendAsyncRequestToDataNodeWithRetry(clientHandler); + } + + private void recordRegionPriorityMap( + Map<TConsensusGroupId, Pair<TRegionReplicaSet, TRegionReplicaSet>> differentPriorityMap) { + LOGGER.info("[RegionPriority] RegionPriorityMap: "); + for (Map.Entry<TConsensusGroupId, Pair<TRegionReplicaSet, TRegionReplicaSet>> + regionPriorityEntry : differentPriorityMap.entrySet()) { + if (!Objects.equals( + regionPriorityEntry.getValue().getRight(), regionPriorityEntry.getValue().getLeft())) { + try { + LOGGER.info( + "[RegionPriority]\t {}: {}->{}", + regionPriorityEntry.getKey(), + regionPriorityEntry.getValue().getLeft() == null + ? "null" + : regionPriorityEntry.getValue().getLeft().getDataNodeLocations().stream() + .map(TDataNodeLocation::getDataNodeId) + .collect(Collectors.toList()), + regionPriorityEntry.getValue().getRight().getDataNodeLocations().stream() + .map(TDataNodeLocation::getDataNodeId) + .collect(Collectors.toList())); + } catch (Exception e) { + LOGGER.error("Unexpected exception", e); + } } } - return differentRegionPriorityMap; + } + + /** @return Map<RegionGroupId, RegionPriority> */ + public Map<TConsensusGroupId, TRegionReplicaSet> getRegionPriorityMap() { + return regionPriorityMap; } private NodeManager getNodeManager() { @@ -323,4 +338,19 @@ public class RouteBalancer { private LoadManager getLoadManager() { return configManager.getLoadManager(); } + + @Override + public void onNodeStatisticsChanged(NodeStatisticsChangeEvent event) { + balanceClusterRoutingPolicy(); + } + + @Override + public void onRegionGroupStatisticsChanged(RegionGroupStatisticsChangeEvent event) { + balanceClusterRoutingPolicy(); + } + + @Override + public void onConsensusStatisticsChanged(ConsensusStatisticsChangeEvent event) { + balanceClusterRoutingPolicy(); + } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/AbstractLoadCache.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/AbstractLoadCache.java index c9e8ac8b2a0..125ec65e925 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/AbstractLoadCache.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/AbstractLoadCache.java @@ -28,6 +28,8 @@ public abstract class AbstractLoadCache { // Max heartbeat cache samples store size private static final int MAXIMUM_WINDOW_SIZE = 100; + // The Status will be set to Unknown when the response time of heartbeat is more than 20s + protected static final long HEARTBEAT_TIMEOUT_TIME_IN_NS = 20_000_000_000L; // Caching the recent MAXIMUM_WINDOW_SIZE heartbeat sample protected final List<AbstractHeartbeatSample> slidingWindow; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/LoadCache.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/LoadCache.java index eba652b4714..4f7cd8b8b89 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/LoadCache.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/LoadCache.java @@ -41,14 +41,12 @@ import org.apache.iotdb.confignode.manager.load.cache.region.RegionGroupCache; import org.apache.iotdb.confignode.manager.load.cache.region.RegionGroupStatistics; import org.apache.iotdb.confignode.manager.load.cache.region.RegionHeartbeatSample; import org.apache.iotdb.confignode.manager.partition.RegionGroupStatus; -import org.apache.iotdb.tsfile.utils.Pair; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Arrays; import java.util.Comparator; -import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Objects; @@ -142,7 +140,7 @@ public class LoadCache { regionReplicaSet -> { TConsensusGroupId consensusGroupId = regionReplicaSet.getRegionId(); regionGroupCacheMap.put(consensusGroupId, new RegionGroupCache()); - consensusCacheMap.put(consensusGroupId, new ConsensusCache(consensusGroupId)); + consensusCacheMap.put(consensusGroupId, new ConsensusCache()); }); } @@ -214,8 +212,8 @@ public class LoadCache { public void cacheConsensusSample( TConsensusGroupId regionGroupId, ConsensusHeartbeatSample sample) { consensusCacheMap - .computeIfAbsent(regionGroupId, empty -> new ConsensusCache(regionGroupId)) - .cacheConsensusSample(sample); + .computeIfAbsent(regionGroupId, empty -> new ConsensusCache()) + .cacheHeartbeatSample(sample); } /** Update the NodeStatistics of all Nodes. */ @@ -233,20 +231,6 @@ public class LoadCache { consensusCacheMap.values().forEach(ConsensusCache::updateCurrentStatistics); } - public Map<TConsensusGroupId, Pair<Integer, Integer>> updateRegionGroupLeader() { - Map<TConsensusGroupId, Pair<Integer, Integer>> differentRegionGroupLeaderMap = new HashMap<>(); - consensusCacheMap.forEach( - (regionGroupId, consensusCache) -> { - int prevLeader = consensusCache.getLeaderId(); - if (consensusCache.periodicUpdate()) { - // Update and record the changed RegionGroupStatistics - differentRegionGroupLeaderMap.put( - regionGroupId, new Pair<>(prevLeader, consensusCache.getLeaderId())); - } - }); - return differentRegionGroupLeaderMap; - } - /** * Get the NodeStatistics of all Nodes. * @@ -255,7 +239,8 @@ public class LoadCache { public Map<Integer, NodeStatistics> getCurrentNodeStatisticsMap() { Map<Integer, NodeStatistics> nodeStatisticsMap = new TreeMap<>(); nodeCacheMap.forEach( - (nodeId, nodeCache) -> nodeStatisticsMap.put(nodeId, nodeCache.getStatistics())); + (nodeId, nodeCache) -> + nodeStatisticsMap.put(nodeId, (NodeStatistics) nodeCache.getCurrentStatistics())); return nodeStatisticsMap; } @@ -505,6 +490,7 @@ public class LoadCache { /** Remove the specified RegionGroup's cache. */ public void removeRegionGroupCache(TConsensusGroupId consensusGroupId) { regionGroupCacheMap.remove(consensusGroupId); + consensusCacheMap.remove(consensusGroupId); } /** @@ -520,22 +506,6 @@ public class LoadCache { return regionLeaderMap; } - /** - * Safely get the latest RegionPriorityMap. - * - * @return Map<RegionGroupId, RegionPriority> - */ - public Map<TConsensusGroupId, TRegionReplicaSet> getRegionPriorityMap() { - Map<TConsensusGroupId, TRegionReplicaSet> regionPriorityMap = new ConcurrentHashMap<>(); - consensusCacheMap.forEach( - (regionGroupId, consensusCache) -> { - if (!ConsensusCache.UN_READY_REGION_PRIORITY.equals(consensusCache.getRegionPriority())) { - regionPriorityMap.put(regionGroupId, consensusCache.getRegionPriority()); - } - }); - return regionPriorityMap; - } - /** * Wait for the specified RegionGroups to finish leader election. * @@ -549,7 +519,7 @@ public class LoadCache { regionGroupIds.forEach( regionGroupId -> { if (!consensusCacheMap.containsKey(regionGroupId) - || consensusCacheMap.get(regionGroupId).isRegionGroupUnready()) { + || consensusCacheMap.get(regionGroupId).isLeaderUnSelected()) { allRegionLeaderElected.set(false); } }); @@ -571,44 +541,6 @@ public class LoadCache { regionGroupIds); } - /** - * Force update the specified RegionGroup's leader. - * - * @param regionGroupId Specified RegionGroupId - * @param leaderId Leader DataNodeId - */ - public void forceUpdateRegionLeader(TConsensusGroupId regionGroupId, int leaderId) { - consensusCacheMap - .computeIfAbsent(regionGroupId, empty -> new ConsensusCache(regionGroupId)) - .forceUpdateRegionLeader(leaderId); - } - - /** - * Force update the specified RegionGroup's priority. - * - * @param regionGroupId Specified RegionGroupId - * @param regionPriority Region route priority - */ - public void forceUpdateRegionPriority( - TConsensusGroupId regionGroupId, TRegionReplicaSet regionPriority) { - consensusCacheMap - .computeIfAbsent(regionGroupId, empty -> new ConsensusCache(regionGroupId)) - .forceUpdateRegionPriority(regionPriority); - } - - /** - * Remove the specified RegionGroup's route cache. - * - * @param regionGroupId Specified RegionGroupId - */ - public void removeRegionRouteCache(TConsensusGroupId regionGroupId) { - consensusCacheMap.remove(regionGroupId); - } - - public boolean existUnreadyRegionGroup() { - return consensusCacheMap.values().stream().anyMatch(ConsensusCache::isRegionGroupUnready); - } - public void updateConfirmedConfigNodeEndPoints( int dataNodeId, Set<TEndPoint> configNodeEndPoints) { confirmedConfigNodeMap.put(dataNodeId, configNodeEndPoints); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/consensus/ConsensusCache.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/consensus/ConsensusCache.java index 514a06b069b..f4d4cdb6c96 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/consensus/ConsensusCache.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/consensus/ConsensusCache.java @@ -19,64 +19,24 @@ package org.apache.iotdb.confignode.manager.load.cache.consensus; -import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId; -import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet; -import org.apache.iotdb.confignode.conf.ConfigNodeConfig; -import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; import org.apache.iotdb.confignode.manager.load.cache.AbstractLoadCache; -import org.apache.iotdb.consensus.ConsensusFactory; public class ConsensusCache extends AbstractLoadCache { - private static final ConfigNodeConfig CONF = ConfigNodeDescriptor.getInstance().getConf(); - private static final String SCHEMA_REGION_CONSENSUS_PROTOCOL_CLASS = - CONF.getSchemaRegionConsensusProtocolClass(); - private static final String DATA_REGION_CONSENSUS_PROTOCOL_CLASS = - CONF.getDataRegionConsensusProtocolClass(); - public static final int UN_READY_LEADER_ID = -1; - public static final TRegionReplicaSet UN_READY_REGION_PRIORITY = new TRegionReplicaSet(); - - private final String consensusProtocolClass; - public ConsensusCache(TConsensusGroupId consensusGroupId) { - switch (consensusGroupId.getType()) { - case SchemaRegion: - this.consensusProtocolClass = SCHEMA_REGION_CONSENSUS_PROTOCOL_CLASS; - break; - case DataRegion: - default: - this.consensusProtocolClass = DATA_REGION_CONSENSUS_PROTOCOL_CLASS; - break; - } - } - - /** - * Cache the latest consensus sample of a RegionGroup. - * - * @param sample the latest heartbeat sample - */ - public synchronized void cacheConsensusSample(ConsensusHeartbeatSample sample) { - switch (consensusProtocolClass) { - case ConsensusFactory.RATIS_CONSENSUS: - // The leader of ratis consensus is self-elected - if (sample.getSampleLogicalTimestamp() - > this.lastConsensusSample.get().getSampleLogicalTimestamp()) { - this.lastConsensusSample.set(sample); - } - break; - case ConsensusFactory.SIMPLE_CONSENSUS: - case ConsensusFactory.IOT_CONSENSUS: - default: - // The leader of other consensus protocol is selected by ConfigNode-leader. - } + public ConsensusCache() { + super(); } + @Override public void updateCurrentStatistics() { - int currentLeader = lastConsensusSample.get().getLeaderId(); - if (currentLeader != UN_READY_LEADER_ID) { - currentStatistics.set( - new ConsensusStatistics(System.nanoTime(), lastConsensusSample.get().getLeaderId())); + ConsensusHeartbeatSample lastSample; + synchronized (slidingWindow) { + lastSample = (ConsensusHeartbeatSample) getLastSample(); + } + if (lastSample != null && lastSample.getLeaderId() != UN_READY_LEADER_ID) { + currentStatistics.set(new ConsensusStatistics(System.nanoTime(), lastSample.getLeaderId())); } } @@ -84,24 +44,11 @@ public class ConsensusCache extends AbstractLoadCache { return (ConsensusStatistics) currentStatistics.get(); } - /** - * Force update the specified RegionGroup's leader. - * - * @param leaderId Leader DataNodeId - */ - public void forceUpdateRegionLeader(int leaderId) { - cacheConsensusSample( - new ConsensusHeartbeatSample( - lastConsensusSample.get().getSampleLogicalTimestamp() + 1L, leaderId)); - this.leaderId.set(leaderId); - } - - public boolean isRegionGroupUnready() { - return UN_READY_LEADER_ID == leaderId.get() - || UN_READY_REGION_PRIORITY.equals(regionPriority.get()); + public boolean isLeaderUnSelected() { + return UN_READY_LEADER_ID == ((ConsensusStatistics) currentStatistics.get()).getLeaderId(); } public int getLeaderId() { - return leaderId.get(); + return ((ConsensusStatistics) currentStatistics.get()).getLeaderId(); } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/BaseNodeCache.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/BaseNodeCache.java index c9501a45135..a7a4552dce0 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/BaseNodeCache.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/BaseNodeCache.java @@ -25,15 +25,13 @@ import org.apache.iotdb.confignode.manager.load.cache.AbstractLoadCache; /** All the statistic interfaces that provided by HeartbeatCache. */ public abstract class BaseNodeCache extends AbstractLoadCache { - // When the response time of heartbeat is more than 20s, the Node is considered as down - public static final long HEARTBEAT_TIMEOUT_TIME_IN_NS = 20_000_000_000L; - protected final int nodeId; /** Constructor for NodeCache with default NodeStatistics. */ protected BaseNodeCache(int nodeId) { super(); this.nodeId = nodeId; + this.currentStatistics.set(NodeStatistics.generateDefaultNodeStatistics()); } public int getNodeId() { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/region/RegionCache.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/region/RegionCache.java index f353b49e735..b98546e6ac0 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/region/RegionCache.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/region/RegionCache.java @@ -20,53 +20,27 @@ package org.apache.iotdb.confignode.manager.load.cache.region; import org.apache.iotdb.commons.cluster.RegionStatus; +import org.apache.iotdb.confignode.manager.load.cache.AbstractLoadCache; -import java.util.Collections; -import java.util.LinkedList; -import java.util.List; -import java.util.concurrent.atomic.AtomicReference; - -import static org.apache.iotdb.confignode.manager.load.cache.node.BaseNodeCache.HEARTBEAT_TIMEOUT_TIME_IN_NS; -import static org.apache.iotdb.confignode.manager.load.cache.node.BaseNodeCache.MAXIMUM_WINDOW_SIZE; - -public class RegionCache { - - private final List<RegionHeartbeatSample> slidingWindow; - private final AtomicReference<RegionStatistics> currentStatistics; +public class RegionCache extends AbstractLoadCache { public RegionCache() { - this.slidingWindow = Collections.synchronizedList(new LinkedList<>()); - this.currentStatistics = - new AtomicReference<>(RegionStatistics.generateDefaultRegionStatistics()); - } - - public void cacheHeartbeatSample(RegionHeartbeatSample newHeartbeatSample) { - synchronized (slidingWindow) { - // Only sequential HeartbeatSamples are accepted. - // And un-sequential HeartbeatSamples will be discarded. - if (getLastSample() == null - || getLastSample().getSampleNanoTimestamp() - < newHeartbeatSample.getSampleNanoTimestamp()) { - slidingWindow.add(newHeartbeatSample); - } - - if (slidingWindow.size() > MAXIMUM_WINDOW_SIZE) { - slidingWindow.remove(0); - } - } + super(); + this.currentStatistics.set(RegionStatistics.generateDefaultRegionStatistics()); } + @Override public void updateCurrentStatistics() { RegionHeartbeatSample lastSample; synchronized (slidingWindow) { - lastSample = getLastSample(); + lastSample = (RegionHeartbeatSample) getLastSample(); } RegionStatus status; long currentNanoTime = System.nanoTime(); if (lastSample == null) { status = RegionStatus.Unknown; - } else if (currentNanoTime - lastSample.getSampleNanoTimestamp() + } else if (currentNanoTime - lastSample.getSampleLogicalTimestamp() > HEARTBEAT_TIMEOUT_TIME_IN_NS) { // TODO: Optimize Unknown judge logic status = RegionStatus.Unknown; @@ -77,10 +51,6 @@ public class RegionCache { } public RegionStatistics getCurrentStatistics() { - return currentStatistics.get(); - } - - private RegionHeartbeatSample getLastSample() { - return slidingWindow.isEmpty() ? null : slidingWindow.get(slidingWindow.size() - 1); + return (RegionStatistics) currentStatistics.get(); } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/EventService.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/EventService.java index 30682e49292..2667ec9f636 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/EventService.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/EventService.java @@ -20,16 +20,9 @@ package org.apache.iotdb.confignode.manager.load.service; import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId; -import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; -import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet; -import org.apache.iotdb.common.rpc.thrift.TSStatus; -import org.apache.iotdb.commons.cluster.NodeStatus; import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.commons.concurrent.ThreadName; import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil; -import org.apache.iotdb.confignode.client.DataNodeRequestType; -import org.apache.iotdb.confignode.client.async.AsyncDataNodeClientPool; -import org.apache.iotdb.confignode.client.async.handlers.AsyncClientHandler; import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; import org.apache.iotdb.confignode.manager.IManager; import org.apache.iotdb.confignode.manager.load.cache.LoadCache; @@ -37,7 +30,6 @@ import org.apache.iotdb.confignode.manager.load.cache.consensus.ConsensusStatist import org.apache.iotdb.confignode.manager.load.cache.node.NodeStatistics; import org.apache.iotdb.confignode.manager.load.cache.region.RegionGroupStatistics; import org.apache.iotdb.confignode.manager.load.subscriber.ConsensusStatisticsChangeEvent; -import org.apache.iotdb.mpp.rpc.thrift.TRegionRouteReq; import org.apache.iotdb.tsfile.utils.Pair; import com.google.common.eventbus.AsyncEventBus; @@ -49,7 +41,6 @@ import java.util.List; import java.util.Map; import java.util.Objects; import java.util.TreeMap; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -266,25 +257,4 @@ public class EventService { } } } - - private void broadcastLatestRegionRouteMap() { - Map<TConsensusGroupId, TRegionReplicaSet> regionPriorityMap = - getLoadManager().getRegionPriorityMap(); - Map<Integer, TDataNodeLocation> dataNodeLocationMap = new ConcurrentHashMap<>(); - // Broadcast the RegionRouteMap to all DataNodes except the unknown ones - getNodeManager() - .filterDataNodeThroughStatus(NodeStatus.Running, NodeStatus.Removing, NodeStatus.ReadOnly) - .forEach( - onlineDataNode -> - dataNodeLocationMap.put( - onlineDataNode.getLocation().getDataNodeId(), onlineDataNode.getLocation())); - - long broadcastTime = System.currentTimeMillis(); - AsyncClientHandler<TRegionRouteReq, TSStatus> clientHandler = - new AsyncClientHandler<>( - DataNodeRequestType.UPDATE_REGION_ROUTE_MAP, - new TRegionRouteReq(broadcastTime, regionPriorityMap), - dataNodeLocationMap); - AsyncDataNodeClientPool.getInstance().sendAsyncRequestToDataNodeWithRetry(clientHandler); - } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java index 1e6cc7d0907..3ba9489bacb 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java @@ -195,7 +195,8 @@ public class HeartbeatService { // Skip itself and the ConfigNode that is processing heartbeat continue; } - ConfigNodeHeartbeatHandler handler = new ConfigNodeHeartbeatHandler(configNodeId, loadCache); + ConfigNodeHeartbeatHandler handler = + new ConfigNodeHeartbeatHandler(configNodeId, configManager.getLoadManager()); AsyncConfigNodeHeartbeatClientPool.getInstance() .getConfigNodeHeartBeat(configNodeLocation.getInternalEndPoint(), heartbeatReq, handler); } @@ -218,7 +219,7 @@ public class HeartbeatService { DataNodeHeartbeatHandler handler = new DataNodeHeartbeatHandler( dataNodeId, - loadCache, + configManager.getLoadManager(), configManager.getClusterQuotaManager().getDeviceNum(), configManager.getClusterQuotaManager().getTimeSeriesNum(), configManager.getClusterQuotaManager().getRegionDisk(), diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/StatisticsService.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/StatisticsService.java index ae0bfbeb953..e57330a28da 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/StatisticsService.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/StatisticsService.java @@ -19,30 +19,18 @@ package org.apache.iotdb.confignode.manager.load.service; -import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId; -import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; -import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet; import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.commons.concurrent.ThreadName; import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil; import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; -import org.apache.iotdb.confignode.manager.IManager; -import org.apache.iotdb.confignode.manager.load.LoadManager; -import org.apache.iotdb.confignode.manager.load.balancer.RouteBalancer; import org.apache.iotdb.confignode.manager.load.cache.LoadCache; -import org.apache.iotdb.confignode.manager.load.subscriber.ConsensusStatisticsChangeEvent; -import org.apache.iotdb.confignode.manager.node.NodeManager; -import org.apache.iotdb.tsfile.utils.Pair; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.Map; -import java.util.Objects; import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -import java.util.stream.Collectors; public class StatisticsService { @@ -51,14 +39,9 @@ public class StatisticsService { private static final long HEARTBEAT_INTERVAL = ConfigNodeDescriptor.getInstance().getConf().getHeartbeatIntervalInMs(); - private final IManager configManager; - private final RouteBalancer routeBalancer; private final LoadCache loadCache; - public StatisticsService( - IManager configManager, RouteBalancer routeBalancer, LoadCache loadCache) { - this.configManager = configManager; - this.routeBalancer = routeBalancer; + public StatisticsService(LoadCache loadCache) { this.loadCache = loadCache; } @@ -102,54 +85,4 @@ public class StatisticsService { loadCache.updateRegionGroupStatistics(); loadCache.updateConsensusStatistics(); } - - public void broadcastRouteChangeEventIfNecessary( - Map<TConsensusGroupId, Pair<Integer, Integer>> differentRegionLeaderMap, - Map<TConsensusGroupId, Pair<TRegionReplicaSet, TRegionReplicaSet>> - differentRegionPriorityMap) { - if (containsChangeEvent(differentRegionLeaderMap) - || containsChangeEvent(differentRegionPriorityMap)) { - eventBus.post( - new ConsensusStatisticsChangeEvent(differentRegionLeaderMap, differentRegionPriorityMap)); - } - } - - private static <T> boolean containsChangeEvent(Map<TConsensusGroupId, Pair<T, T>> map) { - return !map.isEmpty() - && map.values().stream().anyMatch(pair -> !Objects.equals(pair.getLeft(), pair.getRight())); - } - - private void recordRegionPriorityMap( - Map<TConsensusGroupId, Pair<TRegionReplicaSet, TRegionReplicaSet>> priorityMap) { - LOGGER.info("[RegionPriority] RegionPriorityMap: "); - for (Map.Entry<TConsensusGroupId, Pair<TRegionReplicaSet, TRegionReplicaSet>> - regionPriorityEntry : priorityMap.entrySet()) { - if (!Objects.equals( - regionPriorityEntry.getValue().getRight(), regionPriorityEntry.getValue().getLeft())) { - try { - LOGGER.info( - "[RegionPriority]\t {}: {}->{}", - regionPriorityEntry.getKey(), - regionPriorityEntry.getValue().getLeft() == null - ? "null" - : regionPriorityEntry.getValue().getLeft().getDataNodeLocations().stream() - .map(TDataNodeLocation::getDataNodeId) - .collect(Collectors.toList()), - regionPriorityEntry.getValue().getRight().getDataNodeLocations().stream() - .map(TDataNodeLocation::getDataNodeId) - .collect(Collectors.toList())); - } catch (Exception e) { - LOGGER.error("unexcepted exception", e); - } - } - } - } - - private NodeManager getNodeManager() { - return configManager.getNodeManager(); - } - - private LoadManager getLoadManager() { - return configManager.getLoadManager(); - } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/subscriber/IClusterStatusSubscriber.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/subscriber/IClusterStatusSubscriber.java index 399dca7a0f3..778c9838a21 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/subscriber/IClusterStatusSubscriber.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/subscriber/IClusterStatusSubscriber.java @@ -24,6 +24,14 @@ import com.google.common.eventbus.Subscribe; public interface IClusterStatusSubscriber { + @Subscribe + @AllowConcurrentEvents + void onNodeStatisticsChanged(NodeStatisticsChangeEvent event); + + @Subscribe + @AllowConcurrentEvents + void onRegionGroupStatisticsChanged(RegionGroupStatisticsChangeEvent event); + @Subscribe @AllowConcurrentEvents void onConsensusStatisticsChanged(ConsensusStatisticsChangeEvent event); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/subscriber/IClusterStatusSubscriber.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/subscriber/NodeStatisticsChangeEvent.java similarity index 56% copy from iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/subscriber/IClusterStatusSubscriber.java copy to iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/subscriber/NodeStatisticsChangeEvent.java index 399dca7a0f3..04038c0f3cb 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/subscriber/IClusterStatusSubscriber.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/subscriber/NodeStatisticsChangeEvent.java @@ -19,12 +19,22 @@ package org.apache.iotdb.confignode.manager.load.subscriber; -import com.google.common.eventbus.AllowConcurrentEvents; -import com.google.common.eventbus.Subscribe; +import org.apache.iotdb.confignode.manager.load.cache.node.NodeStatistics; +import org.apache.iotdb.tsfile.utils.Pair; -public interface IClusterStatusSubscriber { +import java.util.Map; - @Subscribe - @AllowConcurrentEvents - void onConsensusStatisticsChanged(ConsensusStatisticsChangeEvent event); +public class NodeStatisticsChangeEvent { + + // Map<NodeId, Pair<old NodeStatistics, new NodeStatistics>> + private final Map<Integer, Pair<NodeStatistics, NodeStatistics>> differentNodeStatisticsMap; + + public NodeStatisticsChangeEvent( + Map<Integer, Pair<NodeStatistics, NodeStatistics>> differentNodeStatisticsMap) { + this.differentNodeStatisticsMap = differentNodeStatisticsMap; + } + + public Map<Integer, Pair<NodeStatistics, NodeStatistics>> getDifferentNodeStatisticsMap() { + return differentNodeStatisticsMap; + } } diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/ConsensusCacheTest.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/subscriber/RegionGroupStatisticsChangeEvent.java similarity index 51% copy from iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/ConsensusCacheTest.java copy to iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/subscriber/RegionGroupStatisticsChangeEvent.java index 5db15d7f768..3c02841d0fa 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/ConsensusCacheTest.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/subscriber/RegionGroupStatisticsChangeEvent.java @@ -16,25 +16,28 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.iotdb.confignode.manager.load.cache; + +package org.apache.iotdb.confignode.manager.load.subscriber; import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId; -import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType; -import org.apache.iotdb.confignode.manager.load.cache.consensus.ConsensusCache; -import org.apache.iotdb.confignode.manager.load.cache.consensus.ConsensusHeartbeatSample; +import org.apache.iotdb.confignode.manager.load.cache.region.RegionGroupStatistics; +import org.apache.iotdb.tsfile.utils.Pair; + +import java.util.Map; -import org.junit.Assert; -import org.junit.Test; +public class RegionGroupStatisticsChangeEvent { -public class ConsensusCacheTest { + private final Map<TConsensusGroupId, Pair<RegionGroupStatistics, RegionGroupStatistics>> + differentRegionGroupStatisticsMap; + + public RegionGroupStatisticsChangeEvent( + Map<TConsensusGroupId, Pair<RegionGroupStatistics, RegionGroupStatistics>> + differentRegionGroupStatisticsMap) { + this.differentRegionGroupStatisticsMap = differentRegionGroupStatisticsMap; + } - @Test - public void periodicUpdateTest() { - ConsensusCache consensusCache = - new ConsensusCache(new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 1)); - ConsensusHeartbeatSample sample = new ConsensusHeartbeatSample(1L, 1); - consensusCache.cacheConsensusSample(sample); - Assert.assertTrue(consensusCache.periodicUpdate()); - Assert.assertEquals(1, consensusCache.getLeaderId()); + public Map<TConsensusGroupId, Pair<RegionGroupStatistics, RegionGroupStatistics>> + getDifferentRegionGroupStatisticsMap() { + return differentRegionGroupStatisticsMap; } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeLeaderChangeHandler.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeLeaderChangeHandler.java index 3b0ce2c96a0..8fd8a4270c1 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeLeaderChangeHandler.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeLeaderChangeHandler.java @@ -27,6 +27,8 @@ import org.apache.iotdb.confignode.manager.ConfigManager; import org.apache.iotdb.confignode.manager.load.cache.consensus.ConsensusStatistics; import org.apache.iotdb.confignode.manager.load.subscriber.ConsensusStatisticsChangeEvent; import org.apache.iotdb.confignode.manager.load.subscriber.IClusterStatusSubscriber; +import org.apache.iotdb.confignode.manager.load.subscriber.NodeStatisticsChangeEvent; +import org.apache.iotdb.confignode.manager.load.subscriber.RegionGroupStatisticsChangeEvent; import org.apache.iotdb.tsfile.utils.Pair; import java.util.HashMap; @@ -55,6 +57,16 @@ public class PipeLeaderChangeHandler implements IClusterStatusSubscriber { onConsensusStatisticsChanged(new ConsensusStatisticsChangeEvent(virtualChangeMap)); } + @Override + public void onNodeStatisticsChanged(NodeStatisticsChangeEvent event) { + // Do nothing + } + + @Override + public void onRegionGroupStatisticsChanged(RegionGroupStatisticsChangeEvent event) { + // Do nothing + } + @Override public void onConsensusStatisticsChanged(ConsensusStatisticsChangeEvent event) { // If no pipe tasks, return diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeRuntimeCoordinator.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeRuntimeCoordinator.java index bc44fabc97e..d5bda3e98ec 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeRuntimeCoordinator.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeRuntimeCoordinator.java @@ -24,6 +24,8 @@ import org.apache.iotdb.commons.concurrent.ThreadName; import org.apache.iotdb.confignode.manager.ConfigManager; import org.apache.iotdb.confignode.manager.load.subscriber.ConsensusStatisticsChangeEvent; import org.apache.iotdb.confignode.manager.load.subscriber.IClusterStatusSubscriber; +import org.apache.iotdb.confignode.manager.load.subscriber.NodeStatisticsChangeEvent; +import org.apache.iotdb.confignode.manager.load.subscriber.RegionGroupStatisticsChangeEvent; import javax.validation.constraints.NotNull; @@ -68,6 +70,16 @@ public class PipeRuntimeCoordinator implements IClusterStatusSubscriber { pipeLeaderChangeHandler.onConfigRegionGroupLeaderChanged(); } + @Override + public void onNodeStatisticsChanged(NodeStatisticsChangeEvent event) { + // Do nothing + } + + @Override + public void onRegionGroupStatisticsChanged(RegionGroupStatisticsChangeEvent event) { + // Do nothing + } + @Override public synchronized void onConsensusStatisticsChanged(ConsensusStatisticsChangeEvent event) { pipeLeaderChangeHandler.onConsensusStatisticsChanged(event); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java index 9156ed4f635..aa7f5e8a59b 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java @@ -401,17 +401,19 @@ public class ConfigNodeProcedureEnv { NodeType.DataNode, dataNodeLocation.getDataNodeId(), new NodeHeartbeatSample(currentTime, NodeStatus.Removing)); + Map<TConsensusGroupId, Map<Integer, RegionHeartbeatSample>> removingHeartbeatSampleMap = + new TreeMap<>(); // Force update RegionStatus to Removing getPartitionManager() .getAllReplicaSets(dataNodeLocation.getDataNodeId()) .forEach( replicaSet -> - getLoadManager() - .forceUpdateRegionGroupCache( - replicaSet.getRegionId(), - Collections.singletonMap( - dataNodeLocation.getDataNodeId(), - new RegionHeartbeatSample(currentTime, RegionStatus.Removing)))); + removingHeartbeatSampleMap.put( + replicaSet.getRegionId(), + Collections.singletonMap( + dataNodeLocation.getDataNodeId(), + new RegionHeartbeatSample(currentTime, RegionStatus.Removing)))); + getLoadManager().forceUpdateRegionGroupCache(removingHeartbeatSampleMap); } /** @@ -563,8 +565,6 @@ public class ConfigNodeProcedureEnv { public void activateRegionGroup( Map<TConsensusGroupId, Map<Integer, RegionHeartbeatSample>> activateRegionGroupMap) { getLoadManager().forceUpdateRegionGroupCache(activateRegionGroupMap); - // force balance region leader to skip waiting for leader election - getLoadManager().forceBalanceRegionLeader(); // Wait for leader election getLoadManager().waitForLeaderElection(new ArrayList<>(activateRegionGroupMap.keySet())); } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java index b9ccd7e3418..7ba1a5126ad 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java @@ -42,6 +42,7 @@ import org.apache.iotdb.confignode.consensus.request.write.partition.AddRegionLo import org.apache.iotdb.confignode.consensus.request.write.partition.RemoveRegionLocationPlan; import org.apache.iotdb.confignode.consensus.response.datanode.DataNodeToStatusResp; import org.apache.iotdb.confignode.manager.ConfigManager; +import org.apache.iotdb.confignode.manager.load.cache.consensus.ConsensusHeartbeatSample; import org.apache.iotdb.confignode.manager.partition.PartitionMetrics; import org.apache.iotdb.confignode.persistence.node.NodeInfo; import org.apache.iotdb.confignode.procedure.scheduler.LockQueue; @@ -409,9 +410,6 @@ public class RegionMaintainHandler { // Remove the RegionGroupCache of the regionId configManager.getLoadManager().removeRegionGroupCache(regionId); - - // Broadcast the latest RegionRouteMap when Region migration finished - configManager.getLoadManager().broadcastLatestRegionRouteMap(); } public void removeRegionLocation( @@ -430,9 +428,6 @@ public class RegionMaintainHandler { // Remove the RegionGroupCache of the regionId configManager.getLoadManager().removeRegionGroupCache(regionId); - - // Broadcast the latest RegionRouteMap when Region migration finished - configManager.getLoadManager().broadcastLatestRegionRouteMap(); } /** @@ -659,8 +654,11 @@ public class RegionMaintainHandler { if (newLeaderNode.isPresent()) { configManager .getLoadManager() - .forceUpdateRegionLeader(regionId, newLeaderNode.get().getDataNodeId()); - + .forceUpdateConsensusCache( + Collections.singletonMap( + regionId, + new ConsensusHeartbeatSample( + System.nanoTime(), newLeaderNode.get().getDataNodeId()))); LOGGER.info( "{}, Change region leader finished for IOT_CONSENSUS, regionId: {}, newLeaderNode: {}", REGION_MIGRATE_PROCESS, diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteDatabaseProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteDatabaseProcedure.java index 60ff28252b7..6fd0d2674b1 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteDatabaseProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteDatabaseProcedure.java @@ -124,9 +124,6 @@ public class DeleteDatabaseProcedure env.getConfigManager() .getLoadManager() .removeRegionGroupCache(regionReplicaSet.getRegionId()); - env.getConfigManager() - .getLoadManager() - .removeRegionRouteCache(regionReplicaSet.getRegionId()); if (regionReplicaSet .getRegionId() diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/GreedyPriorityTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/GreedyPriorityTest.java index 3247e8f5829..cd578aba1f9 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/GreedyPriorityTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/GreedyPriorityTest.java @@ -66,7 +66,7 @@ public class GreedyPriorityTest { nodeCacheMap.put(i, new DataNodeHeartbeatCache(i)); nodeCacheMap.get(i).cacheHeartbeatSample(new NodeHeartbeatSample(currentTimeNs, statuses[i])); } - nodeCacheMap.values().forEach(BaseNodeCache::periodicUpdate); + nodeCacheMap.values().forEach(BaseNodeCache::updateCurrentStatistics); /* Get the loadScoreMap */ Map<Integer, Long> loadScoreMap = new ConcurrentHashMap<>(); diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/LeaderPriorityBalancerTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/LeaderPriorityBalancerTest.java index 826526017ad..6f77d459603 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/LeaderPriorityBalancerTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/LeaderPriorityBalancerTest.java @@ -67,7 +67,7 @@ public class LeaderPriorityBalancerTest { .cacheHeartbeatSample(new NodeHeartbeatSample(currentTimeNs, NodeStatus.Running)); } } - nodeCacheMap.values().forEach(BaseNodeCache::periodicUpdate); + nodeCacheMap.values().forEach(BaseNodeCache::updateCurrentStatistics); // Get the loadScoreMap Map<Integer, Long> loadScoreMap = new ConcurrentHashMap<>(); diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/ConsensusCacheTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/ConsensusCacheTest.java index 5db15d7f768..af27e54adc1 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/ConsensusCacheTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/ConsensusCacheTest.java @@ -18,8 +18,6 @@ */ package org.apache.iotdb.confignode.manager.load.cache; -import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId; -import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType; import org.apache.iotdb.confignode.manager.load.cache.consensus.ConsensusCache; import org.apache.iotdb.confignode.manager.load.cache.consensus.ConsensusHeartbeatSample; @@ -30,11 +28,10 @@ public class ConsensusCacheTest { @Test public void periodicUpdateTest() { - ConsensusCache consensusCache = - new ConsensusCache(new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 1)); + ConsensusCache consensusCache = new ConsensusCache(); ConsensusHeartbeatSample sample = new ConsensusHeartbeatSample(1L, 1); - consensusCache.cacheConsensusSample(sample); - Assert.assertTrue(consensusCache.periodicUpdate()); + consensusCache.cacheHeartbeatSample(sample); + consensusCache.updateCurrentStatistics(); Assert.assertEquals(1, consensusCache.getLeaderId()); } } diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/NodeCacheTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/NodeCacheTest.java index 0697be86922..eef3e9b9b6d 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/NodeCacheTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/NodeCacheTest.java @@ -29,34 +29,13 @@ import org.junit.Test; public class NodeCacheTest { @Test - public void forceUpdateTest() { - DataNodeHeartbeatCache dataNodeHeartbeatCache = new DataNodeHeartbeatCache(1); - - // Test default - Assert.assertEquals(NodeStatus.Unknown, dataNodeHeartbeatCache.getNodeStatus()); - Assert.assertEquals(Long.MAX_VALUE, dataNodeHeartbeatCache.getLoadScore()); - - // Test force update to RunningStatus - long currentTime = System.nanoTime() - 2_000_000; - dataNodeHeartbeatCache.forceUpdate(new NodeHeartbeatSample(currentTime, NodeStatus.Running)); - Assert.assertEquals(NodeStatus.Running, dataNodeHeartbeatCache.getNodeStatus()); - Assert.assertEquals(0, dataNodeHeartbeatCache.getLoadScore()); - - // Test force update to ReadOnlyStatus - currentTime += 2000; - dataNodeHeartbeatCache.forceUpdate(new NodeHeartbeatSample(currentTime, NodeStatus.ReadOnly)); - Assert.assertEquals(NodeStatus.ReadOnly, dataNodeHeartbeatCache.getNodeStatus()); - Assert.assertEquals(Long.MAX_VALUE, dataNodeHeartbeatCache.getLoadScore()); - } - - @Test - public void periodicUpdateTest() { + public void updateStatisticsTest() { // Test DataNode heartbeat cache DataNodeHeartbeatCache dataNodeHeartbeatCache = new DataNodeHeartbeatCache(1); long currentTime = System.nanoTime(); dataNodeHeartbeatCache.cacheHeartbeatSample( new NodeHeartbeatSample(currentTime, NodeStatus.Running)); - Assert.assertTrue(dataNodeHeartbeatCache.periodicUpdate()); + dataNodeHeartbeatCache.updateCurrentStatistics(); Assert.assertEquals(NodeStatus.Running, dataNodeHeartbeatCache.getNodeStatus()); Assert.assertEquals(0, dataNodeHeartbeatCache.getLoadScore()); @@ -65,7 +44,7 @@ public class NodeCacheTest { currentTime = System.nanoTime(); configNodeHeartbeatCache.cacheHeartbeatSample( new NodeHeartbeatSample(currentTime, NodeStatus.Running)); - Assert.assertTrue(configNodeHeartbeatCache.periodicUpdate()); + configNodeHeartbeatCache.updateCurrentStatistics(); Assert.assertEquals(NodeStatus.Running, configNodeHeartbeatCache.getNodeStatus()); Assert.assertEquals(0, configNodeHeartbeatCache.getLoadScore()); } diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/RegionGroupCacheTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/RegionGroupCacheTest.java index 264bb520bf5..a961eb310e7 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/RegionGroupCacheTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/RegionGroupCacheTest.java @@ -18,8 +18,6 @@ */ package org.apache.iotdb.confignode.manager.load.cache; -import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId; -import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType; import org.apache.iotdb.commons.cluster.RegionStatus; import org.apache.iotdb.confignode.manager.load.cache.region.RegionGroupCache; import org.apache.iotdb.confignode.manager.load.cache.region.RegionHeartbeatSample; @@ -28,16 +26,12 @@ import org.apache.iotdb.confignode.manager.partition.RegionGroupStatus; import org.junit.Assert; import org.junit.Test; -import java.util.HashMap; -import java.util.Map; - public class RegionGroupCacheTest { @Test public void getRegionStatusTest() { long currentTime = System.nanoTime(); - RegionGroupCache regionGroupCache = - new RegionGroupCache(new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 1)); + RegionGroupCache regionGroupCache = new RegionGroupCache(); regionGroupCache.cacheHeartbeatSample( 0, new RegionHeartbeatSample(currentTime, RegionStatus.Running)); regionGroupCache.cacheHeartbeatSample( @@ -46,7 +40,7 @@ public class RegionGroupCacheTest { 2, new RegionHeartbeatSample(currentTime, RegionStatus.Removing)); regionGroupCache.cacheHeartbeatSample( 3, new RegionHeartbeatSample(currentTime, RegionStatus.ReadOnly)); - Assert.assertTrue(regionGroupCache.periodicUpdate()); + regionGroupCache.updateCurrentStatistics(); Assert.assertEquals( RegionStatus.Running, regionGroupCache.getCurrentStatistics().getRegionStatus(0)); @@ -61,84 +55,64 @@ public class RegionGroupCacheTest { @Test public void getRegionGroupStatusTest() { long currentTime = System.nanoTime(); - RegionGroupCache runningRegionGroup = - new RegionGroupCache(new TConsensusGroupId(TConsensusGroupType.DataRegion, 0)); + RegionGroupCache runningRegionGroup = new RegionGroupCache(); runningRegionGroup.cacheHeartbeatSample( 0, new RegionHeartbeatSample(currentTime, RegionStatus.Running)); runningRegionGroup.cacheHeartbeatSample( 1, new RegionHeartbeatSample(currentTime, RegionStatus.Running)); runningRegionGroup.cacheHeartbeatSample( 2, new RegionHeartbeatSample(currentTime, RegionStatus.Running)); - Assert.assertTrue(runningRegionGroup.periodicUpdate()); + runningRegionGroup.updateCurrentStatistics(); Assert.assertEquals( RegionGroupStatus.Running, runningRegionGroup.getCurrentStatistics().getRegionGroupStatus()); - RegionGroupCache availableRegionGroup = - new RegionGroupCache(new TConsensusGroupId(TConsensusGroupType.DataRegion, 1)); + RegionGroupCache availableRegionGroup = new RegionGroupCache(); availableRegionGroup.cacheHeartbeatSample( 0, new RegionHeartbeatSample(currentTime, RegionStatus.Running)); availableRegionGroup.cacheHeartbeatSample( 1, new RegionHeartbeatSample(currentTime, RegionStatus.Unknown)); availableRegionGroup.cacheHeartbeatSample( 2, new RegionHeartbeatSample(currentTime, RegionStatus.Running)); - Assert.assertTrue(availableRegionGroup.periodicUpdate()); + availableRegionGroup.updateCurrentStatistics(); Assert.assertEquals( RegionGroupStatus.Available, availableRegionGroup.getCurrentStatistics().getRegionGroupStatus()); - RegionGroupCache disabledRegionGroup0 = - new RegionGroupCache(new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 2)); + RegionGroupCache disabledRegionGroup0 = new RegionGroupCache(); disabledRegionGroup0.cacheHeartbeatSample( 0, new RegionHeartbeatSample(currentTime, RegionStatus.Running)); disabledRegionGroup0.cacheHeartbeatSample( 1, new RegionHeartbeatSample(currentTime, RegionStatus.ReadOnly)); disabledRegionGroup0.cacheHeartbeatSample( 2, new RegionHeartbeatSample(currentTime, RegionStatus.Running)); - Assert.assertTrue(disabledRegionGroup0.periodicUpdate()); + disabledRegionGroup0.updateCurrentStatistics(); Assert.assertEquals( RegionGroupStatus.Discouraged, disabledRegionGroup0.getCurrentStatistics().getRegionGroupStatus()); - RegionGroupCache disabledRegionGroup1 = - new RegionGroupCache(new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 3)); + RegionGroupCache disabledRegionGroup1 = new RegionGroupCache(); disabledRegionGroup1.cacheHeartbeatSample( 0, new RegionHeartbeatSample(currentTime, RegionStatus.Running)); disabledRegionGroup1.cacheHeartbeatSample( 1, new RegionHeartbeatSample(currentTime, RegionStatus.Unknown)); disabledRegionGroup1.cacheHeartbeatSample( 2, new RegionHeartbeatSample(currentTime, RegionStatus.Unknown)); - Assert.assertTrue(disabledRegionGroup1.periodicUpdate()); + disabledRegionGroup1.updateCurrentStatistics(); Assert.assertEquals( RegionGroupStatus.Disabled, disabledRegionGroup1.getCurrentStatistics().getRegionGroupStatus()); - RegionGroupCache disabledRegionGroup2 = - new RegionGroupCache(new TConsensusGroupId(TConsensusGroupType.DataRegion, 4)); + RegionGroupCache disabledRegionGroup2 = new RegionGroupCache(); disabledRegionGroup2.cacheHeartbeatSample( 0, new RegionHeartbeatSample(currentTime, RegionStatus.Running)); disabledRegionGroup2.cacheHeartbeatSample( 1, new RegionHeartbeatSample(currentTime, RegionStatus.Running)); disabledRegionGroup2.cacheHeartbeatSample( 2, new RegionHeartbeatSample(currentTime, RegionStatus.Removing)); - Assert.assertTrue(disabledRegionGroup2.periodicUpdate()); + disabledRegionGroup2.updateCurrentStatistics(); Assert.assertEquals( RegionGroupStatus.Disabled, disabledRegionGroup2.getCurrentStatistics().getRegionGroupStatus()); } - - @Test - public void forceUpdateTest() { - long currentTime = System.nanoTime(); - Map<Integer, RegionHeartbeatSample> heartbeatSampleMap = new HashMap<>(); - for (int i = 0; i < 3; i++) { - heartbeatSampleMap.put(i, new RegionHeartbeatSample(currentTime, RegionStatus.Running)); - } - - RegionGroupCache regionGroupCache = - new RegionGroupCache(new TConsensusGroupId(TConsensusGroupType.DataRegion, 10)); - regionGroupCache.forceUpdate(heartbeatSampleMap); - Assert.assertEquals( - RegionGroupStatus.Running, regionGroupCache.getCurrentStatistics().getRegionGroupStatus()); - } } diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/node/NodeStatisticsTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/node/NodeStatisticsTest.java deleted file mode 100644 index 6d08ff69104..00000000000 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/node/NodeStatisticsTest.java +++ /dev/null @@ -1,66 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.iotdb.confignode.persistence.node; - -import org.apache.iotdb.commons.cluster.NodeStatus; -import org.apache.iotdb.confignode.manager.load.cache.node.NodeStatistics; -import org.apache.iotdb.tsfile.utils.PublicBAOS; - -import org.junit.Assert; -import org.junit.Test; - -import java.io.ByteArrayInputStream; -import java.io.DataInputStream; -import java.io.DataOutputStream; -import java.io.IOException; -import java.nio.ByteBuffer; - -public class NodeStatisticsTest { - - @Test - public void NodeStatisticsSerDeTest() throws IOException { - NodeStatistics statistics0 = new NodeStatistics(20000331, NodeStatus.ReadOnly, "DiskFull"); - - // Deserialization from buffer - try (PublicBAOS byteArrayOutputStream = new PublicBAOS(); - DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) { - statistics0.serialize(outputStream); - ByteBuffer buffer = - ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size()); - - NodeStatistics statistics1 = new NodeStatistics(); - statistics1.deserialize(buffer); - Assert.assertEquals(statistics0, statistics1); - } - - // Deserialization from inputStream - try (PublicBAOS byteArrayOutputStream = new PublicBAOS(); - DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) { - statistics0.serialize(outputStream); - - ByteArrayInputStream byteArrayInputStream = - new ByteArrayInputStream(byteArrayOutputStream.toByteArray()); - DataInputStream inputStream = new DataInputStream(byteArrayInputStream); - - NodeStatistics statistics1 = new NodeStatistics(); - statistics1.deserialize(inputStream); - Assert.assertEquals(statistics0, statistics1); - } - } -} diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/partition/statistics/RegionGroupStatisticsTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/partition/statistics/RegionGroupStatisticsTest.java deleted file mode 100644 index 8c29cea2c29..00000000000 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/partition/statistics/RegionGroupStatisticsTest.java +++ /dev/null @@ -1,76 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.iotdb.confignode.persistence.partition.statistics; - -import org.apache.iotdb.commons.cluster.RegionStatus; -import org.apache.iotdb.confignode.manager.load.cache.region.RegionGroupStatistics; -import org.apache.iotdb.confignode.manager.load.cache.region.RegionStatistics; -import org.apache.iotdb.confignode.manager.partition.RegionGroupStatus; -import org.apache.iotdb.tsfile.utils.PublicBAOS; - -import org.junit.Assert; -import org.junit.Test; - -import java.io.ByteArrayInputStream; -import java.io.DataInputStream; -import java.io.DataOutputStream; -import java.io.IOException; -import java.nio.ByteBuffer; -import java.util.HashMap; -import java.util.Map; - -public class RegionGroupStatisticsTest { - - @Test - public void RegionGroupStatisticsSerDeTest() throws IOException { - Map<Integer, RegionStatistics> regionStatisticsMap = new HashMap<>(); - for (int i = 0; i < 3; i++) { - regionStatisticsMap.put(i, new RegionStatistics(RegionStatus.Unknown)); - } - - RegionGroupStatistics statistics0 = - new RegionGroupStatistics(RegionGroupStatus.Disabled, regionStatisticsMap); - - // Deserialization from byteBuffer - try (PublicBAOS byteArrayOutputStream = new PublicBAOS(); - DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) { - statistics0.serialize(outputStream); - ByteBuffer buffer = - ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size()); - - RegionGroupStatistics statistics1 = new RegionGroupStatistics(); - statistics1.deserialize(buffer); - Assert.assertEquals(statistics0, statistics1); - } - - // Deserialization from inputStream - try (PublicBAOS byteArrayOutputStream = new PublicBAOS(); - DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) { - statistics0.serialize(outputStream); - - ByteArrayInputStream byteArrayInputStream = - new ByteArrayInputStream(byteArrayOutputStream.toByteArray()); - DataInputStream inputStream = new DataInputStream(byteArrayInputStream); - - RegionGroupStatistics statistics1 = new RegionGroupStatistics(); - statistics1.deserialize(inputStream); - Assert.assertEquals(statistics0, statistics1); - } - } -} diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/partition/statistics/RegionStatisticsTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/partition/statistics/RegionStatisticsTest.java deleted file mode 100644 index 3d1940beeb1..00000000000 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/partition/statistics/RegionStatisticsTest.java +++ /dev/null @@ -1,66 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.iotdb.confignode.persistence.partition.statistics; - -import org.apache.iotdb.commons.cluster.RegionStatus; -import org.apache.iotdb.confignode.manager.load.cache.region.RegionStatistics; -import org.apache.iotdb.tsfile.utils.PublicBAOS; - -import org.junit.Assert; -import org.junit.Test; - -import java.io.ByteArrayInputStream; -import java.io.DataInputStream; -import java.io.DataOutputStream; -import java.io.IOException; -import java.nio.ByteBuffer; - -public class RegionStatisticsTest { - - @Test - public void RegionStatisticsSerDeTest1() throws IOException { - RegionStatistics statistics0 = new RegionStatistics(RegionStatus.Running); - - // Deserialization from byteBuffer - try (PublicBAOS byteArrayOutputStream = new PublicBAOS(); - DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) { - statistics0.serialize(outputStream); - ByteBuffer buffer = - ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size()); - - RegionStatistics statistics1 = new RegionStatistics(); - statistics1.deserialize(buffer); - Assert.assertEquals(statistics0, statistics1); - } - - // Deserialization from inputStream - try (PublicBAOS byteArrayOutputStream = new PublicBAOS(); - DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) { - statistics0.serialize(outputStream); - - ByteArrayInputStream byteArrayInputStream = - new ByteArrayInputStream(byteArrayOutputStream.toByteArray()); - DataInputStream inputStream = new DataInputStream(byteArrayInputStream); - - RegionStatistics statistics1 = new RegionStatistics(); - statistics1.deserialize(inputStream); - Assert.assertEquals(statistics0, statistics1); - } - } -}
