You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@helix.apache.org by ji...@apache.org on 2019/10/28 22:33:25 UTC
[helix] 47/50: Refactor soft constraints to simply the algorithm
and fix potential issues. (#520)
This is an automated email from the ASF dual-hosted git repository.
jiajunwang pushed a commit to branch wagedRebalancer
in repository https://gitbox.apache.org/repos/asf/helix.git
commit ab1522cbdd1a2538c6efbdc88e0ce639ab1a8c55
Author: Jiajun Wang <18...@users.noreply.github.com>
AuthorDate: Fri Oct 25 21:56:46 2019 -0700
Refactor soft constraints to simply the algorithm and fix potential issues. (#520)
* Refactor soft constraints to simply the algorithm and fix potential issues.
1. Check for zero weight so as to avoid unnecessary calculations.
2. Simply the soft constraint interfaces and implementations. Avoid duplicate code.
3. Adjust partition movements constraint logic to reduce the chance of moving partition when the baseline and best possible assignment diverge.
4. Estimate utilization in addition to the other usage estimation. The estimation will be used as a base when calculating the capacity usage score. This is to ensure the algorithm treats different clusters with different overall usage in the same way.
5. Fix the issue that high utilization calculation does not consider the current proposed replica usage.
6. Use Sigmoid to calculate usage-based soft constraints score. This enhances the assignment result of the algorithm.
7. Adjust the related test cases.
---
helix-core/pom.xml | 7 +-
.../constraints/ConstraintBasedAlgorithm.java | 11 ++-
.../ConstraintBasedAlgorithmFactory.java | 36 ++++-----
.../InstancePartitionsCountConstraint.java | 17 ++---
.../MaxCapacityUsageInstanceConstraint.java | 15 ++--
.../constraints/PartitionMovementConstraint.java | 48 ++++++------
.../ResourcePartitionAntiAffinityConstraint.java | 21 ++----
.../ResourceTopStateAntiAffinityConstraint.java | 23 ++----
.../waged/constraints/SoftConstraint.java | 26 +++----
.../waged/constraints/UsageSoftConstraint.java | 88 ++++++++++++++++++++++
.../rebalancer/waged/model/AssignableNode.java | 25 +++---
.../rebalancer/waged/model/ClusterContext.java | 50 ++++++++++--
.../waged/model/ClusterModelProvider.java | 2 +-
.../resources/soft-constraint-weight.properties | 12 +--
.../constraints/TestConstraintBasedAlgorithm.java | 4 +-
.../TestInstancePartitionsCountConstraint.java | 12 +--
.../TestMaxCapacityUsageInstanceConstraint.java | 22 +++---
.../TestPartitionMovementConstraint.java | 34 ++++-----
...estResourcePartitionAntiAffinityConstraint.java | 16 ++--
...TestResourceTopStateAntiAffinityConstraint.java | 24 +++---
.../TestSoftConstraintNormalizeFunction.java | 4 +-
.../waged/model/AbstractTestClusterModel.java | 8 ++
.../waged/model/ClusterModelTestHelper.java | 11 +--
.../rebalancer/waged/model/TestAssignableNode.java | 14 ++--
.../rebalancer/waged/model/TestClusterContext.java | 14 ++--
.../rebalancer/waged/model/TestClusterModel.java | 19 +----
.../rebalancer/WagedRebalancer/TestNodeSwap.java | 7 ++
27 files changed, 330 insertions(+), 240 deletions(-)
diff --git a/helix-core/pom.xml b/helix-core/pom.xml
index 45b6552..1077cc0 100644
--- a/helix-core/pom.xml
+++ b/helix-core/pom.xml
@@ -37,7 +37,7 @@ under the License.
org.I0Itec.zkclient*,
org.apache.commons.cli*;version="[1.2,2)",
org.apache.commons.io*;version="[1.4,2)",
- org.apache.commons.math*;version="[2.1,3)",
+ org.apache.commons.math*;version="[2.1,4)",
org.apache.jute*;resolution:=optional,
org.apache.zookeeper.server.persistence*;resolution:=optional,
org.apache.zookeeper.server.util*;resolution:=optional,
@@ -140,6 +140,11 @@ under the License.
<version>2.1</version>
</dependency>
<dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-math3</artifactId>
+ <version>3.6.1</version>
+ </dependency>
+ <dependency>
<groupId>commons-codec</groupId>
<artifactId>commons-codec</artifactId>
<version>1.6</version>
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/ConstraintBasedAlgorithm.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/ConstraintBasedAlgorithm.java
index 1a41aef..3f2f845 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/ConstraintBasedAlgorithm.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/ConstraintBasedAlgorithm.java
@@ -109,19 +109,22 @@ class ConstraintBasedAlgorithm implements RebalanceAlgorithm {
return Optional.empty();
}
- Function<AssignableNode, Float> calculatePoints =
+ Function<AssignableNode, Double> calculatePoints =
(candidateNode) -> getAssignmentNormalizedScore(candidateNode, replica, clusterContext);
return candidateNodes.stream().max(Comparator.comparing(calculatePoints));
}
- private float getAssignmentNormalizedScore(AssignableNode node, AssignableReplica replica,
+ private double getAssignmentNormalizedScore(AssignableNode node, AssignableReplica replica,
ClusterContext clusterContext) {
- float sum = 0;
+ double sum = 0;
for (Map.Entry<SoftConstraint, Float> softConstraintEntry : _softConstraints.entrySet()) {
SoftConstraint softConstraint = softConstraintEntry.getKey();
float weight = softConstraintEntry.getValue();
- sum += weight * softConstraint.getAssignmentNormalizedScore(node, replica, clusterContext);
+ if (weight != 0) {
+ // Skip calculating zero weighted constraints.
+ sum += weight * softConstraint.getAssignmentNormalizedScore(node, replica, clusterContext);
+ }
}
return sum;
}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/ConstraintBasedAlgorithmFactory.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/ConstraintBasedAlgorithmFactory.java
index fbf8b19..934bfa7 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/ConstraintBasedAlgorithmFactory.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/ConstraintBasedAlgorithmFactory.java
@@ -24,31 +24,25 @@ import java.util.List;
import java.util.Map;
import java.util.Properties;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Maps;
import org.apache.helix.HelixManagerProperties;
import org.apache.helix.SystemPropertyKeys;
import org.apache.helix.controller.rebalancer.waged.RebalanceAlgorithm;
import org.apache.helix.model.ClusterConfig;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Maps;
-
/**
* The factory class to create an instance of {@link ConstraintBasedAlgorithm}
*/
public class ConstraintBasedAlgorithmFactory {
- // Evenness constraints tend to score within a smaller range.
- // In order to let their scores cause enough difference in the final evaluation result, we need to
- // enlarge the overall weight of the evenness constraints compared with the movement constraint.
- // TODO: Tune or make the following factor configurable.
- private static final int EVENNESS_PREFERENCE_NORMALIZE_FACTOR = 50;
private static final Map<String, Float> MODEL = new HashMap<String, Float>() {
{
// The default setting
- put(PartitionMovementConstraint.class.getSimpleName(), 1f);
- put(InstancePartitionsCountConstraint.class.getSimpleName(), 0.3f);
- put(ResourcePartitionAntiAffinityConstraint.class.getSimpleName(), 0.1f);
- put(ResourceTopStateAntiAffinityConstraint.class.getSimpleName(), 0.1f);
- put(MaxCapacityUsageInstanceConstraint.class.getSimpleName(), 0.5f);
+ put(PartitionMovementConstraint.class.getSimpleName(), 2f);
+ put(InstancePartitionsCountConstraint.class.getSimpleName(), 1f);
+ put(ResourcePartitionAntiAffinityConstraint.class.getSimpleName(), 1f);
+ put(ResourceTopStateAntiAffinityConstraint.class.getSimpleName(), 3f);
+ put(MaxCapacityUsageInstanceConstraint.class.getSimpleName(), 5f);
}
};
@@ -68,21 +62,19 @@ public class ConstraintBasedAlgorithmFactory {
new ValidGroupTagConstraint(), new SamePartitionOnInstanceConstraint());
int evennessPreference =
- preferences.getOrDefault(ClusterConfig.GlobalRebalancePreferenceKey.EVENNESS, 1)
- * EVENNESS_PREFERENCE_NORMALIZE_FACTOR;
+ preferences.getOrDefault(ClusterConfig.GlobalRebalancePreferenceKey.EVENNESS, 1);
int movementPreference =
preferences.getOrDefault(ClusterConfig.GlobalRebalancePreferenceKey.LESS_MOVEMENT, 1);
- float evennessRatio = (float) evennessPreference / (evennessPreference + movementPreference);
- float movementRatio = (float) movementPreference / (evennessPreference + movementPreference);
- List<SoftConstraint> softConstraints = ImmutableList.of(new PartitionMovementConstraint(),
- new InstancePartitionsCountConstraint(), new ResourcePartitionAntiAffinityConstraint(),
- new ResourceTopStateAntiAffinityConstraint(), new MaxCapacityUsageInstanceConstraint());
+ List<SoftConstraint> softConstraints = ImmutableList
+ .of(new PartitionMovementConstraint(), new InstancePartitionsCountConstraint(),
+ new ResourcePartitionAntiAffinityConstraint(),
+ new ResourceTopStateAntiAffinityConstraint(), new MaxCapacityUsageInstanceConstraint());
Map<SoftConstraint, Float> softConstraintsWithWeight = Maps.toMap(softConstraints, key -> {
String name = key.getClass().getSimpleName();
float weight = MODEL.get(name);
- return name.equals(PartitionMovementConstraint.class.getSimpleName()) ? movementRatio * weight
- : evennessRatio * weight;
+ return name.equals(PartitionMovementConstraint.class.getSimpleName()) ?
+ movementPreference * weight : evennessPreference * weight;
});
return new ConstraintBasedAlgorithm(hardConstraints, softConstraintsWithWeight);
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/InstancePartitionsCountConstraint.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/InstancePartitionsCountConstraint.java
index feee05a..948a7d0 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/InstancePartitionsCountConstraint.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/InstancePartitionsCountConstraint.java
@@ -29,20 +29,13 @@ import org.apache.helix.controller.rebalancer.waged.model.ClusterContext;
* Discourage the assignment if the instance's occupancy rate is above average
* The normalized score will be within [0, 1]
*/
-class InstancePartitionsCountConstraint extends SoftConstraint {
- private static final float MAX_SCORE = 1f;
- private static final float MIN_SCORE = 0f;
-
- InstancePartitionsCountConstraint() {
- super(MAX_SCORE, MIN_SCORE);
- }
+class InstancePartitionsCountConstraint extends UsageSoftConstraint {
@Override
- protected float getAssignmentScore(AssignableNode node, AssignableReplica replica,
+ protected double getAssignmentScore(AssignableNode node, AssignableReplica replica,
ClusterContext clusterContext) {
- float doubleEstimatedMaxPartitionCount = 2 * clusterContext.getEstimatedMaxPartitionCount();
- float currentPartitionCount = node.getAssignedReplicaCount();
- return Math.max((doubleEstimatedMaxPartitionCount - currentPartitionCount)
- / doubleEstimatedMaxPartitionCount, 0);
+ int estimatedMaxPartitionCount = clusterContext.getEstimatedMaxPartitionCount();
+ int currentPartitionCount = node.getAssignedReplicaCount();
+ return computeUtilizationScore(estimatedMaxPartitionCount, currentPartitionCount);
}
}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/MaxCapacityUsageInstanceConstraint.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/MaxCapacityUsageInstanceConstraint.java
index 2fe94c6..8f41f5e 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/MaxCapacityUsageInstanceConstraint.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/MaxCapacityUsageInstanceConstraint.java
@@ -30,18 +30,13 @@ import org.apache.helix.controller.rebalancer.waged.model.ClusterContext;
* that it is that much less desirable to assign anything on the given node.
* It is a greedy approach since it evaluates only on the most used capacity key.
*/
-class MaxCapacityUsageInstanceConstraint extends SoftConstraint {
- private static final float MIN_SCORE = 0;
- private static final float MAX_SCORE = 1;
-
- MaxCapacityUsageInstanceConstraint() {
- super(MAX_SCORE, MIN_SCORE);
- }
+class MaxCapacityUsageInstanceConstraint extends UsageSoftConstraint {
@Override
- protected float getAssignmentScore(AssignableNode node, AssignableReplica replica,
+ protected double getAssignmentScore(AssignableNode node, AssignableReplica replica,
ClusterContext clusterContext) {
- float maxCapacityUsage = node.getHighestCapacityUtilization();
- return 1.0f - maxCapacityUsage / 2.0f;
+ float estimatedMaxUtilization = clusterContext.getEstimatedMaxUtilization();
+ float projectedHighestUtilization = node.getProjectedHighestUtilization(replica.getCapacity());
+ return computeUtilizationScore(estimatedMaxUtilization, projectedHighestUtilization);
}
}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/PartitionMovementConstraint.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/PartitionMovementConstraint.java
index a781afc..dc19c19 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/PartitionMovementConstraint.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/PartitionMovementConstraint.java
@@ -38,39 +38,33 @@ import org.apache.helix.model.ResourceAssignment;
* evaluated score will become lower.
*/
class PartitionMovementConstraint extends SoftConstraint {
- private static final float MAX_SCORE = 1f;
- private static final float MIN_SCORE = 0f;
+ private static final double MAX_SCORE = 1f;
+ private static final double MIN_SCORE = 0f;
//TODO: these factors will be tuned based on user's preference
// This factor indicates the default score that is evaluated if only partition allocation matches
// (states are different).
- private static final float ALLOCATION_MATCH_FACTOR = 0.5f;
- // This factor indicates the contribution of the Baseline assignment matching to the final score.
- private static final float BASELINE_MATCH_FACTOR = 0.25f;
+ private static final double ALLOCATION_MATCH_FACTOR = 0.5;
PartitionMovementConstraint() {
super(MAX_SCORE, MIN_SCORE);
}
@Override
- protected float getAssignmentScore(AssignableNode node, AssignableReplica replica,
+ protected double getAssignmentScore(AssignableNode node, AssignableReplica replica,
ClusterContext clusterContext) {
- Map<String, String> bestPossibleStateMap =
+ // Prioritize the previous Best Possible assignment
+ Map<String, String> bestPossibleAssignment =
getStateMap(replica, clusterContext.getBestPossibleAssignment());
- Map<String, String> baselineStateMap =
+ if (!bestPossibleAssignment.isEmpty()) {
+ return calculateAssignmentScale(node, replica, bestPossibleAssignment);
+ }
+ // else, compare the baseline only if the best possible assignment does not contain the replica
+ Map<String, String> baselineAssignment =
getStateMap(replica, clusterContext.getBaselineAssignment());
-
- // Prioritize the matching of the previous Best Possible assignment.
- float scale = calculateAssignmentScale(node, replica, bestPossibleStateMap);
- // If the baseline is also provided, adjust the final score accordingly.
- scale = scale * (1 - BASELINE_MATCH_FACTOR)
- + calculateAssignmentScale(node, replica, baselineStateMap) * BASELINE_MATCH_FACTOR;
-
- return scale;
- }
-
- @Override
- NormalizeFunction getNormalizeFunction() {
- return score -> score * (getMaxScore() - getMinScore()) + getMinScore();
+ if (!baselineAssignment.isEmpty()) {
+ return calculateAssignmentScale(node, replica, baselineAssignment);
+ }
+ return 0;
}
private Map<String, String> getStateMap(AssignableReplica replica,
@@ -83,14 +77,20 @@ class PartitionMovementConstraint extends SoftConstraint {
return assignment.get(resourceName).getReplicaMap(new Partition(partitionName));
}
- private float calculateAssignmentScale(AssignableNode node, AssignableReplica replica,
+ private double calculateAssignmentScale(AssignableNode node, AssignableReplica replica,
Map<String, String> instanceToStateMap) {
String instanceName = node.getInstanceName();
if (!instanceToStateMap.containsKey(instanceName)) {
return 0;
} else {
- return (instanceToStateMap.get(instanceName).equals(replica.getReplicaState()) ? 1
- : ALLOCATION_MATCH_FACTOR);
+ return (instanceToStateMap.get(instanceName).equals(replica.getReplicaState()) ? 1 :
+ ALLOCATION_MATCH_FACTOR);
}
}
+
+ @Override
+ protected NormalizeFunction getNormalizeFunction() {
+ // PartitionMovementConstraint already scale the score properly.
+ return (score) -> score;
+ }
}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/ResourcePartitionAntiAffinityConstraint.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/ResourcePartitionAntiAffinityConstraint.java
index a2f9099..a3b701f 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/ResourcePartitionAntiAffinityConstraint.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/ResourcePartitionAntiAffinityConstraint.java
@@ -29,24 +29,15 @@ import org.apache.helix.controller.rebalancer.waged.model.ClusterContext;
* the same resource be assigned to the same node to minimize the impact of node failure scenarios.
* The score is higher the fewer the partitions are on the node belonging to the same resource.
*/
-class ResourcePartitionAntiAffinityConstraint extends SoftConstraint {
- private static final float MAX_SCORE = 1f;
- private static final float MIN_SCORE = 0f;
-
- ResourcePartitionAntiAffinityConstraint() {
- super(MAX_SCORE, MIN_SCORE);
- }
-
+class ResourcePartitionAntiAffinityConstraint extends UsageSoftConstraint {
@Override
- protected float getAssignmentScore(AssignableNode node, AssignableReplica replica,
+ protected double getAssignmentScore(AssignableNode node, AssignableReplica replica,
ClusterContext clusterContext) {
String resource = replica.getResourceName();
int curPartitionCountForResource = node.getAssignedPartitionsByResource(resource).size();
- int doubleMaxPartitionCountForResource =
- 2 * clusterContext.getEstimatedMaxPartitionByResource(resource);
- // The score measures the twice the max allowed count versus current counts
- // The returned value is a measurement of remaining quota ratio, in the case of exceeding allowed counts, return 0
- return Math.max(((float) doubleMaxPartitionCountForResource - curPartitionCountForResource)
- / doubleMaxPartitionCountForResource, 0);
+ int estimatedMaxPartitionCountForResource =
+ clusterContext.getEstimatedMaxPartitionByResource(resource);
+ return computeUtilizationScore(estimatedMaxPartitionCountForResource,
+ curPartitionCountForResource);
}
}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/ResourceTopStateAntiAffinityConstraint.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/ResourceTopStateAntiAffinityConstraint.java
index 8681dc5..f0f9e13 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/ResourceTopStateAntiAffinityConstraint.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/ResourceTopStateAntiAffinityConstraint.java
@@ -28,26 +28,17 @@ import org.apache.helix.controller.rebalancer.waged.model.ClusterContext;
* The higher number the number of top state partitions assigned to the instance, the lower the
* score, vice versa.
*/
-class ResourceTopStateAntiAffinityConstraint extends SoftConstraint {
- private static final float MAX_SCORE = 1f;
- private static final float MIN_SCORE = 0f;
-
- ResourceTopStateAntiAffinityConstraint() {
- super(MAX_SCORE, MIN_SCORE);
- }
-
+class ResourceTopStateAntiAffinityConstraint extends UsageSoftConstraint {
@Override
- protected float getAssignmentScore(AssignableNode node, AssignableReplica replica,
+ protected double getAssignmentScore(AssignableNode node, AssignableReplica replica,
ClusterContext clusterContext) {
if (!replica.isReplicaTopState()) {
- return (getMaxScore() + getMinScore()) / 2.0f;
+ // For non top state replica, this constraint is not applicable.
+ // So return zero on any assignable node candidate.
+ return 0;
}
-
int curTopPartitionCountForResource = node.getAssignedTopStatePartitionsCount();
- int doubleMaxTopStateCount = 2 * clusterContext.getEstimatedMaxTopStateCount();
-
- return Math.max(
- ((float) doubleMaxTopStateCount - curTopPartitionCountForResource) / doubleMaxTopStateCount,
- 0);
+ int estimatedMaxTopStateCount = clusterContext.getEstimatedMaxTopStateCount();
+ return computeUtilizationScore(estimatedMaxTopStateCount, curTopPartitionCountForResource);
}
}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/SoftConstraint.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/SoftConstraint.java
index f44d262..21bed84 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/SoftConstraint.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/SoftConstraint.java
@@ -30,8 +30,8 @@ import org.apache.helix.controller.rebalancer.waged.model.ClusterContext;
* The lower score the score, the worse the assignment; Intuitively, the assignment is penalized.
*/
abstract class SoftConstraint {
- private float _maxScore = 1000f;
- private float _minScore = -1000f;
+ private final double _maxScore;
+ private final double _minScore;
interface NormalizeFunction {
/**
@@ -40,13 +40,7 @@ abstract class SoftConstraint {
* @param originScore The origin score
* @return The normalized value between (0, 1)
*/
- float scale(float originScore);
- }
-
- /**
- * Default constructor, uses default min/max scores
- */
- SoftConstraint() {
+ double scale(double originScore);
}
/**
@@ -54,25 +48,25 @@ abstract class SoftConstraint {
* @param maxScore The max score
* @param minScore The min score
*/
- SoftConstraint(float maxScore, float minScore) {
+ SoftConstraint(double maxScore, double minScore) {
_maxScore = maxScore;
_minScore = minScore;
}
- float getMaxScore() {
+ protected double getMaxScore() {
return _maxScore;
}
- float getMinScore() {
+ protected double getMinScore() {
return _minScore;
}
/**
* Evaluate and give a score for an potential assignment partition -> instance
* Child class only needs to care about how the score is implemented
- * @return The score of the assignment in float value
+ * @return The score of the assignment in double value
*/
- protected abstract float getAssignmentScore(AssignableNode node, AssignableReplica replica,
+ protected abstract double getAssignmentScore(AssignableNode node, AssignableReplica replica,
ClusterContext clusterContext);
/**
@@ -80,7 +74,7 @@ abstract class SoftConstraint {
* It's the only exposed method to the caller
* @return The score is normalized to be within MinScore and MaxScore
*/
- float getAssignmentNormalizedScore(AssignableNode node, AssignableReplica replica,
+ double getAssignmentNormalizedScore(AssignableNode node, AssignableReplica replica,
ClusterContext clusterContext) {
return getNormalizeFunction().scale(getAssignmentScore(node, replica, clusterContext));
}
@@ -90,7 +84,7 @@ abstract class SoftConstraint {
* Child class could override the method and customize the method on its own
* @return The MinMaxScaler instance by default
*/
- NormalizeFunction getNormalizeFunction() {
+ protected NormalizeFunction getNormalizeFunction() {
return (score) -> (score - getMinScore()) / (getMaxScore() - getMinScore());
}
}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/UsageSoftConstraint.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/UsageSoftConstraint.java
new file mode 100644
index 0000000..9fe0632
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/UsageSoftConstraint.java
@@ -0,0 +1,88 @@
+package org.apache.helix.controller.rebalancer.waged.constraints;
+
+/*
+ * 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.
+ */
+
+import java.util.function.Function;
+
+import org.apache.commons.math3.analysis.function.Sigmoid;
+
+/**
+ * The soft constraint that evaluates the assignment proposal based on usage.
+ */
+abstract class UsageSoftConstraint extends SoftConstraint {
+ private static final double MAX_SCORE = 1f;
+ private static final double MIN_SCORE = 0f;
+ /**
+ * Alpha is used to adjust the curve of sigmoid function.
+ * Intuitively, this is for tolerating the inaccuracy of the estimation.
+ * Ideally, if we have the prefect estimation, we can use a segmented function here, which
+ * scores the assignment with 1.0 if projected usage is below the estimation, and scores 0.0
+ * if the projected usage exceeds the estimation. However, in reality, it is hard to get a
+ * prefect estimation. With the curve of sigmoid, the algorithm reacts differently and
+ * reasonally even the usage is a little bit more or less than the estimation for a certain
+ * extend.
+ * As tested, when we have the input number which surrounds 1, the default alpha value will
+ * ensure a curve that has sigmoid(0.95) = 0.90, sigmoid(1.05) = 0.1. Meaning the constraint
+ * can handle the estimation inaccuracy of +-5%.
+ * To adjust the curve:
+ * 1. Smaller alpha will increase the curve's scope. So the function will be handler a wilder
+ * range of inaccuracy. However, the downside is more random movements since the evenness
+ * score would be more changable and nondefinitive.
+ * 2. Larger alpha will decrease the curve's scope. In that case, we might want to change to
+ * use segmented function so as to speed up the algorthm.
+ **/
+ private static final int DEFAULT_ALPHA = 44;
+ private static final Sigmoid SIGMOID = new Sigmoid();
+
+ UsageSoftConstraint() {
+ super(MAX_SCORE, MIN_SCORE);
+ }
+
+ /**
+ * Compute the utilization score based on the estimated and current usage numbers.
+ * The score = currentUsage / estimatedUsage.
+ * In short, a smaller score means better assignment proposal.
+ *
+ * @param estimatedUsage The estimated usage that is between [0.0, 1.0]
+ * @param currentUsage The current usage that is between [0.0, 1.0]
+ * @return The score between [0.0, 1.0] that evaluates the utilization.
+ */
+ protected double computeUtilizationScore(double estimatedUsage, double currentUsage) {
+ if (estimatedUsage == 0) {
+ return 0;
+ }
+ return currentUsage / estimatedUsage;
+ }
+
+ /**
+ * Compute evaluation score based on the utilization data.
+ * The normalized score is evaluated using a sigmoid function.
+ * When the usage is smaller than 1.0, the constraint returns a value that is very close to the
+ * max score.
+ * When the usage is close or larger than 1.0, the constraint returns a score that is very close
+ * to the min score. Note even in this case, more usage will still be assigned with a
+ * smaller score.
+ */
+ @Override
+ protected NormalizeFunction getNormalizeFunction() {
+ return (score) -> SIGMOID.value(-(score - 1) * DEFAULT_ALPHA) * (MAX_SCORE
+ - MIN_SCORE);
+ }
+}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/model/AssignableNode.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/model/AssignableNode.java
index 919acb3..6beda6a 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/model/AssignableNode.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/model/AssignableNode.java
@@ -58,8 +58,6 @@ public class AssignableNode implements Comparable<AssignableNode> {
private Map<String, Map<String, AssignableReplica>> _currentAssignedReplicaMap;
// A map of <capacity key, capacity value> that tracks the current available node capacity
private Map<String, Integer> _remainingCapacity;
- // The maximum capacity utilization (0.0 - 1.0) across all the capacity categories.
- private float _highestCapacityUtilization;
/**
* Update the node with a ClusterDataCache. This resets the current assignment and recalculates
@@ -81,7 +79,6 @@ public class AssignableNode implements Comparable<AssignableNode> {
_remainingCapacity = new HashMap<>(instanceCapacity);
_maxPartition = clusterConfig.getMaxPartitionsPerInstance();
_currentAssignedReplicaMap = new HashMap<>();
- _highestCapacityUtilization = 0f;
}
/**
@@ -144,8 +141,6 @@ public class AssignableNode implements Comparable<AssignableNode> {
}
AssignableReplica removedReplica = partitionMap.remove(partitionName);
- // Recalculate utilization because of release
- _highestCapacityUtilization = 0;
removedReplica.getCapacity().entrySet().stream()
.forEach(entry -> updateCapacityAndUtilization(entry.getKey(), -1 * entry.getValue()));
}
@@ -221,14 +216,22 @@ public class AssignableNode implements Comparable<AssignableNode> {
/**
* Return the most concerning capacity utilization number for evenly partition assignment.
- * The method dynamically returns the highest utilization number among all the capacity
- * categories.
+ * The method dynamically calculates the projected highest utilization number among all the
+ * capacity categories assuming the new capacity usage is added to the node.
* For example, if the current node usage is {CPU: 0.9, MEM: 0.4, DISK: 0.6}. Then this call shall
* return 0.9.
+ * @param newUsage the proposed new additional capacity usage.
* @return The highest utilization number of the node among all the capacity category.
*/
- public float getHighestCapacityUtilization() {
- return _highestCapacityUtilization;
+ public float getProjectedHighestUtilization(Map<String, Integer> newUsage) {
+ float highestCapacityUtilization = 0;
+ for (String capacityKey : _maxAllowedCapacity.keySet()) {
+ float capacityValue = _maxAllowedCapacity.get(capacityKey);
+ float utilization = (capacityValue - _remainingCapacity.get(capacityKey) + newUsage
+ .getOrDefault(capacityKey, 0)) / capacityValue;
+ highestCapacityUtilization = Math.max(highestCapacityUtilization, utilization);
+ }
+ return highestCapacityUtilization;
}
public String getInstanceName() {
@@ -337,10 +340,6 @@ public class AssignableNode implements Comparable<AssignableNode> {
}
int newCapacity = _remainingCapacity.get(capacityKey) - usage;
_remainingCapacity.put(capacityKey, newCapacity);
- // For the purpose of constraint calculation, the max utilization cannot be larger than 100%.
- float utilization = Math.min((float) (_maxAllowedCapacity.get(capacityKey) - newCapacity)
- / _maxAllowedCapacity.get(capacityKey), 1);
- _highestCapacityUtilization = Math.max(_highestCapacityUtilization, utilization);
}
/**
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/model/ClusterContext.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/model/ClusterContext.java
index 892cad3..4705be5 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/model/ClusterContext.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/model/ClusterContext.java
@@ -35,14 +35,14 @@ import org.apache.helix.model.ResourceAssignment;
* This class tracks the rebalance-related global cluster status.
*/
public class ClusterContext {
- private final static float ERROR_MARGIN_FOR_ESTIMATED_MAX_COUNT = 1.1f;
-
// This estimation helps to ensure global partition count evenness
private final int _estimatedMaxPartitionCount;
// This estimation helps to ensure global top state replica count evenness
private final int _estimatedMaxTopStateCount;
// This estimation helps to ensure per-resource partition count evenness
private final Map<String, Integer> _estimatedMaxPartitionByResource = new HashMap<>();
+ // This estimation helps to ensure global resource usage evenness.
+ private final float _estimatedMaxUtilization;
// map{zoneName : map{resourceName : set(partitionNames)}}
private Map<String, Map<String, Set<String>>> _assignmentForFaultZoneMap = new HashMap<>();
@@ -55,12 +55,15 @@ public class ClusterContext {
/**
* Construct the cluster context based on the current instance status.
* @param replicaSet All the partition replicas that are managed by the rebalancer
- * @param instanceCount The count of all the active instances that can be used to host partitions.
+ * @param nodeSet All the active nodes that are managed by the rebalancer
*/
- ClusterContext(Set<AssignableReplica> replicaSet, int instanceCount,
+ ClusterContext(Set<AssignableReplica> replicaSet, Set<AssignableNode> nodeSet,
Map<String, ResourceAssignment> baselineAssignment, Map<String, ResourceAssignment> bestPossibleAssignment) {
+ int instanceCount = nodeSet.size();
int totalReplicas = 0;
int totalTopStateReplicas = 0;
+ Map<String, Integer> totalUsage = new HashMap<>();
+ Map<String, Integer> totalCapacity = new HashMap<>();
for (Map.Entry<String, List<AssignableReplica>> entry : replicaSet.stream()
.collect(Collectors.groupingBy(AssignableReplica::getResourceName))
@@ -71,9 +74,32 @@ public class ClusterContext {
int replicaCnt = Math.max(1, estimateAvgReplicaCount(replicas, instanceCount));
_estimatedMaxPartitionByResource.put(entry.getKey(), replicaCnt);
- totalTopStateReplicas += entry.getValue().stream().filter(AssignableReplica::isReplicaTopState).count();
+ for (AssignableReplica replica : entry.getValue()) {
+ if (replica.isReplicaTopState()) {
+ totalTopStateReplicas += 1;
+ }
+ replica.getCapacity().entrySet().stream().forEach(capacityEntry -> totalUsage
+ .compute(capacityEntry.getKey(),
+ (k, v) -> (v == null) ? capacityEntry.getValue() : (v + capacityEntry.getValue())));
+ }
+ }
+ nodeSet.stream().forEach(node -> node.getMaxCapacity().entrySet().stream().forEach(
+ capacityEntry -> totalCapacity.compute(capacityEntry.getKey(),
+ (k, v) -> (v == null) ? capacityEntry.getValue() : (v + capacityEntry.getValue()))));
+
+ if (totalCapacity.isEmpty()) {
+ // If no capacity is configured, we treat the cluster as fully utilized.
+ _estimatedMaxUtilization = 1f;
+ } else {
+ float estimatedMaxUsage = 0;
+ for (String capacityKey : totalCapacity.keySet()) {
+ int maxCapacity = totalCapacity.get(capacityKey);
+ int usage = totalUsage.getOrDefault(capacityKey, 0);
+ float utilization = (maxCapacity == 0) ? 1 : (float) usage / maxCapacity;
+ estimatedMaxUsage = Math.max(estimatedMaxUsage, utilization);
+ }
+ _estimatedMaxUtilization = estimatedMaxUsage;
}
-
_estimatedMaxPartitionCount = estimateAvgReplicaCount(totalReplicas, instanceCount);
_estimatedMaxTopStateCount = estimateAvgReplicaCount(totalTopStateReplicas, instanceCount);
_baselineAssignment = baselineAssignment;
@@ -105,6 +131,10 @@ public class ClusterContext {
return _estimatedMaxTopStateCount;
}
+ public float getEstimatedMaxUtilization() {
+ return _estimatedMaxUtilization;
+ }
+
public Set<String> getPartitionsForResourceAndFaultZone(String resourceName, String faultZoneId) {
return _assignmentForFaultZoneMap.getOrDefault(faultZoneId, Collections.emptyMap())
.getOrDefault(resourceName, Collections.emptySet());
@@ -131,6 +161,12 @@ public class ClusterContext {
}
private int estimateAvgReplicaCount(int replicaCount, int instanceCount) {
- return (int) Math.ceil((float) replicaCount / instanceCount * ERROR_MARGIN_FOR_ESTIMATED_MAX_COUNT);
+ // Use the floor to ensure evenness.
+ // Note if we calculate estimation based on ceil, we might have some low usage participants.
+ // For example, if the evaluation is between 1 and 2. While we use 2, many participants will be
+ // allocated with 2 partitions. And the other participants only has 0 partitions. Otherwise,
+ // if we use 1, most participant will have 1 partition assigned and several participant has 2
+ // partitions. The later scenario is what we want to achieve.
+ return (int) Math.floor((float) replicaCount / instanceCount);
}
}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/model/ClusterModelProvider.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/model/ClusterModelProvider.java
index 4c32f4c..4722e7d 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/model/ClusterModelProvider.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/model/ClusterModelProvider.java
@@ -86,7 +86,7 @@ public class ClusterModelProvider {
// Construct and initialize cluster context.
ClusterContext context = new ClusterContext(
replicaMap.values().stream().flatMap(Set::stream).collect(Collectors.toSet()),
- activeInstances.size(), baselineAssignment, bestPossibleAssignment);
+ assignableNodes, baselineAssignment, bestPossibleAssignment);
// Initial the cluster context with the allocated assignments.
context.setAssignmentForFaultZoneMap(mapAssignmentToFaultZone(assignableNodes));
diff --git a/helix-core/src/main/resources/soft-constraint-weight.properties b/helix-core/src/main/resources/soft-constraint-weight.properties
index 3e87c9d..c3c7931 100644
--- a/helix-core/src/main/resources/soft-constraint-weight.properties
+++ b/helix-core/src/main/resources/soft-constraint-weight.properties
@@ -17,8 +17,10 @@
# under the License.
#
-PartitionMovementConstraint=1f
-InstancePartitionsCountConstraint=0.3f
-ResourcePartitionAntiAffinityConstraint=0.1f
-ResourceTopStateAntiAffinityConstraint=0.1f
-MaxCapacityUsageInstanceConstraint=0.5f
\ No newline at end of file
+# Define the constraint weights for the WAGED rebalancer in this file.
+#
+# PartitionMovementConstraint=1f
+# InstancePartitionsCountConstraint=0.15f
+# ResourcePartitionAntiAffinityConstraint=0.05f
+# ResourceTopStateAntiAffinityConstraint=0.3f
+# MaxCapacityUsageInstanceConstraint=0.6f
diff --git a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestConstraintBasedAlgorithm.java b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestConstraintBasedAlgorithm.java
index b2deaef..e0e2eb3 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestConstraintBasedAlgorithm.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestConstraintBasedAlgorithm.java
@@ -44,7 +44,7 @@ public class TestConstraintBasedAlgorithm {
HardConstraint mockHardConstraint = mock(HardConstraint.class);
SoftConstraint mockSoftConstraint = mock(SoftConstraint.class);
when(mockHardConstraint.isAssignmentValid(any(), any(), any())).thenReturn(false);
- when(mockSoftConstraint.getAssignmentNormalizedScore(any(), any(), any())).thenReturn(1.0f);
+ when(mockSoftConstraint.getAssignmentNormalizedScore(any(), any(), any())).thenReturn(1.0);
_algorithm = new ConstraintBasedAlgorithm(ImmutableList.of(mockHardConstraint),
ImmutableMap.of(mockSoftConstraint, 1f));
@@ -61,7 +61,7 @@ public class TestConstraintBasedAlgorithm {
HardConstraint mockHardConstraint = mock(HardConstraint.class);
SoftConstraint mockSoftConstraint = mock(SoftConstraint.class);
when(mockHardConstraint.isAssignmentValid(any(), any(), any())).thenReturn(true);
- when(mockSoftConstraint.getAssignmentNormalizedScore(any(), any(), any())).thenReturn(1.0f);
+ when(mockSoftConstraint.getAssignmentNormalizedScore(any(), any(), any())).thenReturn(1.0);
_algorithm = new ConstraintBasedAlgorithm(ImmutableList.of(mockHardConstraint),
ImmutableMap.of(mockSoftConstraint, 1f));
ClusterModel clusterModel = new ClusterModelTestHelper().getDefaultClusterModel();
diff --git a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestInstancePartitionsCountConstraint.java b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestInstancePartitionsCountConstraint.java
index 63622e2..a54379e 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestInstancePartitionsCountConstraint.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestInstancePartitionsCountConstraint.java
@@ -38,26 +38,26 @@ public class TestInstancePartitionsCountConstraint {
@Test
public void testWhenInstanceIsIdle() {
when(_testNode.getAssignedReplicaCount()).thenReturn(0);
- float score =
+ double score =
_constraint.getAssignmentNormalizedScore(_testNode, _testReplica, _clusterContext);
- Assert.assertEquals(score, 1.0f);
+ Assert.assertEquals(score, 1.0);
}
@Test
public void testWhenInstanceIsFull() {
when(_testNode.getAssignedReplicaCount()).thenReturn(10);
when(_clusterContext.getEstimatedMaxPartitionCount()).thenReturn(10);
- float score =
+ double score =
_constraint.getAssignmentNormalizedScore(_testNode, _testReplica, _clusterContext);
- Assert.assertEquals(score, 0.5f);
+ Assert.assertEquals(score, 0.5);
}
@Test
public void testWhenInstanceHalfOccupied() {
when(_testNode.getAssignedReplicaCount()).thenReturn(10);
when(_clusterContext.getEstimatedMaxPartitionCount()).thenReturn(20);
- float score =
+ double score =
_constraint.getAssignmentNormalizedScore(_testNode, _testReplica, _clusterContext);
- Assert.assertEquals(score, 0.75f);
+ Assert.assertTrue(score > 0.99);
}
}
diff --git a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestMaxCapacityUsageInstanceConstraint.java b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestMaxCapacityUsageInstanceConstraint.java
index 7026dee..1bc0793 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestMaxCapacityUsageInstanceConstraint.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestMaxCapacityUsageInstanceConstraint.java
@@ -19,10 +19,6 @@ package org.apache.helix.controller.rebalancer.waged.constraints;
* under the License.
*/
-import static org.mockito.Mockito.CALLS_REAL_METHODS;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
import org.apache.helix.controller.rebalancer.waged.model.AssignableNode;
import org.apache.helix.controller.rebalancer.waged.model.AssignableReplica;
import org.apache.helix.controller.rebalancer.waged.model.ClusterContext;
@@ -30,6 +26,10 @@ import org.testng.Assert;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+import static org.mockito.Matchers.anyMap;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
public class TestMaxCapacityUsageInstanceConstraint {
private AssignableReplica _testReplica;
private AssignableNode _testNode;
@@ -38,19 +38,19 @@ public class TestMaxCapacityUsageInstanceConstraint {
@BeforeMethod
public void setUp() {
- _testNode = mock(AssignableNode.class, CALLS_REAL_METHODS);
+ _testNode = mock(AssignableNode.class);
_testReplica = mock(AssignableReplica.class);
_clusterContext = mock(ClusterContext.class);
}
@Test
public void testGetNormalizedScore() {
- when(_testNode.getHighestCapacityUtilization()).thenReturn(0.8f);
- float score =
- _constraint.getAssignmentScore(_testNode, _testReplica, _clusterContext);
- Assert.assertEquals(score, 0.6f);
- float normalizedScore =
+ when(_testNode.getProjectedHighestUtilization(anyMap())).thenReturn(0.8f);
+ when(_clusterContext.getEstimatedMaxUtilization()).thenReturn(1f);
+ double score = _constraint.getAssignmentScore(_testNode, _testReplica, _clusterContext);
+ Assert.assertTrue(score > 0.99);
+ double normalizedScore =
_constraint.getAssignmentNormalizedScore(_testNode, _testReplica, _clusterContext);
- Assert.assertEquals(normalizedScore, 0.6f);
+ Assert.assertTrue(normalizedScore > 0.99);
}
}
diff --git a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestPartitionMovementConstraint.java b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestPartitionMovementConstraint.java
index d3af35e..2629c25 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestPartitionMovementConstraint.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestPartitionMovementConstraint.java
@@ -59,11 +59,11 @@ public class TestPartitionMovementConstraint {
public void testGetAssignmentScoreWhenBestPossibleBaselineMissing() {
when(_clusterContext.getBaselineAssignment()).thenReturn(Collections.emptyMap());
when(_clusterContext.getBestPossibleAssignment()).thenReturn(Collections.emptyMap());
- float score = _constraint.getAssignmentScore(_testNode, _testReplica, _clusterContext);
- float normalizedScore =
+ double score = _constraint.getAssignmentScore(_testNode, _testReplica, _clusterContext);
+ double normalizedScore =
_constraint.getAssignmentNormalizedScore(_testNode, _testReplica, _clusterContext);
- Assert.assertEquals(score, 0f);
- Assert.assertEquals(normalizedScore, 0f);
+ Assert.assertEquals(score, 0.0);
+ Assert.assertEquals(normalizedScore, 0.0);
}
@Test
@@ -77,20 +77,20 @@ public class TestPartitionMovementConstraint {
when(_clusterContext.getBestPossibleAssignment()).thenReturn(assignmentMap);
// when the calculated states are both equal to the replica's current state
when(_testReplica.getReplicaState()).thenReturn("Master");
- float score = _constraint.getAssignmentScore(_testNode, _testReplica, _clusterContext);
- float normalizedScore =
+ double score = _constraint.getAssignmentScore(_testNode, _testReplica, _clusterContext);
+ double normalizedScore =
_constraint.getAssignmentNormalizedScore(_testNode, _testReplica, _clusterContext);
- Assert.assertEquals(score, 1f);
- Assert.assertEquals(normalizedScore, 1f);
+ Assert.assertEquals(score, 1.0);
+ Assert.assertEquals(normalizedScore, 1.0);
// when the calculated states are both different from the replica's current state
when(_testReplica.getReplicaState()).thenReturn("Slave");
score = _constraint.getAssignmentScore(_testNode, _testReplica, _clusterContext);
normalizedScore =
_constraint.getAssignmentNormalizedScore(_testNode, _testReplica, _clusterContext);
- Assert.assertEquals(score, 0.5f);
- Assert.assertEquals(normalizedScore, 0.5f);
+ Assert.assertEquals(score, 0.5);
+ Assert.assertEquals(normalizedScore, 0.5);
}
@Test
@@ -107,21 +107,21 @@ public class TestPartitionMovementConstraint {
.thenReturn(ImmutableMap.of(RESOURCE, bestPossibleResourceAssignment));
// when the replica's state matches with best possible only
when(_testReplica.getReplicaState()).thenReturn("Master");
- float score = _constraint.getAssignmentScore(_testNode, _testReplica, _clusterContext);
- float normalizedScore =
+ double score = _constraint.getAssignmentScore(_testNode, _testReplica, _clusterContext);
+ double normalizedScore =
_constraint.getAssignmentNormalizedScore(_testNode, _testReplica, _clusterContext);
- Assert.assertEquals(score, 0.875f);
- Assert.assertEquals(normalizedScore, 0.875f);
+ Assert.assertEquals(score, 1.0);
+ Assert.assertEquals(normalizedScore, 1.0);
// when the replica's state matches with baseline only
when(_testReplica.getReplicaState()).thenReturn("Slave");
score = _constraint.getAssignmentScore(_testNode, _testReplica, _clusterContext);
normalizedScore =
_constraint.getAssignmentNormalizedScore(_testNode, _testReplica, _clusterContext);
- // The calculated score is lower than previous 0.875f cause the replica's state matches with
+ // The calculated score is lower than previous value cause the replica's state matches with
// best possible is preferred
- Assert.assertEquals(score, 0.625f);
- Assert.assertEquals(normalizedScore, 0.625f);
+ Assert.assertEquals(score, 0.5);
+ Assert.assertEquals(normalizedScore, 0.5);
}
}
diff --git a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestResourcePartitionAntiAffinityConstraint.java b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestResourcePartitionAntiAffinityConstraint.java
index c6830cf..e3381d1 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestResourcePartitionAntiAffinityConstraint.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestResourcePartitionAntiAffinityConstraint.java
@@ -47,10 +47,10 @@ public class TestResourcePartitionAntiAffinityConstraint {
ImmutableSet.of(TEST_PARTITION + "1", TEST_PARTITION + "2", TEST_PARTITION + "3"));
when(_clusterContext.getEstimatedMaxPartitionByResource(TEST_RESOURCE)).thenReturn(10);
- float score = _constraint.getAssignmentScore(_testNode, _testReplica, _clusterContext);
- float normalizedScore = _constraint.getAssignmentNormalizedScore(_testNode, _testReplica, _clusterContext);
- Assert.assertEquals(score, 0.85f);
- Assert.assertEquals(normalizedScore, 0.85f);
+ double score = _constraint.getAssignmentScore(_testNode, _testReplica, _clusterContext);
+ double normalizedScore = _constraint.getAssignmentNormalizedScore(_testNode, _testReplica, _clusterContext);
+ Assert.assertTrue(score > 0.99);
+ Assert.assertTrue(score > 0.99);
}
@Test
@@ -59,9 +59,9 @@ public class TestResourcePartitionAntiAffinityConstraint {
when(_testNode.getAssignedPartitionsByResource(TEST_RESOURCE)).thenReturn(Collections.emptySet());
when(_clusterContext.getEstimatedMaxPartitionByResource(TEST_RESOURCE)).thenReturn(10);
- float score = _constraint.getAssignmentScore(_testNode, _testReplica, _clusterContext);
- float normalizedScore = _constraint.getAssignmentNormalizedScore(_testNode, _testReplica, _clusterContext);
- Assert.assertEquals(score, 1f);
- Assert.assertEquals(normalizedScore, 1f);
+ double score = _constraint.getAssignmentScore(_testNode, _testReplica, _clusterContext);
+ double normalizedScore = _constraint.getAssignmentNormalizedScore(_testNode, _testReplica, _clusterContext);
+ Assert.assertEquals(score, 1.0);
+ Assert.assertEquals(normalizedScore, 1.0);
}
}
diff --git a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestResourceTopStateAntiAffinityConstraint.java b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestResourceTopStateAntiAffinityConstraint.java
index c3414d4..ca2421f 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestResourceTopStateAntiAffinityConstraint.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestResourceTopStateAntiAffinityConstraint.java
@@ -47,11 +47,11 @@ public class TestResourceTopStateAntiAffinityConstraint {
@Test
public void testGetAssignmentScoreWhenReplicaNotTopState() {
when(_testReplica.isReplicaTopState()).thenReturn(false);
- float score = _constraint.getAssignmentScore(_testNode, _testReplica, _clusterContext);
- float normalizedScore =
+ double score = _constraint.getAssignmentScore(_testNode, _testReplica, _clusterContext);
+ double normalizedScore =
_constraint.getAssignmentNormalizedScore(_testNode, _testReplica, _clusterContext);
- Assert.assertEquals(score, 0.5f);
- Assert.assertEquals(normalizedScore, 0.5f);
+ Assert.assertEquals(score, 0.0);
+ Assert.assertEquals(normalizedScore, 0.0);
verifyZeroInteractions(_testNode);
verifyZeroInteractions(_clusterContext);
}
@@ -61,11 +61,11 @@ public class TestResourceTopStateAntiAffinityConstraint {
when(_testReplica.isReplicaTopState()).thenReturn(true);
when(_testNode.getAssignedTopStatePartitionsCount()).thenReturn(20);
when(_clusterContext.getEstimatedMaxTopStateCount()).thenReturn(20);
- float score = _constraint.getAssignmentScore(_testNode, _testReplica, _clusterContext);
- float normalizedScore =
+ double score = _constraint.getAssignmentScore(_testNode, _testReplica, _clusterContext);
+ double normalizedScore =
_constraint.getAssignmentNormalizedScore(_testNode, _testReplica, _clusterContext);
- Assert.assertEquals(score, 0.5f);
- Assert.assertEquals(normalizedScore, 0.5f);
+ Assert.assertEquals(score, 0.5);
+ Assert.assertEquals(normalizedScore, 0.5);
}
@Test
@@ -73,10 +73,10 @@ public class TestResourceTopStateAntiAffinityConstraint {
when(_testReplica.isReplicaTopState()).thenReturn(true);
when(_testNode.getAssignedTopStatePartitionsCount()).thenReturn(0);
when(_clusterContext.getEstimatedMaxTopStateCount()).thenReturn(20);
- float score = _constraint.getAssignmentScore(_testNode, _testReplica, _clusterContext);
- float normalizedScore =
+ double score = _constraint.getAssignmentScore(_testNode, _testReplica, _clusterContext);
+ double normalizedScore =
_constraint.getAssignmentNormalizedScore(_testNode, _testReplica, _clusterContext);
- Assert.assertEquals(score, 1f);
- Assert.assertEquals(normalizedScore, 1f);
+ Assert.assertEquals(score, 1.0);
+ Assert.assertEquals(normalizedScore, 1.0);
}
}
diff --git a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestSoftConstraintNormalizeFunction.java b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestSoftConstraintNormalizeFunction.java
index b523959..ad34705 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestSoftConstraintNormalizeFunction.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/constraints/TestSoftConstraintNormalizeFunction.java
@@ -32,14 +32,14 @@ public class TestSoftConstraintNormalizeFunction {
int minScore = 0;
SoftConstraint softConstraint = new SoftConstraint(maxScore, minScore) {
@Override
- protected float getAssignmentScore(AssignableNode node, AssignableReplica replica,
+ protected double getAssignmentScore(AssignableNode node, AssignableReplica replica,
ClusterContext clusterContext) {
return 0;
}
};
for (int i = minScore; i <= maxScore; i++) {
- float normalized = softConstraint.getNormalizeFunction().scale(i);
+ double normalized = softConstraint.getNormalizeFunction().scale(i);
Assert.assertTrue(normalized <= 1 && normalized >= 0,
String.format("input: %s, output: %s", i, normalized));
}
diff --git a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/model/AbstractTestClusterModel.java b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/model/AbstractTestClusterModel.java
index 54cbd41..0fec67b 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/model/AbstractTestClusterModel.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/model/AbstractTestClusterModel.java
@@ -193,4 +193,12 @@ public abstract class AbstractTestClusterModel {
}
return assignmentSet;
}
+
+ protected Set<AssignableNode> generateNodes(ResourceControllerDataProvider testCache) {
+ Set<AssignableNode> nodeSet = new HashSet<>();
+ testCache.getInstanceConfigMap().values().stream()
+ .forEach(config -> nodeSet.add(new AssignableNode(testCache.getClusterConfig(),
+ testCache.getInstanceConfigMap().get(_testInstanceId), config.getInstanceName())));
+ return nodeSet;
+ }
}
diff --git a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/model/ClusterModelTestHelper.java b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/model/ClusterModelTestHelper.java
index 585c26f..131d92a 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/model/ClusterModelTestHelper.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/model/ClusterModelTestHelper.java
@@ -21,7 +21,6 @@ package org.apache.helix.controller.rebalancer.waged.model;
import java.io.IOException;
import java.util.Collections;
-import java.util.HashSet;
import java.util.Set;
import org.apache.helix.controller.dataproviders.ResourceControllerDataProvider;
@@ -35,15 +34,7 @@ public class ClusterModelTestHelper extends AbstractTestClusterModel {
Set<AssignableNode> assignableNodes = generateNodes(testCache);
ClusterContext context =
- new ClusterContext(assignableReplicas, 2, Collections.emptyMap(), Collections.emptyMap());
+ new ClusterContext(assignableReplicas, assignableNodes, Collections.emptyMap(), Collections.emptyMap());
return new ClusterModel(context, assignableReplicas, assignableNodes);
}
-
- private Set<AssignableNode> generateNodes(ResourceControllerDataProvider testCache) {
- Set<AssignableNode> nodeSet = new HashSet<>();
- testCache.getInstanceConfigMap().values().stream()
- .forEach(config -> nodeSet.add(new AssignableNode(testCache.getClusterConfig(),
- testCache.getInstanceConfigMap().get(_testInstanceId), config.getInstanceName())));
- return nodeSet;
- }
}
diff --git a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/model/TestAssignableNode.java b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/model/TestAssignableNode.java
index 168226d..187c423 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/model/TestAssignableNode.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/model/TestAssignableNode.java
@@ -19,10 +19,9 @@ package org.apache.helix.controller.rebalancer.waged.model;
* under the License.
*/
-import static org.mockito.Mockito.when;
-
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -38,6 +37,8 @@ import org.testng.Assert;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
+import static org.mockito.Mockito.when;
+
public class TestAssignableNode extends AbstractTestClusterModel {
@BeforeClass
public void initialize() {
@@ -67,7 +68,8 @@ public class TestAssignableNode extends AbstractTestClusterModel {
assignableNode.assignInitBatch(assignmentSet);
Assert.assertEquals(assignableNode.getAssignedPartitionsMap(), expectedAssignment);
Assert.assertEquals(assignableNode.getAssignedReplicaCount(), 4);
- Assert.assertEquals(assignableNode.getHighestCapacityUtilization(), 16.0 / 20.0, 0.005);
+ Assert.assertEquals(assignableNode.getProjectedHighestUtilization(Collections.EMPTY_MAP),
+ 16.0 / 20.0, 0.005);
Assert.assertEquals(assignableNode.getMaxCapacity(), _capacityDataMap);
Assert.assertEquals(assignableNode.getMaxPartition(), 5);
Assert.assertEquals(assignableNode.getInstanceTags(), _testInstanceTags);
@@ -107,7 +109,8 @@ public class TestAssignableNode extends AbstractTestClusterModel {
Assert.assertEquals(assignableNode.getAssignedPartitionsMap(), expectedAssignment);
Assert.assertEquals(assignableNode.getAssignedReplicaCount(), 3);
- Assert.assertEquals(assignableNode.getHighestCapacityUtilization(), 11.0 / 20.0, 0.005);
+ Assert.assertEquals(assignableNode.getProjectedHighestUtilization(Collections.EMPTY_MAP),
+ 11.0 / 20.0, 0.005);
Assert.assertEquals(assignableNode.getMaxCapacity(), _capacityDataMap);
Assert.assertEquals(assignableNode.getMaxPartition(), 5);
Assert.assertEquals(assignableNode.getInstanceTags(), _testInstanceTags);
@@ -140,7 +143,8 @@ public class TestAssignableNode extends AbstractTestClusterModel {
Assert.assertEquals(assignableNode.getAssignedPartitionsMap(), expectedAssignment);
Assert.assertEquals(assignableNode.getAssignedReplicaCount(), 4);
- Assert.assertEquals(assignableNode.getHighestCapacityUtilization(), 16.0 / 20.0, 0.005);
+ Assert.assertEquals(assignableNode.getProjectedHighestUtilization(Collections.EMPTY_MAP),
+ 16.0 / 20.0, 0.005);
Assert.assertEquals(assignableNode.getMaxCapacity(), _capacityDataMap);
Assert.assertEquals(assignableNode.getMaxPartition(), 5);
Assert.assertEquals(assignableNode.getInstanceTags(), _testInstanceTags);
diff --git a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/model/TestClusterContext.java b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/model/TestClusterContext.java
index d8b93c0..732ae8f 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/model/TestClusterContext.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/model/TestClusterContext.java
@@ -44,10 +44,11 @@ public class TestClusterContext extends AbstractTestClusterModel {
ResourceControllerDataProvider testCache = setupClusterDataCache();
Set<AssignableReplica> assignmentSet = generateReplicas(testCache);
- ClusterContext context = new ClusterContext(assignmentSet, 2, new HashMap<>(), new HashMap<>());
+ ClusterContext context =
+ new ClusterContext(assignmentSet, generateNodes(testCache), new HashMap<>(),
+ new HashMap<>());
- // Note that we left some margin for the max estimation.
- Assert.assertEquals(context.getEstimatedMaxPartitionCount(), 3);
+ Assert.assertEquals(context.getEstimatedMaxPartitionCount(), 4);
Assert.assertEquals(context.getEstimatedMaxTopStateCount(), 2);
Assert.assertEquals(context.getAssignmentForFaultZoneMap(), Collections.emptyMap());
for (String resourceName : _resourceNames) {
@@ -81,9 +82,10 @@ public class TestClusterContext extends AbstractTestClusterModel {
public void testDuplicateAssign() throws IOException {
ResourceControllerDataProvider testCache = setupClusterDataCache();
Set<AssignableReplica> assignmentSet = generateReplicas(testCache);
- ClusterContext context = new ClusterContext(assignmentSet, 2, new HashMap<>(), new HashMap<>());
- context
- .addPartitionToFaultZone(_testFaultZoneId, _resourceNames.get(0), _partitionNames.get(0));
+ ClusterContext context =
+ new ClusterContext(assignmentSet, generateNodes(testCache), new HashMap<>(),
+ new HashMap<>());
+ context.addPartitionToFaultZone(_testFaultZoneId, _resourceNames.get(0), _partitionNames.get(0));
// Insert again and trigger the error.
context
.addPartitionToFaultZone(_testFaultZoneId, _resourceNames.get(0), _partitionNames.get(0));
diff --git a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/model/TestClusterModel.java b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/model/TestClusterModel.java
index 12146b7..60967ca 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/model/TestClusterModel.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/model/TestClusterModel.java
@@ -21,7 +21,6 @@ package org.apache.helix.controller.rebalancer.waged.model;
import java.io.IOException;
import java.util.Collections;
-import java.util.HashSet;
import java.util.Set;
import org.apache.helix.HelixException;
@@ -36,17 +35,6 @@ public class TestClusterModel extends AbstractTestClusterModel {
super.initialize();
}
- /**
- * Generate AssignableNodes according to the instances included in the cluster data cache.
- */
- Set<AssignableNode> generateNodes(ResourceControllerDataProvider testCache) {
- Set<AssignableNode> nodeSet = new HashSet<>();
- testCache.getInstanceConfigMap().values().stream().forEach(config -> nodeSet.add(
- new AssignableNode(testCache.getClusterConfig(),
- testCache.getInstanceConfigMap().get(_testInstanceId), config.getInstanceName())));
- return nodeSet;
- }
-
@Test
public void testNormalUsage() throws IOException {
// Test 1 - initialize the cluster model based on the data cache.
@@ -54,9 +42,10 @@ public class TestClusterModel extends AbstractTestClusterModel {
Set<AssignableReplica> assignableReplicas = generateReplicas(testCache);
Set<AssignableNode> assignableNodes = generateNodes(testCache);
- ClusterContext context = new ClusterContext(assignableReplicas, 2, Collections.emptyMap(), Collections.emptyMap());
- ClusterModel clusterModel =
- new ClusterModel(context, assignableReplicas, assignableNodes);
+ ClusterContext context =
+ new ClusterContext(assignableReplicas, assignableNodes, Collections.emptyMap(),
+ Collections.emptyMap());
+ ClusterModel clusterModel = new ClusterModel(context, assignableReplicas, assignableNodes);
Assert.assertTrue(clusterModel.getContext().getAssignmentForFaultZoneMap().values().stream()
.allMatch(resourceMap -> resourceMap.values().isEmpty()));
diff --git a/helix-core/src/test/java/org/apache/helix/integration/rebalancer/WagedRebalancer/TestNodeSwap.java b/helix-core/src/test/java/org/apache/helix/integration/rebalancer/WagedRebalancer/TestNodeSwap.java
index cbc9fea..a87be34 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/rebalancer/WagedRebalancer/TestNodeSwap.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/rebalancer/WagedRebalancer/TestNodeSwap.java
@@ -73,6 +73,13 @@ public class TestNodeSwap extends ZkTestBase {
clusterConfig.setDelayRebalaceEnabled(true);
// Set a long enough time to ensure delayed rebalance is activate
clusterConfig.setRebalanceDelayTime(3000000);
+
+ // TODO remove this setup once issue https://github.com/apache/helix/issues/532 is fixed
+ Map<ClusterConfig.GlobalRebalancePreferenceKey, Integer> preference = new HashMap<>();
+ preference.put(ClusterConfig.GlobalRebalancePreferenceKey.EVENNESS, 0);
+ preference.put(ClusterConfig.GlobalRebalancePreferenceKey.LESS_MOVEMENT, 10);
+ clusterConfig.setGlobalRebalancePreference(preference);
+
configAccessor.setClusterConfig(CLUSTER_NAME, clusterConfig);
Set<String> nodes = new HashSet<>();