This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
The following commit(s) were added to refs/heads/master by this push:
new 882276b0a Do not drop a second replica while another one is already
being dropped (#3175)
882276b0a is described below
commit 882276b0a3b709d1aa27c432746e2871a850b18a
Author: Junkai Xue <[email protected]>
AuthorDate: Mon Nov 9 20:31:01 2026 -0800
Do not drop a second replica while another one is already being dropped
(#3175)
When the preference list of a partition changes while a replica that is
no longer in it still has a pending message bringing it down,
DelayedAutoRebalancer picked another, still active replica to drop. The
partition then ran with fewer active replicas than required until the
new replica came up, which is what
TestZeroReplicaAvoidance.testWagedRebalancer
intermittently observed in the external view.
When the extra replicas are dropped, the ones that are already in the
initial state are now chosen first, then the ones with a pending
transition to the initial state or DROPPED, and only then the remaining
ones in their existing order.
---
.../rebalancer/DelayedAutoRebalancer.java | 43 ++++++++++-
.../TestDelayedAutoRebalancer.MasterSlave.json | 24 +++++-
.../TestDelayedAutoRebalancer.OnlineOffline.json | 87 ++++++++++++++++++++++
3 files changed, 150 insertions(+), 4 deletions(-)
diff --git
a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/DelayedAutoRebalancer.java
b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/DelayedAutoRebalancer.java
index 939e894cb..93ac0fd62 100644
---
a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/DelayedAutoRebalancer.java
+++
b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/DelayedAutoRebalancer.java
@@ -21,6 +21,7 @@ package org.apache.helix.controller.rebalancer;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -402,8 +403,26 @@ public class DelayedAutoRebalancer extends
AbstractRebalancer<ResourceController
if
(!currentMapWithPreferenceList.containsValue(HelixDefinedState.ERROR.name())
&& bestPossibleStateMap.size() > numReplicas &&
readyToDrop(currentStateMap,
bestPossibleStateMap, preferenceList, combinedPreferenceList)) {
- for (int i = 0; i < combinedPreferenceList.size() - numReplicas; i++) {
- String instanceToDrop =
combinedPreferenceList.get(combinedPreferenceList.size() - i - 1);
+ int numToDrop = combinedPreferenceList.size() - numReplicas;
+ // The instances that are not in the preference list are sorted to the
end of the combined
+ // list. Collect them from the end, i.e. in the order in which they
would be dropped.
+ Set<String> preferenceSet = new HashSet<>(preferenceList);
+ List<String> dropCandidates = new ArrayList<>();
+ for (int i = combinedPreferenceList.size() - 1;
+ i >= 0 && !preferenceSet.contains(combinedPreferenceList.get(i));
i--) {
+ dropCandidates.add(combinedPreferenceList.get(i));
+ }
+ // Replicas that are already inactive, or are being brought down by a
pending message, are
+ // dropped first. Otherwise, if the preference list changes while such a
message is in
+ // flight (or the combined list is not ordered by state, as in a
topology aware cluster),
+ // another replica that is still active would be picked on top of the
one that is already
+ // leaving, and the partition would end up with fewer active replicas
than required.
+ String initialState = stateModelDef.getInitialState();
+ dropCandidates.sort(Comparator.comparingInt(
+ instance -> getDropRank(currentStateMap.get(instance),
pendingStates.get(instance),
+ initialState)));
+ for (String instanceToDrop : dropCandidates.subList(0,
+ Math.min(numToDrop, dropCandidates.size()))) {
bestPossibleStateMap.put(instanceToDrop,
HelixDefinedState.DROPPED.name());
}
}
@@ -423,6 +442,26 @@ public class DelayedAutoRebalancer extends
AbstractRebalancer<ResourceController
return bestPossibleStateMap;
}
+ /**
+ * Rank a replica that is not in the preference list for dropping; the lower
the rank, the
+ * earlier it is dropped.
+ * 0: the replica is already in the initial state.
+ * 1: the replica is being brought down by a pending message. An ERROR
replica being reset is
+ * not leaving, it is kept as ERROR in the best possible state anyway.
+ * 2: any other replica.
+ */
+ private static int getDropRank(String currentState, String pendingState,
String initialState) {
+ if (initialState.equals(currentState)) {
+ return 0;
+ }
+ if (!HelixDefinedState.ERROR.name().equals(currentState) && (
+ initialState.equals(pendingState) || HelixDefinedState.DROPPED.name()
+ .equals(pendingState))) {
+ return 1;
+ }
+ return 2;
+ }
+
private boolean readyToDrop(Map<String, String> currentStateMap,
Map<String, String> bestPossibleMap, List<String> preferenceList,
List<String> combinedPreferenceList) {
diff --git
a/helix-core/src/test/resources/TestDelayedAutoRebalancer.MasterSlave.json
b/helix-core/src/test/resources/TestDelayedAutoRebalancer.MasterSlave.json
index 22bfbe311..7a889486b 100644
--- a/helix-core/src/test/resources/TestDelayedAutoRebalancer.MasterSlave.json
+++ b/helix-core/src/test/resources/TestDelayedAutoRebalancer.MasterSlave.json
@@ -252,8 +252,28 @@
"localhost_3": "MASTER",
"localhost_2": "DROPPED"
}
+ },
+ {
+ "preferenceList": [
+ "localhost_3",
+ "localhost_4",
+ "localhost_5"
+ ],
+ "currentStates": {
+ "localhost_0": "SLAVE",
+ "localhost_1": "SLAVE",
+ "localhost_2": "SLAVE",
+ "localhost_3": "MASTER"
+ },
+ "bestPossibleStates": {
+ "localhost_3": "MASTER",
+ "localhost_1": "SLAVE",
+ "localhost_2": "SLAVE",
+ "localhost_0": "DROPPED"
+ },
+ "pendingMessages": {
+ "localhost_0": "SLAVE:OFFLINE"
+ }
}
]
}
-
-
diff --git
a/helix-core/src/test/resources/TestDelayedAutoRebalancer.OnlineOffline.json
b/helix-core/src/test/resources/TestDelayedAutoRebalancer.OnlineOffline.json
index 55f5b3c3a..b50b8e7c6 100644
--- a/helix-core/src/test/resources/TestDelayedAutoRebalancer.OnlineOffline.json
+++ b/helix-core/src/test/resources/TestDelayedAutoRebalancer.OnlineOffline.json
@@ -93,6 +93,93 @@
"localhost_1": "ONLINE",
"localhost_2": "ERROR"
}
+ },
+ {
+ "preferenceList": [
+ "localhost_2",
+ "localhost_3",
+ "localhost_4"
+ ],
+ "currentStates": {
+ "localhost_0": "ONLINE",
+ "localhost_1": "ONLINE",
+ "localhost_2": "ONLINE",
+ "localhost_3": "ONLINE"
+ },
+ "bestPossibleStates": {
+ "localhost_2": "ONLINE",
+ "localhost_3": "ONLINE",
+ "localhost_1": "ONLINE",
+ "localhost_0": "DROPPED"
+ },
+ "pendingMessages": {
+ "localhost_0": "ONLINE:OFFLINE"
+ }
+ },
+ {
+ "preferenceList": [
+ "localhost_2",
+ "localhost_3",
+ "localhost_4"
+ ],
+ "currentStates": {
+ "localhost_0": "ONLINE",
+ "localhost_1": "OFFLINE",
+ "localhost_2": "ONLINE",
+ "localhost_3": "ONLINE"
+ },
+ "bestPossibleStates": {
+ "localhost_2": "ONLINE",
+ "localhost_3": "ONLINE",
+ "localhost_0": "ONLINE",
+ "localhost_1": "DROPPED"
+ },
+ "pendingMessages": {
+ "localhost_0": "ONLINE:OFFLINE"
+ }
+ },
+ {
+ "preferenceList": [
+ "localhost_2",
+ "localhost_3"
+ ],
+ "currentStates": {
+ "localhost_0": "ONLINE",
+ "localhost_1": "ONLINE",
+ "localhost_2": "ONLINE",
+ "localhost_3": "ONLINE"
+ },
+ "bestPossibleStates": {
+ "localhost_2": "ONLINE",
+ "localhost_3": "ONLINE",
+ "localhost_0": "DROPPED",
+ "localhost_1": "DROPPED"
+ },
+ "pendingMessages": {
+ "localhost_0": "ONLINE:OFFLINE"
+ }
+ },
+ {
+ "preferenceList": [
+ "localhost_2",
+ "localhost_3",
+ "localhost_4"
+ ],
+ "currentStates": {
+ "localhost_0": "ERROR",
+ "localhost_1": "ONLINE",
+ "localhost_2": "ONLINE",
+ "localhost_3": "ONLINE"
+ },
+ "bestPossibleStates": {
+ "localhost_2": "ONLINE",
+ "localhost_3": "ONLINE",
+ "localhost_0": "ERROR",
+ "localhost_1": "ONLINE"
+ },
+ "pendingMessages": {
+ "localhost_0": "ERROR:OFFLINE"
+ }
}
]
}
