You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ta...@apache.org on 2024/12/24 13:04:55 UTC
(iotdb) branch master updated: [remove datanode] Remove LoadScore sort to fix RegionPriority order (#14527)
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 17729a987b2 [remove datanode] Remove LoadScore sort to fix RegionPriority order (#14527)
17729a987b2 is described below
commit 17729a987b25edf4257335db42c9080aff2cdb33
Author: Xiangpeng Hu <65...@users.noreply.github.com>
AuthorDate: Tue Dec 24 21:04:49 2024 +0800
[remove datanode] Remove LoadScore sort to fix RegionPriority order (#14527)
---
.../iotdb/confignode/manager/load/LoadManager.java | 9 ---
.../manager/load/balancer/RouteBalancer.java | 7 +--
.../router/priority/GreedyPriorityBalancer.java | 42 +-------------
.../router/priority/IPriorityBalancer.java | 5 +-
.../router/priority/LeaderPriorityBalancer.java | 21 +++----
.../confignode/manager/load/cache/LoadCache.java | 16 ------
.../router/priority/GreedyPriorityTest.java | 13 +----
.../priority/LeaderPriorityBalancerTest.java | 65 +---------------------
.../apache/iotdb/consensus/iot/IoTConsensus.java | 8 ++-
.../service/IoTConsensusRPCServiceProcessor.java | 6 +-
10 files changed, 28 insertions(+), 164 deletions(-)
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 bb4c6fa068f..835f92a302e 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
@@ -219,15 +219,6 @@ public class LoadManager {
return loadCache.getFreeDiskSpace(dataNodeId);
}
- /**
- * Get the loadScore of each DataNode.
- *
- * @return Map<DataNodeId, loadScore>
- */
- public Map<Integer, Long> getAllDataNodeLoadScores() {
- return loadCache.getAllDataNodeLoadScores();
- }
-
/**
* Get the lowest loadScore DataNode.
*
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 1490b125121..f88302115cd 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
@@ -296,20 +296,17 @@ public class RouteBalancer implements IClusterStatusSubscriber {
new TreeMap<>();
try {
Map<TConsensusGroupId, Integer> regionLeaderMap = getLoadManager().getRegionLeaderMap();
- Map<Integer, Long> dataNodeLoadScoreMap = getLoadManager().getAllDataNodeLoadScores();
// Balancing region priority in each SchemaRegionGroup
Map<TConsensusGroupId, TRegionReplicaSet> optimalRegionPriorityMap =
priorityRouter.generateOptimalRoutePriority(
getPartitionManager().getAllReplicaSets(TConsensusGroupType.SchemaRegion),
- regionLeaderMap,
- dataNodeLoadScoreMap);
+ regionLeaderMap);
// Balancing region priority in each DataRegionGroup
optimalRegionPriorityMap.putAll(
priorityRouter.generateOptimalRoutePriority(
getPartitionManager().getAllReplicaSets(TConsensusGroupType.DataRegion),
- regionLeaderMap,
- dataNodeLoadScoreMap));
+ regionLeaderMap));
optimalRegionPriorityMap.forEach(
(regionGroupId, optimalRegionPriority) -> {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/GreedyPriorityBalancer.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/GreedyPriorityBalancer.java
index 6eacf6f906b..3d2ef54deb2 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/GreedyPriorityBalancer.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/GreedyPriorityBalancer.java
@@ -19,16 +19,11 @@
package org.apache.iotdb.confignode.manager.load.balancer.router.priority;
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.tsfile.utils.Pair;
-
-import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
-import java.util.Vector;
/** The GreedyPriorityBalancer always pick the Replica with the lowest loadScore */
public class GreedyPriorityBalancer implements IPriorityBalancer {
@@ -39,48 +34,15 @@ public class GreedyPriorityBalancer implements IPriorityBalancer {
@Override
public Map<TConsensusGroupId, TRegionReplicaSet> generateOptimalRoutePriority(
- List<TRegionReplicaSet> replicaSets,
- Map<TConsensusGroupId, Integer> regionLeaderMap,
- Map<Integer, Long> dataNodeLoadScoreMap) {
+ List<TRegionReplicaSet> replicaSets, Map<TConsensusGroupId, Integer> regionLeaderMap) {
Map<TConsensusGroupId, TRegionReplicaSet> regionPriorityMap = new TreeMap<>();
replicaSets.forEach(
replicaSet -> {
- TRegionReplicaSet sortedReplicaSet =
- sortReplicasByLoadScore(replicaSet, dataNodeLoadScoreMap);
- regionPriorityMap.put(sortedReplicaSet.getRegionId(), sortedReplicaSet);
+ regionPriorityMap.put(replicaSet.getRegionId(), replicaSet);
});
return regionPriorityMap;
}
-
- protected static TRegionReplicaSet sortReplicasByLoadScore(
- TRegionReplicaSet replicaSet, Map<Integer, Long> dataNodeLoadScoreMap) {
- TRegionReplicaSet sortedReplicaSet = new TRegionReplicaSet();
- sortedReplicaSet.setRegionId(replicaSet.getRegionId());
-
- // List<Pair<loadScore, TDataNodeLocation>> for sorting
- List<Pair<Long, TDataNodeLocation>> sortList = new Vector<>();
- replicaSet
- .getDataNodeLocations()
- .forEach(
- dataNodeLocation -> {
- // The absenteeism of loadScoreMap means ConfigNode-leader doesn't receive any
- // heartbeat from that DataNode.
- // In this case we put a maximum loadScore into the sortList.
- sortList.add(
- new Pair<>(
- dataNodeLoadScoreMap.computeIfAbsent(
- dataNodeLocation.getDataNodeId(), empty -> Long.MAX_VALUE),
- dataNodeLocation));
- });
-
- sortList.sort(Comparator.comparingLong(Pair::getLeft));
- for (Pair<Long, TDataNodeLocation> entry : sortList) {
- sortedReplicaSet.addToDataNodeLocations(entry.getRight());
- }
-
- return sortedReplicaSet;
- }
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/IPriorityBalancer.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/IPriorityBalancer.java
index 47d86509645..0fa270c8908 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/IPriorityBalancer.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/IPriorityBalancer.java
@@ -34,12 +34,9 @@ public interface IPriorityBalancer {
*
* @param replicaSets All RegionGroups
* @param regionLeaderMap The current leader of each RegionGroup
- * @param dataNodeLoadScoreMap The current load score of each DataNode
* @return Map<TConsensusGroupId, TRegionReplicaSet>, The optimal route priority for each
* RegionGroup. The replica with higher sorting result have higher priority.
*/
Map<TConsensusGroupId, TRegionReplicaSet> generateOptimalRoutePriority(
- List<TRegionReplicaSet> replicaSets,
- Map<TConsensusGroupId, Integer> regionLeaderMap,
- Map<Integer, Long> dataNodeLoadScoreMap);
+ List<TRegionReplicaSet> replicaSets, Map<TConsensusGroupId, Integer> regionLeaderMap);
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/LeaderPriorityBalancer.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/LeaderPriorityBalancer.java
index 04113f5d54b..5fc9b5ddb45 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/LeaderPriorityBalancer.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/LeaderPriorityBalancer.java
@@ -35,31 +35,24 @@ public class LeaderPriorityBalancer extends GreedyPriorityBalancer implements IP
@Override
public Map<TConsensusGroupId, TRegionReplicaSet> generateOptimalRoutePriority(
- List<TRegionReplicaSet> replicaSets,
- Map<TConsensusGroupId, Integer> regionLeaderMap,
- Map<Integer, Long> dataNodeLoadScoreMap) {
-
+ List<TRegionReplicaSet> replicaSets, Map<TConsensusGroupId, Integer> regionLeaderMap) {
Map<TConsensusGroupId, TRegionReplicaSet> regionPriorityMap = new TreeMap<>();
replicaSets.forEach(
replicaSet -> {
- /* 1. Sort replicaSet by loadScore */
- TRegionReplicaSet sortedReplicaSet =
- sortReplicasByLoadScore(replicaSet, dataNodeLoadScoreMap);
- /* 2. Pick leader if leader exists and available */
+ /* 1. Pick leader if leader exists and available */
int leaderId = regionLeaderMap.getOrDefault(replicaSet.getRegionId(), -1);
- if (leaderId != -1
- && dataNodeLoadScoreMap.getOrDefault(leaderId, Long.MAX_VALUE) < Long.MAX_VALUE) {
- for (int i = 0; i < sortedReplicaSet.getDataNodeLocationsSize(); i++) {
- if (sortedReplicaSet.getDataNodeLocations().get(i).getDataNodeId() == leaderId) {
- Collections.swap(sortedReplicaSet.getDataNodeLocations(), 0, i);
+ if (leaderId != -1) {
+ for (int i = 0; i < replicaSet.getDataNodeLocationsSize(); i++) {
+ if (replicaSet.getDataNodeLocations().get(i).getDataNodeId() == leaderId) {
+ Collections.swap(replicaSet.getDataNodeLocations(), 0, i);
break;
}
}
}
- regionPriorityMap.put(sortedReplicaSet.getRegionId(), sortedReplicaSet);
+ regionPriorityMap.put(replicaSet.getRegionId(), replicaSet);
});
return regionPriorityMap;
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 1886d7e76d4..e1080b94740 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
@@ -553,22 +553,6 @@ public class LoadCache {
.orElse(0d);
}
- /**
- * Get the loadScore of each DataNode.
- *
- * @return Map<DataNodeId, loadScore>
- */
- public Map<Integer, Long> getAllDataNodeLoadScores() {
- Map<Integer, Long> result = new ConcurrentHashMap<>();
- nodeCacheMap.forEach(
- (dataNodeId, heartbeatCache) -> {
- if (heartbeatCache instanceof DataNodeHeartbeatCache) {
- result.put(dataNodeId, heartbeatCache.getLoadScore());
- }
- });
- return result;
- }
-
/**
* Get the lowest loadScore DataNode.
*
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 a18bca54f99..48104bf78a3 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
@@ -36,7 +36,6 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
public class GreedyPriorityTest {
@@ -68,17 +67,11 @@ public class GreedyPriorityTest {
}
nodeCacheMap.values().forEach(baseNodeCache -> baseNodeCache.updateCurrentStatistics(false));
- /* Get the loadScoreMap */
- Map<Integer, Long> loadScoreMap = new ConcurrentHashMap<>();
- nodeCacheMap.forEach(
- (dataNodeId, heartbeatCache) ->
- loadScoreMap.put(dataNodeId, heartbeatCache.getLoadScore()));
-
/* Build TRegionReplicaSet */
TConsensusGroupId groupId1 = new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 1);
TRegionReplicaSet regionReplicaSet1 =
new TRegionReplicaSet(
- groupId1, Arrays.asList(dataNodeLocations.get(1), dataNodeLocations.get(0)));
+ groupId1, Arrays.asList(dataNodeLocations.get(0), dataNodeLocations.get(1)));
TConsensusGroupId groupId2 = new TConsensusGroupId(TConsensusGroupType.DataRegion, 2);
TRegionReplicaSet regionReplicaSet2 =
new TRegionReplicaSet(
@@ -88,7 +81,7 @@ public class GreedyPriorityTest {
Map<TConsensusGroupId, TRegionReplicaSet> result =
new GreedyPriorityBalancer()
.generateOptimalRoutePriority(
- Arrays.asList(regionReplicaSet1, regionReplicaSet2), new HashMap<>(), loadScoreMap);
+ Arrays.asList(regionReplicaSet1, regionReplicaSet2), new HashMap<>());
Assert.assertEquals(2, result.size());
TRegionReplicaSet result1 = result.get(groupId1);
@@ -97,7 +90,7 @@ public class GreedyPriorityTest {
}
TRegionReplicaSet result2 = result.get(groupId2);
- for (int i = 3; i < 4; i++) {
+ for (int i = 2; i < 4; i++) {
Assert.assertEquals(dataNodeLocations.get(i), result2.getDataNodeLocations().get(i - 2));
}
}
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 7b126ed6e84..5af3a677aab 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
@@ -33,11 +33,9 @@ import org.junit.Test;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
public class LeaderPriorityBalancerTest {
@@ -69,25 +67,19 @@ public class LeaderPriorityBalancerTest {
}
nodeCacheMap.values().forEach(baseNodeCache -> baseNodeCache.updateCurrentStatistics(false));
- // Get the loadScoreMap
- Map<Integer, Long> loadScoreMap = new ConcurrentHashMap<>();
- nodeCacheMap.forEach(
- (dataNodeId, heartbeatCache) ->
- loadScoreMap.put(dataNodeId, heartbeatCache.getLoadScore()));
-
// Build TRegionReplicaSet
TConsensusGroupId groupId1 = new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 1);
TRegionReplicaSet regionReplicaSet1 =
new TRegionReplicaSet(
groupId1,
Arrays.asList(
- dataNodeLocations.get(2), dataNodeLocations.get(1), dataNodeLocations.get(0)));
+ dataNodeLocations.get(0), dataNodeLocations.get(1), dataNodeLocations.get(2)));
TConsensusGroupId groupId2 = new TConsensusGroupId(TConsensusGroupType.DataRegion, 2);
TRegionReplicaSet regionReplicaSet2 =
new TRegionReplicaSet(
groupId2,
Arrays.asList(
- dataNodeLocations.get(5), dataNodeLocations.get(4), dataNodeLocations.get(3)));
+ dataNodeLocations.get(3), dataNodeLocations.get(4), dataNodeLocations.get(5)));
List<TRegionReplicaSet> regionReplicaSets = Arrays.asList(regionReplicaSet1, regionReplicaSet2);
// Build leaderMap
@@ -97,67 +89,16 @@ public class LeaderPriorityBalancerTest {
// Check result
Map<TConsensusGroupId, TRegionReplicaSet> result =
- new LeaderPriorityBalancer()
- .generateOptimalRoutePriority(regionReplicaSets, leaderMap, loadScoreMap);
+ new LeaderPriorityBalancer().generateOptimalRoutePriority(regionReplicaSets, leaderMap);
TRegionReplicaSet result1 = result.get(groupId1);
// Leader first
Assert.assertEquals(dataNodeLocations.get(1), result1.getDataNodeLocations().get(0));
- // The others will be sorted by loadScore
Assert.assertEquals(dataNodeLocations.get(0), result1.getDataNodeLocations().get(1));
Assert.assertEquals(dataNodeLocations.get(2), result1.getDataNodeLocations().get(2));
TRegionReplicaSet result2 = result.get(groupId2);
// Leader first
Assert.assertEquals(dataNodeLocations.get(4), result2.getDataNodeLocations().get(0));
- // The others will be sorted by loadScore
Assert.assertEquals(dataNodeLocations.get(3), result2.getDataNodeLocations().get(1));
Assert.assertEquals(dataNodeLocations.get(5), result2.getDataNodeLocations().get(2));
}
-
- @Test
- public void testLeaderUnavailable() {
- // Build TDataNodeLocations
- List<TDataNodeLocation> dataNodeLocations = new ArrayList<>();
- for (int i = 0; i < 3; i++) {
- dataNodeLocations.add(
- new TDataNodeLocation(
- i,
- new TEndPoint("0.0.0.0", 6667 + i),
- new TEndPoint("0.0.0.0", 10730 + i),
- new TEndPoint("0.0.0.0", 10740 + i),
- new TEndPoint("0.0.0.0", 10760 + i),
- new TEndPoint("0.0.0.0", 10750 + i)));
- }
-
- // Build TRegionReplicaSet
- TConsensusGroupId groupId1 = new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 1);
- TRegionReplicaSet regionReplicaSet1 =
- new TRegionReplicaSet(
- groupId1,
- Arrays.asList(
- dataNodeLocations.get(2), dataNodeLocations.get(1), dataNodeLocations.get(0)));
-
- // Build leaderMap
- Map<TConsensusGroupId, Integer> leaderMap = new HashMap<>();
- leaderMap.put(groupId1, 1);
-
- // Build loadScoreMap
- Map<Integer, Long> loadScoreMap = new ConcurrentHashMap<>();
- loadScoreMap.put(0, 10L);
- loadScoreMap.put(2, 20L);
- // The leader is DataNode-1, but it's unavailable
- loadScoreMap.put(1, Long.MAX_VALUE);
-
- // Check result
- Map<TConsensusGroupId, TRegionReplicaSet> result =
- new LeaderPriorityBalancer()
- .generateOptimalRoutePriority(
- Collections.singletonList(regionReplicaSet1), leaderMap, loadScoreMap);
- // Only sorted by loadScore since the leader is unavailable
- Assert.assertEquals(
- dataNodeLocations.get(0), result.get(groupId1).getDataNodeLocations().get(0));
- Assert.assertEquals(
- dataNodeLocations.get(2), result.get(groupId1).getDataNodeLocations().get(1));
- Assert.assertEquals(
- dataNodeLocations.get(1), result.get(groupId1).getDataNodeLocations().get(2));
- }
}
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
index 987ba658538..75d1ca39918 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
@@ -234,9 +234,11 @@ public class IoTConsensus implements IConsensus {
if (impl.isReadOnly()) {
return StatusUtils.getStatus(TSStatusCode.SYSTEM_READ_ONLY);
} else if (!impl.isActive()) {
- return RpcUtils.getStatus(
- TSStatusCode.WRITE_PROCESS_REJECT,
- "peer is inactive and not ready to receive sync log request.");
+ String message =
+ String.format(
+ "Peer is inactive and not ready to write request, %s, DataNode Id: %s",
+ groupId.toString(), impl.getThisNode().getNodeId());
+ return RpcUtils.getStatus(TSStatusCode.WRITE_PROCESS_REJECT, message);
} else {
return impl.write(request);
}
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java
index dc88d8eb98b..f6a67c3e8c6 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java
@@ -97,8 +97,12 @@ public class IoTConsensusRPCServiceProcessor implements IoTConsensusIService.Ifa
return new TSyncLogEntriesRes(Collections.singletonList(status));
}
if (!impl.isActive()) {
+ String message =
+ String.format(
+ "Peer is inactive and not ready to receive sync log request, %s, DataNode Id: %s",
+ groupId, impl.getThisNode().getNodeId());
TSStatus status = new TSStatus(TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode());
- status.setMessage("peer is inactive and not ready to receive sync log request");
+ status.setMessage(message);
return new TSyncLogEntriesRes(Collections.singletonList(status));
}
BatchIndexedConsensusRequest logEntriesInThisBatch =