This is an automated email from the ASF dual-hosted git repository. yongzao pushed a commit to branch balancer_rename_and_annotations in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit fa3d7b62499a82f9e5bef9a32f2876f817b1b1fb Author: YongzaoDan <[email protected]> AuthorDate: Wed Oct 2 22:14:54 2024 +0800 annotations --- .../manager/load/balancer/RegionBalancer.java | 4 +- .../manager/load/balancer/RouteBalancer.java | 4 +- .../region/GreedyRegionGroupAllocator.java | 57 +++-- .../PartiteGraphPlacementRegionGroupAllocator.java | 236 ++++++++++++++++++ ...artiteGraphReplicationRegionGroupAllocator.java | 256 -------------------- ...r.java => CostFlowSelectionLeaderBalancer.java} | 263 +++++++++++---------- .../router/leader/CFDLeaderBalancerTest.java | 3 +- .../leader/LeaderBalancerComparisonTest.java | 4 +- 8 files changed, 413 insertions(+), 414 deletions(-) diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java index 6ba5b34c3dc..1b553eda448 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java @@ -33,7 +33,7 @@ import org.apache.iotdb.confignode.manager.load.LoadManager; import org.apache.iotdb.confignode.manager.load.balancer.region.GreedyCopySetRegionGroupAllocator; import org.apache.iotdb.confignode.manager.load.balancer.region.GreedyRegionGroupAllocator; import org.apache.iotdb.confignode.manager.load.balancer.region.IRegionGroupAllocator; -import org.apache.iotdb.confignode.manager.load.balancer.region.PartiteGraphReplicationRegionGroupAllocator; +import org.apache.iotdb.confignode.manager.load.balancer.region.PartiteGraphPlacementRegionGroupAllocator; import org.apache.iotdb.confignode.manager.node.NodeManager; import org.apache.iotdb.confignode.manager.partition.PartitionManager; import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager; @@ -59,7 +59,7 @@ public class RegionBalancer { this.regionGroupAllocator = new GreedyRegionGroupAllocator(); break; case PGR: - this.regionGroupAllocator = new PartiteGraphReplicationRegionGroupAllocator(); + this.regionGroupAllocator = new PartiteGraphPlacementRegionGroupAllocator(); break; case GCR: default: 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 a5d4e3d0166..77d534c237d 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 @@ -34,8 +34,8 @@ import org.apache.iotdb.confignode.manager.IManager; import org.apache.iotdb.confignode.manager.ProcedureManager; import org.apache.iotdb.confignode.manager.load.LoadManager; import org.apache.iotdb.confignode.manager.load.balancer.router.leader.AbstractLeaderBalancer; +import org.apache.iotdb.confignode.manager.load.balancer.router.leader.CostFlowSelectionLeaderBalancer; import org.apache.iotdb.confignode.manager.load.balancer.router.leader.GreedyLeaderBalancer; -import org.apache.iotdb.confignode.manager.load.balancer.router.leader.MinCostFlowLeaderBalancer; 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; @@ -127,7 +127,7 @@ public class RouteBalancer implements IClusterStatusSubscriber { break; case AbstractLeaderBalancer.CFD_POLICY: default: - this.leaderBalancer = new MinCostFlowLeaderBalancer(); + this.leaderBalancer = new CostFlowSelectionLeaderBalancer(); break; } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/GreedyRegionGroupAllocator.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/GreedyRegionGroupAllocator.java index d05a8accbec..6535b426928 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/GreedyRegionGroupAllocator.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/GreedyRegionGroupAllocator.java @@ -24,23 +24,48 @@ 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.tsfile.utils.Pair; - +import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.Objects; +import java.util.Random; import java.util.stream.Collectors; -import static java.util.Map.Entry.comparingByValue; - /** Allocate Region Greedily */ public class GreedyRegionGroupAllocator implements IRegionGroupAllocator { + public static final Random RANDOM = new Random(); + public GreedyRegionGroupAllocator() { // Empty constructor } + public static class DataNodeEntry implements Comparable<DataNodeEntry> { + + public int dataNodeId; + public int regionCount; + public double freeDiskSpace; + public int randomWeight; + + public DataNodeEntry(int dataNodeId, int regionCount, double freeDiskSpace) { + this.dataNodeId = dataNodeId; + this.regionCount = regionCount; + this.freeDiskSpace = freeDiskSpace; + this.randomWeight = RANDOM.nextInt(); + } + + @Override + public int compareTo(DataNodeEntry other) { + if (this.regionCount != other.regionCount) { + return this.regionCount - other.regionCount; + } else if (this.freeDiskSpace != other.freeDiskSpace) { + return (int) (other.freeDiskSpace - this.freeDiskSpace); + } else { + return this.randomWeight - other.randomWeight; + } + } + } + @Override public TRegionReplicaSet generateOptimalRegionReplicasDistribution( Map<Integer, TDataNodeConfiguration> availableDataNodeMap, @@ -73,27 +98,19 @@ public class GreedyRegionGroupAllocator implements IRegionGroupAllocator { regionCounter.merge(dataNodeLocation.getDataNodeId(), 1, Integer::sum))); /* Construct priority map */ - Map<TDataNodeLocation, Pair<Integer, Double>> priorityMap = - new HashMap<>(availableDataNodeMap.size()); + List<DataNodeEntry> entryList = new ArrayList<>(); availableDataNodeMap.forEach( (datanodeId, dataNodeConfiguration) -> - priorityMap.put( - dataNodeConfiguration.getLocation(), - new Pair<>( + entryList.add( + new DataNodeEntry( + datanodeId, regionCounter.getOrDefault(datanodeId, 0), freeDiskSpaceMap.getOrDefault(datanodeId, 0d)))); // Sort weightList - return priorityMap.entrySet().stream() - .sorted( - comparingByValue( - (o1, o2) -> - !Objects.equals(o1.getLeft(), o2.getLeft()) - // Compare the first key(The number of Regions) by ascending order - ? o1.getLeft() - o2.getLeft() - // Compare the second key(The free disk space) by descending order - : (int) (o2.getRight() - o1.getRight()))) - .map(entry -> entry.getKey().deepCopy()) + return entryList.stream() + .sorted() + .map(entry -> availableDataNodeMap.get(entry.dataNodeId).getLocation()) .collect(Collectors.toList()); } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/PartiteGraphPlacementRegionGroupAllocator.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/PartiteGraphPlacementRegionGroupAllocator.java new file mode 100644 index 00000000000..b67938397ef --- /dev/null +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/PartiteGraphPlacementRegionGroupAllocator.java @@ -0,0 +1,236 @@ +/* + * 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.manager.load.balancer.region; + +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.TRegionReplicaSet; +import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; +import org.apache.iotdb.confignode.manager.load.balancer.region.GreedyRegionGroupAllocator.DataNodeEntry; + +import org.apache.tsfile.utils.Pair; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.TreeMap; +import java.util.stream.Collectors; + +public class PartiteGraphPlacementRegionGroupAllocator implements IRegionGroupAllocator { + + private static final GreedyRegionGroupAllocator GREEDY_ALLOCATOR = + new GreedyRegionGroupAllocator(); + + private int subGraphCount; + private int replicationFactor; + private int regionPerDataNode; + + private int dataNodeNum; + // The number of allocated Regions in each DataNode + private int[] regionCounter; + // The number of edges in current cluster + private int[][] combinationCounter; + private Map<Integer, Integer> fakeToRealIdMap; + + private int alphaDataNodeNum; + // Pair<combinationSum, RegionSum> + Pair<Integer, Integer> bestValue; + private int[] bestAlphaNodes; + + @Override + public TRegionReplicaSet generateOptimalRegionReplicasDistribution( + Map<Integer, TDataNodeConfiguration> availableDataNodeMap, + Map<Integer, Double> freeDiskSpaceMap, + List<TRegionReplicaSet> allocatedRegionGroups, + List<TRegionReplicaSet> databaseAllocatedRegionGroups, + int replicationFactor, + TConsensusGroupId consensusGroupId) { + this.regionPerDataNode = + (int) + (consensusGroupId.getType().equals(TConsensusGroupType.DataRegion) + ? ConfigNodeDescriptor.getInstance().getConf().getDataRegionPerDataNode() + : ConfigNodeDescriptor.getInstance().getConf().getSchemaRegionPerDataNode()); + prepare(replicationFactor, availableDataNodeMap, allocatedRegionGroups); + + // Select alpha nodes set + for (int i = 0; i < subGraphCount; i++) { + subGraphSearch(i, freeDiskSpaceMap); + } + if (bestValue.left == Integer.MAX_VALUE) { + // Use greedy allocator as alternative if no alpha nodes set is found + return GREEDY_ALLOCATOR.generateOptimalRegionReplicasDistribution( + availableDataNodeMap, + freeDiskSpaceMap, + allocatedRegionGroups, + databaseAllocatedRegionGroups, + replicationFactor, + consensusGroupId); + } + + // Select the beta nodes sets + List<Integer> betaDataNodes = partiteGraphSearch(bestAlphaNodes[0] % subGraphCount); + if (betaDataNodes.size() < replicationFactor - alphaDataNodeNum) { + // Use greedy allocator as alternative if no beta nodes set is found + return GREEDY_ALLOCATOR.generateOptimalRegionReplicasDistribution( + availableDataNodeMap, + freeDiskSpaceMap, + allocatedRegionGroups, + databaseAllocatedRegionGroups, + replicationFactor, + consensusGroupId); + } + + // The next placement scheme is alpha \cup beta + TRegionReplicaSet result = new TRegionReplicaSet(); + result.setRegionId(consensusGroupId); + for (int i = 0; i < alphaDataNodeNum; i++) { + result.addToDataNodeLocations( + availableDataNodeMap.get(fakeToRealIdMap.get(bestAlphaNodes[i])).getLocation()); + } + for (int i = 0; i < replicationFactor - alphaDataNodeNum; i++) { + result.addToDataNodeLocations( + availableDataNodeMap.get(fakeToRealIdMap.get(betaDataNodes.get(i))).getLocation()); + } + return result; + } + + private void prepare( + int replicationFactor, + Map<Integer, TDataNodeConfiguration> availableDataNodeMap, + List<TRegionReplicaSet> allocatedRegionGroups) { + this.subGraphCount = replicationFactor / 2 + (replicationFactor % 2 == 0 ? 0 : 1); + this.replicationFactor = replicationFactor; + + // Initialize the fake index for each DataNode, + // since the index of DataNode might not be continuous + this.fakeToRealIdMap = new TreeMap<>(); + Map<Integer, Integer> realToFakeIdMap = new TreeMap<>(); + this.dataNodeNum = availableDataNodeMap.size(); + List<Integer> dataNodeIdList = + availableDataNodeMap.values().stream() + .map(c -> c.getLocation().getDataNodeId()) + .collect(Collectors.toList()); + for (int i = 0; i < dataNodeNum; i++) { + fakeToRealIdMap.put(i, dataNodeIdList.get(i)); + realToFakeIdMap.put(dataNodeIdList.get(i), i); + } + + // Compute regionCounter and combinationCounter + this.regionCounter = new int[dataNodeNum]; + Arrays.fill(regionCounter, 0); + this.combinationCounter = new int[dataNodeNum][dataNodeNum]; + for (int i = 0; i < dataNodeNum; i++) { + Arrays.fill(combinationCounter[i], 0); + } + for (TRegionReplicaSet regionReplicaSet : allocatedRegionGroups) { + List<TDataNodeLocation> dataNodeLocations = regionReplicaSet.getDataNodeLocations(); + for (int i = 0; i < dataNodeLocations.size(); i++) { + int fakeIId = realToFakeIdMap.get(dataNodeLocations.get(i).getDataNodeId()); + regionCounter[fakeIId]++; + for (int j = i + 1; j < dataNodeLocations.size(); j++) { + int fakeJId = realToFakeIdMap.get(dataNodeLocations.get(j).getDataNodeId()); + combinationCounter[fakeIId][fakeJId] = 1; + combinationCounter[fakeJId][fakeIId] = 1; + } + } + } + + // Reset the optimal result + this.alphaDataNodeNum = replicationFactor / 2 + 1; + this.bestValue = new Pair<>(Integer.MAX_VALUE, Integer.MAX_VALUE); + this.bestAlphaNodes = new int[alphaDataNodeNum]; + } + + private Pair<Integer, Integer> valuation(int[] nodes) { + int edgeSum = 0; + int regionSum = 0; + for (int iota : nodes) { + for (int kappa : nodes) { + edgeSum += combinationCounter[iota][kappa]; + } + regionSum += regionCounter[iota]; + } + return new Pair<>(edgeSum, regionSum); + } + + private void subGraphSearch(int firstIndex, Map<Integer, Double> freeDiskSpaceMap) { + List<DataNodeEntry> entryList = new ArrayList<>(); + for (int index = firstIndex; index < dataNodeNum; index += subGraphCount) { + // Prune: skip filled DataNodes + if (regionCounter[index] < regionPerDataNode) { + entryList.add(new DataNodeEntry(index, regionCounter[index], freeDiskSpaceMap.get(fakeToRealIdMap.get(index)))); + } + } + if (entryList.size() < alphaDataNodeNum) { + // Skip: not enough DataNodes + return; + } + Collections.sort(entryList); + int[] alphaNodes = new int[alphaDataNodeNum]; + for (int i = 0; i < alphaDataNodeNum - 1; i++) { + alphaNodes[i] = entryList.get(i).dataNodeId; + } + for (int i = alphaDataNodeNum - 1; i < entryList.size(); i++) { + alphaNodes[alphaDataNodeNum - 1] = entryList.get(i).dataNodeId; + Pair<Integer, Integer> currentValue = valuation(alphaNodes); + if (currentValue.left < bestValue.left + || (currentValue.left.equals(bestValue.left) && currentValue.right < bestValue.right)) { + bestValue = currentValue; + System.arraycopy(alphaNodes, 0, bestAlphaNodes, 0, alphaDataNodeNum); + } + } + } + + private List<Integer> partiteGraphSearch(int alphaIndex) { + List<Integer> betaNodes = new ArrayList<>(); + int[] tmpNodes = new int[alphaDataNodeNum + 1]; + System.arraycopy(bestAlphaNodes, 0, tmpNodes, 0, alphaDataNodeNum); + for (int partiteIndex = 0; partiteIndex < subGraphCount; partiteIndex++) { + if (partiteIndex == alphaIndex) { + // Skip the alphaIndex subgraph + continue; + } + int selectedDataNode = -1; + Pair<Integer, Integer> tmpValue = new Pair<>(Integer.MAX_VALUE, Integer.MAX_VALUE); + for (int i = partiteIndex; i < dataNodeNum; i += subGraphCount) { + if (regionCounter[i] >= regionPerDataNode) { + // Pruning: skip filled DataNodes + continue; + } + tmpNodes[alphaDataNodeNum] = i; + Pair<Integer, Integer> currentValue = valuation(tmpNodes); + if (currentValue.left < tmpValue.left + || (currentValue.left.equals(tmpValue.left) && currentValue.right < tmpValue.right)) { + tmpValue = currentValue; + selectedDataNode = i; + } + } + if (selectedDataNode == -1) { + return new ArrayList<>(); + } + betaNodes.add(selectedDataNode); + } + return betaNodes; + } +} 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 deleted file mode 100644 index b24acc1bd46..00000000000 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/PartiteGraphReplicationRegionGroupAllocator.java +++ /dev/null @@ -1,256 +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.manager.load.balancer.region; - -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.TRegionReplicaSet; -import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; - -import java.util.ArrayList; -import java.util.Arrays; -import java.util.List; -import java.util.Map; -import java.util.Random; -import java.util.TreeMap; -import java.util.stream.Collectors; - -public class PartiteGraphReplicationRegionGroupAllocator implements IRegionGroupAllocator { - - private static final Random RANDOM = new Random(); - private static final GreedyRegionGroupAllocator GREEDY_ALLOCATOR = - new GreedyRegionGroupAllocator(); - - private int subGraphCount; - private int replicationFactor; - private int regionPerDataNode; - - private int dataNodeNum; - // The number of allocated Regions in each DataNode - private int[] regionCounter; - // The number of edges in current cluster - private int[][] combinationCounter; - private Map<Integer, Integer> fakeToRealIdMap; - - private int alphaDataNodeNum; - // First Key: the sum of overlapped 2-Region combination Regions with - // other allocated RegionGroups is minimal - private int optimalEdgeSum; - // Second Key: the sum of DataRegions in selected DataNodes is minimal - private int optimalRegionSum; - private int[] optimalAlphaNodes; - - @Override - public TRegionReplicaSet generateOptimalRegionReplicasDistribution( - Map<Integer, TDataNodeConfiguration> availableDataNodeMap, - Map<Integer, Double> freeDiskSpaceMap, - List<TRegionReplicaSet> allocatedRegionGroups, - List<TRegionReplicaSet> databaseAllocatedRegionGroups, - int replicationFactor, - TConsensusGroupId consensusGroupId) { - - this.regionPerDataNode = - (int) - (consensusGroupId.getType().equals(TConsensusGroupType.DataRegion) - ? ConfigNodeDescriptor.getInstance().getConf().getDataRegionPerDataNode() - : ConfigNodeDescriptor.getInstance().getConf().getSchemaRegionPerDataNode()); - prepare(replicationFactor, availableDataNodeMap, allocatedRegionGroups); - - // Select a set of optimal alpha nodes - for (int i = 0; i < subGraphCount; i++) { - subGraphSearch(i, 0, alphaDataNodeNum, 0, 0, new int[alphaDataNodeNum]); - } - if (optimalEdgeSum == Integer.MAX_VALUE) { - return GREEDY_ALLOCATOR.generateOptimalRegionReplicasDistribution( - availableDataNodeMap, - freeDiskSpaceMap, - allocatedRegionGroups, - databaseAllocatedRegionGroups, - replicationFactor, - consensusGroupId); - } - - // Select the set of optimal beta nodes - List<Integer> partiteNodes = partiteGraphSearch(optimalAlphaNodes[0] % subGraphCount); - if (partiteNodes.size() < replicationFactor - alphaDataNodeNum) { - return GREEDY_ALLOCATOR.generateOptimalRegionReplicasDistribution( - availableDataNodeMap, - freeDiskSpaceMap, - allocatedRegionGroups, - databaseAllocatedRegionGroups, - replicationFactor, - consensusGroupId); - } - - TRegionReplicaSet result = new TRegionReplicaSet(); - result.setRegionId(consensusGroupId); - for (int i = 0; i < alphaDataNodeNum; i++) { - result.addToDataNodeLocations( - availableDataNodeMap.get(fakeToRealIdMap.get(optimalAlphaNodes[i])).getLocation()); - } - for (int i = 0; i < replicationFactor - alphaDataNodeNum; i++) { - result.addToDataNodeLocations( - availableDataNodeMap.get(fakeToRealIdMap.get(partiteNodes.get(i))).getLocation()); - } - return result; - } - - private void prepare( - int replicationFactor, - Map<Integer, TDataNodeConfiguration> availableDataNodeMap, - List<TRegionReplicaSet> allocatedRegionGroups) { - - this.subGraphCount = replicationFactor / 2 + (replicationFactor % 2 == 0 ? 0 : 1); - this.replicationFactor = replicationFactor; - - this.fakeToRealIdMap = new TreeMap<>(); - Map<Integer, Integer> realToFakeIdMap = new TreeMap<>(); - this.dataNodeNum = availableDataNodeMap.size(); - List<Integer> dataNodeIdList = - availableDataNodeMap.values().stream() - .map(c -> c.getLocation().getDataNodeId()) - .collect(Collectors.toList()); - for (int i = 0; i < dataNodeNum; i++) { - fakeToRealIdMap.put(i, dataNodeIdList.get(i)); - realToFakeIdMap.put(dataNodeIdList.get(i), i); - } - - // Compute regionCounter and combinationCounter - this.regionCounter = new int[dataNodeNum]; - Arrays.fill(regionCounter, 0); - this.combinationCounter = new int[dataNodeNum][dataNodeNum]; - for (int i = 0; i < dataNodeNum; i++) { - Arrays.fill(combinationCounter[i], 0); - } - for (TRegionReplicaSet regionReplicaSet : allocatedRegionGroups) { - List<TDataNodeLocation> dataNodeLocations = regionReplicaSet.getDataNodeLocations(); - for (int i = 0; i < dataNodeLocations.size(); i++) { - int fakeIId = realToFakeIdMap.get(dataNodeLocations.get(i).getDataNodeId()); - regionCounter[fakeIId]++; - for (int j = i + 1; j < dataNodeLocations.size(); j++) { - int fakeJId = realToFakeIdMap.get(dataNodeLocations.get(j).getDataNodeId()); - combinationCounter[fakeIId][fakeJId] = 1; - combinationCounter[fakeJId][fakeIId] = 1; - } - } - } - - // Reset the optimal result - this.alphaDataNodeNum = replicationFactor / 2 + 1; - this.optimalEdgeSum = Integer.MAX_VALUE; - this.optimalRegionSum = Integer.MAX_VALUE; - this.optimalAlphaNodes = new int[alphaDataNodeNum]; - } - - private void subGraphSearch( - int firstIndex, - int currentReplica, - int replicaNum, - int combinationSum, - int regionSum, - int[] currentReplicaSet) { - - if (currentReplica == replicaNum) { - if (combinationSum < optimalEdgeSum - || (combinationSum == optimalEdgeSum && regionSum < optimalRegionSum)) { - // Reset the optimal result when a better one is found - optimalEdgeSum = combinationSum; - optimalRegionSum = regionSum; - optimalAlphaNodes = Arrays.copyOf(currentReplicaSet, replicationFactor); - } else if (combinationSum == optimalEdgeSum - && regionSum == optimalRegionSum - && RANDOM.nextBoolean()) { - optimalAlphaNodes = Arrays.copyOf(currentReplicaSet, replicationFactor); - } - return; - } - - for (int i = firstIndex; i < dataNodeNum; i += subGraphCount) { - if (regionCounter[i] >= regionPerDataNode) { - // Pruning: skip full DataNodes - continue; - } - int nxtCombinationSum = combinationSum; - for (int j = 0; j < currentReplica; j++) { - nxtCombinationSum += combinationCounter[i][currentReplicaSet[j]]; - } - if (combinationSum > optimalEdgeSum) { - // Pruning: no needs for further searching when the first key - // is bigger than the historical optimal result - return; - } - int nxtRegionSum = regionSum + regionCounter[i]; - if (combinationSum == optimalEdgeSum && regionSum > optimalRegionSum) { - // Pruning: no needs for further searching when the second key - // is bigger than the historical optimal result - return; - } - currentReplicaSet[currentReplica] = i; - subGraphSearch( - i + subGraphCount, - currentReplica + 1, - replicaNum, - nxtCombinationSum, - nxtRegionSum, - currentReplicaSet); - } - } - - private List<Integer> partiteGraphSearch(int selected) { - List<Integer> partiteNodes = new ArrayList<>(); - for (int partiteIndex = 0; partiteIndex < subGraphCount; partiteIndex++) { - if (partiteIndex == selected) { - continue; - } - int selectedDataNode = -1; - int bestScatterWidth = 0; - int bestRegionSum = Integer.MAX_VALUE; - for (int i = partiteIndex; i < dataNodeNum; i += subGraphCount) { - if (regionCounter[i] >= regionPerDataNode) { - continue; - } - int scatterWidth = alphaDataNodeNum; - for (int k = 0; k < alphaDataNodeNum; k++) { - scatterWidth -= combinationCounter[i][optimalAlphaNodes[k]]; - } - if (scatterWidth < bestScatterWidth) { - continue; - } - if (scatterWidth > bestScatterWidth) { - bestScatterWidth = scatterWidth; - bestRegionSum = regionCounter[i]; - selectedDataNode = i; - } else if (regionCounter[i] < bestRegionSum) { - bestRegionSum = regionCounter[i]; - selectedDataNode = i; - } else if (regionCounter[i] == bestRegionSum && RANDOM.nextBoolean()) { - selectedDataNode = i; - } - } - if (selectedDataNode == -1) { - return new ArrayList<>(); - } - partiteNodes.add(selectedDataNode); - } - return partiteNodes; - } -} diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/MinCostFlowLeaderBalancer.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/CostFlowSelectionLeaderBalancer.java similarity index 52% rename from iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/MinCostFlowLeaderBalancer.java rename to iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/CostFlowSelectionLeaderBalancer.java index 669f2dc9f40..588219d70c0 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/MinCostFlowLeaderBalancer.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/CostFlowSelectionLeaderBalancer.java @@ -36,48 +36,48 @@ import java.util.TreeMap; import java.util.concurrent.ConcurrentHashMap; /** Leader distribution balancer that uses minimum cost flow algorithm */ -public class MinCostFlowLeaderBalancer extends AbstractLeaderBalancer { +public class CostFlowSelectionLeaderBalancer extends AbstractLeaderBalancer { private static final int INFINITY = Integer.MAX_VALUE; - /** Graph nodes */ - // Super source node - private static final int S_NODE = 0; - - // Super terminal node - private static final int T_NODE = 1; - // Maximum index of graph nodes - private int maxNode = T_NODE + 1; - // Map<RegionGroupId, rNode> - private final Map<TConsensusGroupId, Integer> rNodeMap; - // Map<Database, Map<DataNodeId, sDNode>> - private final Map<String, Map<Integer, Integer>> sDNodeMap; - // Map<Database, Map<sDNode, DataNodeId>> - private final Map<String, Map<Integer, Integer>> sDNodeReflect; - // Map<DataNodeId, tDNode> - private final Map<Integer, Integer> tDNodeMap; + /** Graph vertices */ + // Super source vertex + private static final int S_VERTEX = 0; + + // Super terminal vertex + private static final int T_VERTEX = 1; + // Maximum index of graph vertices + private int maxVertex = T_VERTEX + 1; + // Map<RegionGroupId, rVertex> + private final Map<TConsensusGroupId, Integer> rVertexMap; + // Map<Database, Map<DataNodeId, sDVertex>> + private final Map<String, Map<Integer, Integer>> sDVertexMap; + // Map<Database, Map<sDVertex, DataNodeId>> + private final Map<String, Map<Integer, Integer>> sDVertexReflect; + // Map<DataNodeId, tDVertex> + private final Map<Integer, Integer> tDVertexMap; /** Graph edges */ // Maximum index of graph edges private int maxEdge = 0; - private final List<MinCostFlowEdge> minCostFlowEdges; - private int[] nodeHeadEdge; - private int[] nodeCurrentEdge; + private final List<CostFlowEdge> costFlowEdges; + private int[] vertexHeadEdge; + private int[] vertexCurrentEdge; - private boolean[] isNodeVisited; - private int[] nodeMinimumCost; + private boolean[] isVertexVisited; + private int[] vertexMinimumCost; private int maximumFlow = 0; private int minimumCost = 0; - public MinCostFlowLeaderBalancer() { + public CostFlowSelectionLeaderBalancer() { super(); - this.rNodeMap = new TreeMap<>(); - this.sDNodeMap = new TreeMap<>(); - this.sDNodeReflect = new TreeMap<>(); - this.tDNodeMap = new TreeMap<>(); - this.minCostFlowEdges = new ArrayList<>(); + this.rVertexMap = new TreeMap<>(); + this.sDVertexMap = new TreeMap<>(); + this.sDVertexReflect = new TreeMap<>(); + this.tDVertexMap = new TreeMap<>(); + this.costFlowEdges = new ArrayList<>(); } @Override @@ -94,7 +94,7 @@ public class MinCostFlowLeaderBalancer extends AbstractLeaderBalancer { dataNodeStatisticsMap, regionStatisticsMap); Map<TConsensusGroupId, Integer> result; - constructMCFGraph(); + constructFlowNetwork(); dinicAlgorithm(); result = collectLeaderDistribution(); clear(); @@ -104,45 +104,47 @@ public class MinCostFlowLeaderBalancer extends AbstractLeaderBalancer { @Override protected void clear() { super.clear(); - this.rNodeMap.clear(); - this.sDNodeMap.clear(); - this.sDNodeReflect.clear(); - this.tDNodeMap.clear(); - this.minCostFlowEdges.clear(); - this.nodeHeadEdge = null; - this.nodeCurrentEdge = null; - this.isNodeVisited = null; - this.nodeMinimumCost = null; - this.maxNode = T_NODE + 1; + this.rVertexMap.clear(); + this.sDVertexMap.clear(); + this.sDVertexReflect.clear(); + this.tDVertexMap.clear(); + this.costFlowEdges.clear(); + this.vertexHeadEdge = null; + this.vertexCurrentEdge = null; + this.isVertexVisited = null; + this.vertexMinimumCost = null; + this.maxVertex = T_VERTEX + 1; this.maxEdge = 0; } - private void constructMCFGraph() { + private void constructFlowNetwork() { this.maximumFlow = 0; this.minimumCost = 0; - /* Indicate nodes in mcf */ + /* Indicate vertices */ for (Map.Entry<String, List<TConsensusGroupId>> databaseEntry : databaseRegionGroupMap.entrySet()) { String database = databaseEntry.getKey(); - sDNodeMap.put(database, new TreeMap<>()); - sDNodeReflect.put(database, new TreeMap<>()); + sDVertexMap.put(database, new TreeMap<>()); + sDVertexReflect.put(database, new TreeMap<>()); for (TConsensusGroupId regionGroupId : databaseEntry.getValue()) { if (regionGroupIntersection.contains(regionGroupId)) { - rNodeMap.put(regionGroupId, maxNode++); + // Map region to region vertices + rVertexMap.put(regionGroupId, maxVertex++); regionLocationMap .get(regionGroupId) .forEach( dataNodeId -> { if (isDataNodeAvailable(dataNodeId)) { - if (!sDNodeMap.get(database).containsKey(dataNodeId)) { - sDNodeMap.get(database).put(dataNodeId, maxNode); - sDNodeReflect.get(database).put(maxNode, dataNodeId); - maxNode += 1; + // Map DataNode to DataNode vertices + if (!sDVertexMap.get(database).containsKey(dataNodeId)) { + sDVertexMap.get(database).put(dataNodeId, maxVertex); + sDVertexReflect.get(database).put(maxVertex, dataNodeId); + maxVertex += 1; } - if (!tDNodeMap.containsKey(dataNodeId)) { - tDNodeMap.put(dataNodeId, maxNode); - maxNode += 1; + if (!tDVertexMap.containsKey(dataNodeId)) { + tDVertexMap.put(dataNodeId, maxVertex); + maxVertex += 1; } } }); @@ -151,48 +153,49 @@ public class MinCostFlowLeaderBalancer extends AbstractLeaderBalancer { } /* Prepare arrays */ - isNodeVisited = new boolean[maxNode]; - nodeMinimumCost = new int[maxNode]; - nodeCurrentEdge = new int[maxNode]; - nodeHeadEdge = new int[maxNode]; - Arrays.fill(nodeHeadEdge, -1); - - /* Construct edges: sNode -> rNodes */ - for (int rNode : rNodeMap.values()) { - // Capacity: 1, Cost: 0, each RegionGroup should elect exactly 1 leader - addAdjacentEdges(S_NODE, rNode, 1, 0); + isVertexVisited = new boolean[maxVertex]; + vertexMinimumCost = new int[maxVertex]; + vertexCurrentEdge = new int[maxVertex]; + vertexHeadEdge = new int[maxVertex]; + Arrays.fill(vertexHeadEdge, -1); + + /* Construct edges: sVertex -> rVertices */ + for (int rVertex : rVertexMap.values()) { + // Capacity: 1, Cost: 0, select exactly 1 leader for each RegionGroup + addAdjacentEdges(S_VERTEX, rVertex, 1, 0); } - /* Construct edges: rNodes -> sdNodes */ + /* Construct edges: rVertices -> sDVertices */ for (Map.Entry<String, List<TConsensusGroupId>> databaseEntry : databaseRegionGroupMap.entrySet()) { String database = databaseEntry.getKey(); for (TConsensusGroupId regionGroupId : databaseEntry.getValue()) { if (regionGroupIntersection.contains(regionGroupId)) { - int rNode = rNodeMap.get(regionGroupId); + int rVertex = rVertexMap.get(regionGroupId); regionLocationMap .get(regionGroupId) .forEach( dataNodeId -> { if (isDataNodeAvailable(dataNodeId) && isRegionAvailable(regionGroupId, dataNodeId)) { - int sDNode = sDNodeMap.get(database).get(dataNodeId); - // Capacity: 1, Cost: 1 if sDNode is the current leader of the rNode, 0 - // otherwise. - // Therefore, the RegionGroup will keep the leader as constant as possible. + int sDVertex = sDVertexMap.get(database).get(dataNodeId); + // Capacity: 1, Cost: 1 if the DataNode is the current leader of the + // RegionGroup; + // 0 otherwise. Thus, the RegionGroup will keep the leader as constant as + // possible. int cost = Objects.equals( regionLeaderMap.getOrDefault(regionGroupId, -1), dataNodeId) ? 0 : 1; - addAdjacentEdges(rNode, sDNode, 1, cost); + addAdjacentEdges(rVertex, sDVertex, 1, cost); } }); } } } - /* Construct edges: sDNodes -> tDNodes */ + /* Construct edges: sDVertices -> tDVertices */ for (Map.Entry<String, List<TConsensusGroupId>> databaseEntry : databaseRegionGroupMap.entrySet()) { String database = databaseEntry.getKey(); @@ -205,21 +208,20 @@ public class MinCostFlowLeaderBalancer extends AbstractLeaderBalancer { .forEach( dataNodeId -> { if (isDataNodeAvailable(dataNodeId)) { - int sDNode = sDNodeMap.get(database).get(dataNodeId); - int tDNode = tDNodeMap.get(dataNodeId); + int sDVertex = sDVertexMap.get(database).get(dataNodeId); + int tDVertex = tDVertexMap.get(dataNodeId); int leaderCount = leaderCounter.merge(dataNodeId, 1, Integer::sum); - // Capacity: 1, Cost: x^2 for the x-th edge at the current sDNode. + // Capacity: 1, Cost: 2*x-1 for the x-th edge at the current sDVertex. // Thus, the leader distribution will be as balance as possible within each - // Database - // based on the Jensen's-Inequality. - addAdjacentEdges(sDNode, tDNode, 1, leaderCount * leaderCount); + // Database according to the Jensen's-Inequality. + addAdjacentEdges(sDVertex, tDVertex, 1, 2 * leaderCount - 1); } }); } } } - /* Construct edges: tDNodes -> tNode */ + /* Construct edges: tDVertices -> tVertex */ // Map<DataNodeId, possible maximum leader> // Count the possible maximum number of leader in each DataNode Map<Integer, Integer> maxLeaderCounter = new TreeMap<>(); @@ -227,89 +229,88 @@ public class MinCostFlowLeaderBalancer extends AbstractLeaderBalancer { (regionGroupId, dataNodeIds) -> dataNodeIds.forEach( dataNodeId -> { - if (isDataNodeAvailable(dataNodeId) && tDNodeMap.containsKey(dataNodeId)) { - int tDNode = tDNodeMap.get(dataNodeId); + if (isDataNodeAvailable(dataNodeId) && tDVertexMap.containsKey(dataNodeId)) { + int tDVertex = tDVertexMap.get(dataNodeId); int leaderCount = maxLeaderCounter.merge(dataNodeId, 1, Integer::sum); - // Cost: x^2 for the x-th edge at the current dNode. + // Capacity: 1, Cost: 2*x-1 for the x-th edge at the current tDVertex. // Thus, the leader distribution will be as balance as possible within the - // cluster - // Based on the Jensen's-Inequality. - addAdjacentEdges(tDNode, T_NODE, 1, leaderCount * leaderCount); + // cluster according to the Jensen's-Inequality. + addAdjacentEdges(tDVertex, T_VERTEX, 1, 2 * leaderCount - 1); } })); } - private void addAdjacentEdges(int fromNode, int destNode, int capacity, int cost) { - addEdge(fromNode, destNode, capacity, cost); - addEdge(destNode, fromNode, 0, -cost); + private void addAdjacentEdges(int fromVertex, int destVertex, int capacity, int cost) { + addEdge(fromVertex, destVertex, capacity, cost); + addEdge(destVertex, fromVertex, 0, -cost); } - private void addEdge(int fromNode, int destNode, int capacity, int cost) { - MinCostFlowEdge edge = new MinCostFlowEdge(destNode, capacity, cost, nodeHeadEdge[fromNode]); - minCostFlowEdges.add(edge); - nodeHeadEdge[fromNode] = maxEdge++; + private void addEdge(int fromVertex, int destVertex, int capacity, int cost) { + CostFlowEdge edge = new CostFlowEdge(destVertex, capacity, cost, vertexHeadEdge[fromVertex]); + costFlowEdges.add(edge); + vertexHeadEdge[fromVertex] = maxEdge++; } /** - * Check whether there is an augmented path in the MCF graph by Bellman-Ford algorithm. + * Check whether there is an augmented path in the flow network by SPFA algorithm. * * <p>Notice: Never use Dijkstra algorithm to replace this since there might exist negative * circles. * * @return True if there exist augmented paths, false otherwise. */ - private boolean bellmanFordCheck() { - Arrays.fill(isNodeVisited, false); - Arrays.fill(nodeMinimumCost, INFINITY); + private boolean SPFACheck() { + Arrays.fill(isVertexVisited, false); + Arrays.fill(vertexMinimumCost, INFINITY); Queue<Integer> queue = new LinkedList<>(); - nodeMinimumCost[S_NODE] = 0; - isNodeVisited[S_NODE] = true; - queue.offer(S_NODE); + vertexMinimumCost[S_VERTEX] = 0; + isVertexVisited[S_VERTEX] = true; + queue.offer(S_VERTEX); while (!queue.isEmpty()) { - int currentNode = queue.poll(); - isNodeVisited[currentNode] = false; - for (int currentEdge = nodeHeadEdge[currentNode]; + int currentVertex = queue.poll(); + isVertexVisited[currentVertex] = false; + for (int currentEdge = vertexHeadEdge[currentVertex]; currentEdge >= 0; - currentEdge = minCostFlowEdges.get(currentEdge).nextEdge) { - MinCostFlowEdge edge = minCostFlowEdges.get(currentEdge); + currentEdge = costFlowEdges.get(currentEdge).nextEdge) { + CostFlowEdge edge = costFlowEdges.get(currentEdge); if (edge.capacity > 0 - && nodeMinimumCost[currentNode] + edge.cost < nodeMinimumCost[edge.destNode]) { - nodeMinimumCost[edge.destNode] = nodeMinimumCost[currentNode] + edge.cost; - if (!isNodeVisited[edge.destNode]) { - isNodeVisited[edge.destNode] = true; - queue.offer(edge.destNode); + && vertexMinimumCost[currentVertex] + edge.cost < vertexMinimumCost[edge.destVertex]) { + vertexMinimumCost[edge.destVertex] = vertexMinimumCost[currentVertex] + edge.cost; + if (!isVertexVisited[edge.destVertex]) { + isVertexVisited[edge.destVertex] = true; + queue.offer(edge.destVertex); } } } } - return nodeMinimumCost[T_NODE] < INFINITY; + return vertexMinimumCost[T_VERTEX] < INFINITY; } /** Do augmentation by dfs algorithm */ - private int dfsAugmentation(int currentNode, int inputFlow) { - if (currentNode == T_NODE || inputFlow == 0) { + private int dfsAugmentation(int currentVertex, int inputFlow) { + if (currentVertex == T_VERTEX || inputFlow == 0) { return inputFlow; } int currentEdge; int outputFlow = 0; - isNodeVisited[currentNode] = true; - for (currentEdge = nodeCurrentEdge[currentNode]; + isVertexVisited[currentVertex] = true; + for (currentEdge = vertexCurrentEdge[currentVertex]; currentEdge >= 0; - currentEdge = minCostFlowEdges.get(currentEdge).nextEdge) { - MinCostFlowEdge edge = minCostFlowEdges.get(currentEdge); - if (nodeMinimumCost[currentNode] + edge.cost == nodeMinimumCost[edge.destNode] + currentEdge = costFlowEdges.get(currentEdge).nextEdge) { + CostFlowEdge edge = costFlowEdges.get(currentEdge); + if (vertexMinimumCost[currentVertex] + edge.cost == vertexMinimumCost[edge.destVertex] && edge.capacity > 0 - && !isNodeVisited[edge.destNode]) { + && !isVertexVisited[edge.destVertex]) { - int subOutputFlow = dfsAugmentation(edge.destNode, Math.min(inputFlow, edge.capacity)); + int subOutputFlow = dfsAugmentation(edge.destVertex, Math.min(inputFlow, edge.capacity)); minimumCost += subOutputFlow * edge.cost; edge.capacity -= subOutputFlow; - minCostFlowEdges.get(currentEdge ^ 1).capacity += subOutputFlow; + costFlowEdges.get(currentEdge ^ 1).capacity += subOutputFlow; inputFlow -= subOutputFlow; outputFlow += subOutputFlow; @@ -319,19 +320,19 @@ public class MinCostFlowLeaderBalancer extends AbstractLeaderBalancer { } } } - nodeCurrentEdge[currentNode] = currentEdge; + vertexCurrentEdge[currentVertex] = currentEdge; if (outputFlow > 0) { - isNodeVisited[currentNode] = false; + isVertexVisited[currentVertex] = false; } return outputFlow; } private void dinicAlgorithm() { - while (bellmanFordCheck()) { + while (SPFACheck()) { int currentFlow; - System.arraycopy(nodeHeadEdge, 0, nodeCurrentEdge, 0, maxNode); - while ((currentFlow = dfsAugmentation(S_NODE, INFINITY)) > 0) { + System.arraycopy(vertexHeadEdge, 0, vertexCurrentEdge, 0, maxVertex); + while ((currentFlow = dfsAugmentation(S_VERTEX, INFINITY)) > 0) { maximumFlow += currentFlow; } } @@ -353,13 +354,13 @@ public class MinCostFlowLeaderBalancer extends AbstractLeaderBalancer { return; } boolean matchLeader = false; - for (int currentEdge = nodeHeadEdge[rNodeMap.get(regionGroupId)]; + for (int currentEdge = vertexHeadEdge[rVertexMap.get(regionGroupId)]; currentEdge >= 0; - currentEdge = minCostFlowEdges.get(currentEdge).nextEdge) { - MinCostFlowEdge edge = minCostFlowEdges.get(currentEdge); - if (edge.destNode != S_NODE && edge.capacity == 0) { + currentEdge = costFlowEdges.get(currentEdge).nextEdge) { + CostFlowEdge edge = costFlowEdges.get(currentEdge); + if (edge.destVertex != S_VERTEX && edge.capacity == 0) { matchLeader = true; - result.put(regionGroupId, sDNodeReflect.get(database).get(edge.destNode)); + result.put(regionGroupId, sDVertexReflect.get(database).get(edge.destVertex)); } } if (!matchLeader) { @@ -380,15 +381,15 @@ public class MinCostFlowLeaderBalancer extends AbstractLeaderBalancer { return minimumCost; } - private static class MinCostFlowEdge { + private static class CostFlowEdge { - private final int destNode; + private final int destVertex; private int capacity; private final int cost; private final int nextEdge; - private MinCostFlowEdge(int destNode, int capacity, int cost, int nextEdge) { - this.destNode = destNode; + private CostFlowEdge(int destVertex, int capacity, int cost, int nextEdge) { + this.destVertex = destVertex; this.capacity = capacity; this.cost = cost; this.nextEdge = nextEdge; diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/CFDLeaderBalancerTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/CFDLeaderBalancerTest.java index f2f51e044ff..fea4c33cec4 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/CFDLeaderBalancerTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/CFDLeaderBalancerTest.java @@ -47,7 +47,8 @@ import java.util.stream.Collectors; public class CFDLeaderBalancerTest { - private static final MinCostFlowLeaderBalancer BALANCER = new MinCostFlowLeaderBalancer(); + private static final CostFlowSelectionLeaderBalancer BALANCER = + new CostFlowSelectionLeaderBalancer(); private static final String DATABASE = "root.database"; diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/LeaderBalancerComparisonTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/LeaderBalancerComparisonTest.java index 885c97989da..49efa2dcc64 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/LeaderBalancerComparisonTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/LeaderBalancerComparisonTest.java @@ -57,8 +57,8 @@ public class LeaderBalancerComparisonTest { private static FileWriter WRITER; private static final GreedyLeaderBalancer GREEDY_LEADER_BALANCER = new GreedyLeaderBalancer(); - private static final MinCostFlowLeaderBalancer MIN_COST_FLOW_LEADER_BALANCER = - new MinCostFlowLeaderBalancer(); + private static final CostFlowSelectionLeaderBalancer MIN_COST_FLOW_LEADER_BALANCER = + new CostFlowSelectionLeaderBalancer(); private static final Random RANDOM = new Random(); private static final int TEST_MAX_DATA_NODE_NUM = 100;
