You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@helix.apache.org by jx...@apache.org on 2021/05/16 21:34:37 UTC
[helix] branch replica_level_throttle updated (d356b00 -> 78d274d)
This is an automated email from the ASF dual-hosted git repository.
jxue pushed a change to branch replica_level_throttle
in repository https://gitbox.apache.org/repos/asf/helix.git.
discard d356b00 Applying per replica logic for entire stage (#1724)
omit 53c4e6b Apply per replica level throttling logic to Recovery/Load Rebalance and PendingMessage #1719
omit ad02d8d Change throttling logic to per message (#1714)
omit 22f32b9 [Replica Level Throttle] Add per replica rebalance type compute logic (#1703)
add 524ce38 Improve WAGED simulation API by throwing exception on FAILED_TO_CALCULATE (#1701)
add b3ff081 Update restlet repository url and adjust the repository order to fix build. (#1726)
add e2ec350 Correct getAssignmentForWagedFullAutoImpl snapshot behavior (#1730)
add 1758f2d Fix unstable test TestZkCallbackHandlerLeak. (#1734)
add 74236c5 Use correct session ID in PerInstanceAccessor (#1733)
add 167320c Change Workflow monitor and PerInstanceResource monitor from static m… (#1732)
new d7f6cbd [Replica Level Throttle] Add per replica rebalance type compute logic (#1703)
new a65c37e Change throttling logic to per message (#1714)
new 78d274d Apply per replica level throttling logic to Recovery/Load Rebalance and PendingMessage #1719
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (d356b00)
\
N -- N -- N refs/heads/replica_level_throttle (78d274d)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../helix/controller/common/PartitionStateMap.java | 9 +-
.../rebalancer/waged/ReadOnlyWagedRebalancer.java | 16 ++
.../rebalancer/waged/WagedRebalancer.java | 19 +-
.../stages/IntermediateStateCalcStage.java | 244 ++++++++++-----------
.../helix/controller/stages/MessageOutput.java | 4 -
.../monitoring/mbeans/ClusterStatusMonitor.java | 46 ++--
.../helix/monitoring/mbeans/JobMonitorMBean.java | 76 -------
.../mbeans/PerInstanceResourceMonitor.java | 70 ++++--
.../mbeans/PerInstanceResourceMonitorMBean.java | 34 ---
.../mbeans/StateTransitionStatMonitorMBean.java | 56 -----
.../helix/monitoring/mbeans/WorkflowMonitor.java | 176 ++++++++-------
.../monitoring/mbeans/WorkflowMonitorMBean.java | 75 -------
.../main/java/org/apache/helix/util/HelixUtil.java | 11 +-
.../integration/TestZkCallbackHandlerLeak.java | 195 +++++-----------
.../WagedRebalancer/TestWagedRebalance.java | 24 ++
.../integration/task/TestWorkflowTermination.java | 3 +-
.../helix/monitoring/TestWorkflowMonitor.java | 2 +-
.../resources/helix/PerInstanceAccessor.java | 22 +-
.../helix/rest/server/TestPerInstanceAccessor.java | 19 ++
pom.xml | 8 +-
20 files changed, 441 insertions(+), 668 deletions(-)
delete mode 100644 helix-core/src/main/java/org/apache/helix/monitoring/mbeans/JobMonitorMBean.java
delete mode 100644 helix-core/src/main/java/org/apache/helix/monitoring/mbeans/PerInstanceResourceMonitorMBean.java
delete mode 100644 helix-core/src/main/java/org/apache/helix/monitoring/mbeans/StateTransitionStatMonitorMBean.java
delete mode 100644 helix-core/src/main/java/org/apache/helix/monitoring/mbeans/WorkflowMonitorMBean.java
[helix] 03/03: Apply per replica level throttling logic to
Recovery/Load Rebalance and PendingMessage #1719
Posted by jx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jxue pushed a commit to branch replica_level_throttle
in repository https://gitbox.apache.org/repos/asf/helix.git
commit 78d274d672e04c94a6c163917fe547e8cb371873
Author: Junkai Xue <jx...@linkedin.com>
AuthorDate: Thu May 6 13:22:48 2021 -0700
Apply per replica level throttling logic to Recovery/Load Rebalance and PendingMessage #1719
This commit contains change of:
Recovery/Load Rebalance and pending messages charge adapting per replica throttling.
Change partition level comparator based on 1) missed top states, 2) missed active replicas and 3) ideal state matching or not.
---
.../stages/IntermediateStateCalcStage.java | 299 +++++++--------------
1 file changed, 101 insertions(+), 198 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/IntermediateStateCalcStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/IntermediateStateCalcStage.java
index c46a6ce..6a5d2f1 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/IntermediateStateCalcStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/IntermediateStateCalcStage.java
@@ -365,9 +365,7 @@ public class IntermediateStateCalcStage extends AbstractBaseStage {
resourceName, partitionsWithErrorStateReplica));
}
- chargePendingTransition(resource, currentStateOutput, throttleController,
- partitionsNeedRecovery, partitionsNeedLoadBalance, cache,
- bestPossiblePartitionStateMap, intermediatePartitionStateMap);
+ chargePendingTransition(resource, currentStateOutput, throttleController, cache, preferenceLists, stateModelDef, intermediatePartitionStateMap);
// Perform recovery balance
Set<Partition> recoveryThrottledPartitions =
@@ -423,211 +421,114 @@ public class IntermediateStateCalcStage extends AbstractBaseStage {
}
/**
- * Check for a partition, whether all transitions for its replicas are downward transitions. Note
- * that this function does NOT check for ERROR states.
- * @param currentStateMap
- * @param bestPossibleMap
- * @param stateModelDef
- * @return true if there are; false otherwise
+ * Determine the message is downward message or not.
+ * @param message message for load rebalance
+ * @param stateModelDefinition state model definition object for this resource
+ * @return set of messages allowed for downward state transitions
*/
- private boolean isLoadBalanceDownwardForAllReplicas(Map<String, String> currentStateMap,
- Map<String, String> bestPossibleMap, StateModelDefinition stateModelDef) {
- Set<String> allInstances = new HashSet<>();
- allInstances.addAll(currentStateMap.keySet());
- allInstances.addAll(bestPossibleMap.keySet());
- Map<String, Integer> statePriorityMap = stateModelDef.getStatePriorityMap();
-
- for (String instance : allInstances) {
- String currentState = currentStateMap.get(instance);
- String bestPossibleState = bestPossibleMap.get(instance);
- if (currentState == null) {
- return false; // null -> state is upward
- }
- if (bestPossibleState != null) {
- // Compare priority values and return if an upward transition is found
- // Note that lower integer value implies higher priority
- if (!statePriorityMap.containsKey(currentState)
- || !statePriorityMap.containsKey(bestPossibleState)) {
- // If the state is not found in statePriorityMap, consider it not strictly downward by
- // default because we can't determine whether it is downward
- return false;
- }
- if (statePriorityMap.get(currentState) > statePriorityMap.get(bestPossibleState)) {
- return false;
- }
- }
- }
- return true;
+ private boolean isLoadBalanceDownwardStateTransition(Message message, StateModelDefinition stateModelDefinition) {
+ // state model definition is not found
+ if (stateModelDefinition == null) {
+ return false;
+ }
+
+ Map<String, Integer> statePriorityMap = stateModelDefinition.getStatePriorityMap();
+ // Compare priority values and return if an upward transition is found
+ // Note that lower integer value implies higher priority
+ // If the state is not found in statePriorityMap, consider it not strictly downward by
+ // default because we can't determine whether it is downward
+ if (statePriorityMap.containsKey(message.getFromState()) && statePriorityMap.containsKey(message.getToState())
+ && statePriorityMap.get(message.getFromState()) > statePriorityMap.get(message.getToState())) {
+ return true;
+ }
+ return false;
}
/**
* Check and charge all pending transitions for throttling.
*/
private void chargePendingTransition(Resource resource, CurrentStateOutput currentStateOutput,
- StateTransitionThrottleController throttleController, Set<Partition> partitionsNeedRecovery,
- Set<Partition> partitionsNeedLoadbalance, ResourceControllerDataProvider cache,
- PartitionStateMap bestPossiblePartitionStateMap,
+ StateTransitionThrottleController throttleController, ResourceControllerDataProvider cache,
+ Map<String, List<String>> preferenceLists, StateModelDefinition stateModelDefinition,
PartitionStateMap intermediatePartitionStateMap) {
String resourceName = resource.getResourceName();
-
// check and charge pending transitions
for (Partition partition : resource.getPartitions()) {
+ // To clarify that custom mode does not apply recovery/load rebalance since user can define different number of
+ // replicas for different partitions. Actually, the custom will stopped from resource level checks if this resource
+ // is not FULL_AUTO, we will return best possible state and do nothing.
+ Map<String, Integer> requiredStates =
+ getRequiredStates(resourceName, cache, preferenceLists.get(partition.getPartitionName()));
// Maps instance to its current state
- Map<String, String> currentStateMap =
- currentStateOutput.getCurrentStateMap(resourceName, partition);
+ Map<String, String> currentStateMap = currentStateOutput.getCurrentStateMap(resourceName, partition);
// Maps instance to its pending (next) state
- Map<String, String> pendingMap =
- currentStateOutput.getPendingStateMap(resourceName, partition);
-
- StateTransitionThrottleConfig.RebalanceType rebalanceType = RebalanceType.NONE;
- if (partitionsNeedRecovery.contains(partition)) {
- rebalanceType = StateTransitionThrottleConfig.RebalanceType.RECOVERY_BALANCE;
- } else if (partitionsNeedLoadbalance.contains(partition)) {
- rebalanceType = StateTransitionThrottleConfig.RebalanceType.LOAD_BALANCE;
- }
-
- if (pendingMap.size() > 0) {
- boolean shouldChargePartition = false;
- for (String instance : pendingMap.keySet()) {
- String currentState = currentStateMap.get(instance);
- String pendingState = pendingMap.get(instance);
- if (pendingState != null && !pendingState.equals(currentState)
- && !cache.getDisabledInstancesForPartition(resourceName, partition.getPartitionName())
- .contains(instance)) {
- // Only charge this instance if the partition is not disabled
- throttleController.chargeInstance(rebalanceType, instance);
- shouldChargePartition = true;
- // If there is a pending state transition for the partition, that means that an assignment
- // has already been made and the state transition message has already been sent out for the partition
- // in a previous pipeline run. We must honor this and reflect it by charging for the pending state transition message.
-
- // Since the assignment has already been made for the pending message, we do a special treatment
- // for it by setting the best possible state directly in intermediatePartitionStateMap so that the pending
- // message won't be double-assigned or double-charged in recovery or load balance.
- handlePendingStateTransitionsForThrottling(partition, partitionsNeedRecovery,
- partitionsNeedLoadbalance, rebalanceType, bestPossiblePartitionStateMap,
- intermediatePartitionStateMap);
- }
- }
- if (shouldChargePartition) {
- throttleController.chargeCluster(rebalanceType);
+ List<Message> pendingMessages =
+ new ArrayList<>(currentStateOutput.getPendingMessageMap(resourceName, partition).values());
+ Collections.sort(pendingMessages, new MessagePriorityComparator(preferenceLists.get(partition.getPartitionName()),
+ stateModelDefinition.getStatePriorityMap()));
+
+ for (Message message : pendingMessages) {
+ StateTransitionThrottleConfig.RebalanceType rebalanceType =
+ getRebalanceTypePerMessage(requiredStates, message, currentStateMap);
+ String currentState = currentStateMap.get(message.getTgtName());
+ if (!message.getToState().equals(currentState) && message.getFromState().equals(currentState)
+ && !cache.getDisabledInstancesForPartition(resourceName, partition.getPartitionName())
+ .contains(message.getTgtName())) {
+ throttleController.chargeInstance(rebalanceType, message.getTgtName());
throttleController.chargeResource(rebalanceType, resourceName);
+ throttleController.chargeCluster(rebalanceType);
}
+ intermediatePartitionStateMap.setState(partition, message.getTgtName(), message.getToState());
}
}
}
/**
- * Sort partitions according to partition priority {@link PartitionPriorityComparator}, and for
- * each partition, throttle state transitions if needed. Also populate
- * intermediatePartitionStateMap either with BestPossibleState (if no throttling is necessary) or
- * CurrentState (if throttled).
- * @param resource
- * @param bestPossiblePartitionStateMap
- * @param throttleController
- * @param intermediatePartitionStateMap
- * @param partitionsNeedRecovery
- * @param currentStateOutput
- * @param topState
- * @param cache
- * @return a set of partitions that need recovery but did not get recovered due to throttling
+ * Thin wrapper for per message throttling with recovery rebalance type. Also populate
+ * intermediatePartitionStateMap with generated messages from {@link MessageGenerationPhase}.
+ * @param resource the resource to throttle
+ * @param throttleController throttle controller object
+ * @param messageToThrottle the message to be throttled
+ * @param intermediatePartitionStateMap output result for this stage that intermediate state map
+ * @param cache cache object for computational metadata from external storage
+ * @param messagesThrottled messages that have already been throttled
+ * @param resourceMessageMap the map for all messages from MessageSelectStage. Remove the message
+ * if it has been throttled
*/
- private Set<Partition> recoveryRebalance(Resource resource,
- PartitionStateMap bestPossiblePartitionStateMap,
- StateTransitionThrottleController throttleController,
- PartitionStateMap intermediatePartitionStateMap, Set<Partition> partitionsNeedRecovery,
- CurrentStateOutput currentStateOutput, String topState,
- ResourceControllerDataProvider cache) {
- String resourceName = resource.getResourceName();
- Set<Partition> partitionRecoveryBalanceThrottled = new HashSet<>();
-
- // Maps Partition -> Instance -> State
- Map<Partition, Map<String, String>> currentStateMap =
- currentStateOutput.getCurrentStateMap(resourceName);
- List<Partition> partitionsNeedRecoveryPrioritized = new ArrayList<>(partitionsNeedRecovery);
-
- // We want the result of the intermediate state calculation to be deterministic. We sort here by
- // partition name to ensure that the order is consistent for inputs fed into
- // PartitionPriorityComparator sort
- partitionsNeedRecoveryPrioritized.sort(Comparator.comparing(Partition::getPartitionName));
- partitionsNeedRecoveryPrioritized.sort(new PartitionPriorityComparator(
- bestPossiblePartitionStateMap.getStateMap(), currentStateMap, topState, true));
-
- // For each partition, apply throttling if needed.
- for (Partition partition : partitionsNeedRecoveryPrioritized) {
- throttleStateTransitionsForPartition(throttleController, resourceName, partition,
- currentStateOutput, bestPossiblePartitionStateMap, partitionRecoveryBalanceThrottled,
- intermediatePartitionStateMap, RebalanceType.RECOVERY_BALANCE, cache);
- }
- LogUtil.logInfo(logger, _eventId, String.format(
- "For resource %s: Num of partitions needing recovery: %d, Num of partitions needing recovery"
- + " but throttled (not recovered): %d",
- resourceName, partitionsNeedRecovery.size(), partitionRecoveryBalanceThrottled.size()));
- return partitionRecoveryBalanceThrottled;
+ private void recoveryRebalance(Resource resource, Partition partition,
+ StateTransitionThrottleController throttleController, Message messageToThrottle,
+ PartitionStateMap intermediatePartitionStateMap, ResourceControllerDataProvider cache,
+ Set<Message> messagesThrottled, Map<Partition, List<Message>> resourceMessageMap) {
+ throttleStateTransitionsForReplica(throttleController, resource.getResourceName(), partition, messageToThrottle,
+ messagesThrottled, intermediatePartitionStateMap, RebalanceType.RECOVERY_BALANCE, cache, resourceMessageMap);
}
/**
- * Sort partitions according to partition priority {@link PartitionPriorityComparator}, and for
- * each partition, throttle state transitions if needed. Also populate
- * intermediatePartitionStateMap either with BestPossibleState (if no throttling is necessary) or
- * CurrentState (if throttled).
- * @param resource
- * @param currentStateOutput
- * @param bestPossiblePartitionStateMap
- * @param throttleController
- * @param intermediatePartitionStateMap
- * @param partitionsNeedLoadbalance
- * @param currentStateMap
- * @param onlyDownwardLoadBalance true when only allowing downward transitions
- * @param stateModelDef for determining whether a partition's transitions are strictly downward
- * @param cache
- * @return
+ * Thin wrapper for per message throttling with load rebalance type. Also populate
+ * intermediatePartitionStateMap with generated messages from {@link MessageGenerationPhase}.
+ * @param resource the resource to throttle
+ * @param throttleController throttle controller object
+ * @param messageToThrottle the message to be throttle
+ * @param intermediatePartitionStateMap output result for this stage that intermediate state map
+ * @param cache cache object for computational metadata from external storage
+ * @param onlyDownwardLoadBalance does allow only downward load balance
+ * @param stateModelDefinition state model definition of this resource
+ * @param messagesThrottled messages are already throttled
+ * @param resourceMessageMap the map for all messages from MessageSelectStage. Remove the message
+ * if it has been throttled
*/
- private Set<Partition> loadRebalance(Resource resource, CurrentStateOutput currentStateOutput,
- PartitionStateMap bestPossiblePartitionStateMap,
- StateTransitionThrottleController throttleController,
- PartitionStateMap intermediatePartitionStateMap, Set<Partition> partitionsNeedLoadbalance,
- Map<Partition, Map<String, String>> currentStateMap, boolean onlyDownwardLoadBalance,
- StateModelDefinition stateModelDef, ResourceControllerDataProvider cache) {
- String resourceName = resource.getResourceName();
- Set<Partition> partitionsLoadbalanceThrottled = new HashSet<>();
-
- List<Partition> partitionsNeedLoadRebalancePrioritized =
- new ArrayList<>(partitionsNeedLoadbalance);
-
- // We want the result of the intermediate state calculation to be deterministic. We sort here by
- // partition name to ensure that the order is consistent for inputs fed into
- // PartitionPriorityComparator sort
- partitionsNeedLoadRebalancePrioritized.sort(Comparator.comparing(Partition::getPartitionName));
- partitionsNeedLoadRebalancePrioritized.sort(new PartitionPriorityComparator(
- bestPossiblePartitionStateMap.getStateMap(), currentStateMap, "", false));
-
- for (Partition partition : partitionsNeedLoadRebalancePrioritized) {
- // If this is a downward load balance, check if the partition's transition is strictly
- // downward
- if (onlyDownwardLoadBalance) {
- Map<String, String> currentStateMapForPartition =
- currentStateOutput.getCurrentStateMap(resourceName, partition);
- Map<String, String> bestPossibleMapForPartition =
- bestPossiblePartitionStateMap.getPartitionMap(partition);
- if (!isLoadBalanceDownwardForAllReplicas(currentStateMapForPartition,
- bestPossibleMapForPartition, stateModelDef)) {
- // For downward load balance, if a partition's transitions are not strictly downward,
- // set currentState to intermediateState
- intermediatePartitionStateMap.setState(partition, currentStateMapForPartition);
- continue;
- }
- }
- throttleStateTransitionsForPartition(throttleController, resourceName, partition,
- currentStateOutput, bestPossiblePartitionStateMap, partitionsLoadbalanceThrottled,
- intermediatePartitionStateMap, RebalanceType.LOAD_BALANCE, cache);
- }
- LogUtil.logInfo(logger, _eventId,
- String.format(
- "For resource %s: Num of partitions needing load-balance: %d, Num of partitions needing"
- + " load-balance but throttled (not load-balanced): %d",
- resourceName, partitionsNeedLoadbalance.size(), partitionsLoadbalanceThrottled.size()));
- return partitionsLoadbalanceThrottled;
+ private void loadRebalance(Resource resource, Partition partition,
+ StateTransitionThrottleController throttleController, Message messageToThrottle,
+ PartitionStateMap intermediatePartitionStateMap, ResourceControllerDataProvider cache,
+ boolean onlyDownwardLoadBalance, StateModelDefinition stateModelDefinition, Set<Message> messagesThrottled,
+ Map<Partition, List<Message>> resourceMessageMap) {
+ if (onlyDownwardLoadBalance && isLoadBalanceDownwardStateTransition(messageToThrottle, stateModelDefinition)) {
+ // Remove the message already allowed for downward state transitions.
+ intermediatePartitionStateMap.setState(partition, messageToThrottle.getTgtName(), messageToThrottle.getToState());
+ return;
+ }
+ throttleStateTransitionsForReplica(throttleController, resource.getResourceName(), partition, messageToThrottle,
+ messagesThrottled, intermediatePartitionStateMap, RebalanceType.LOAD_BALANCE, cache, resourceMessageMap);
}
/**
@@ -643,11 +544,13 @@ public class IntermediateStateCalcStage extends AbstractBaseStage {
* of IntermediateStateCalcStage
* @param rebalanceType the rebalance type to charge quota
* @param cache cached cluster metadata required by the throttle controller
+ * @param resourceMessageMap the map for all messages from MessageSelectStage. Remove the message
+ * if it has been throttled.
*/
private void throttleStateTransitionsForReplica(StateTransitionThrottleController throttleController,
String resourceName, Partition partition, Message messageToThrottle, Set<Message> messagesThrottled,
PartitionStateMap intermediatePartitionStateMap, RebalanceType rebalanceType,
- ResourceControllerDataProvider cache) {
+ ResourceControllerDataProvider cache, Map<Partition, List<Message>> resourceMessageMap) {
boolean hasReachedThrottlingLimit = false;
if (throttleController.shouldThrottleForResource(rebalanceType, resourceName)) {
hasReachedThrottlingLimit = true;
@@ -678,6 +581,9 @@ public class IntermediateStateCalcStage extends AbstractBaseStage {
intermediatePartitionStateMap.setState(partition, messageToThrottle.getTgtName(), messageToThrottle.getToState());
} else {
// Intermediate Map is based on current state
+ // Remove the message from MessageSelection result if it has been throttled since the message will be dispatched
+ // by next stage if it is not removed.
+ resourceMessageMap.get(partition).remove(messageToThrottle);
messagesThrottled.add(messageToThrottle);
}
}
@@ -912,29 +818,26 @@ public class IntermediateStateCalcStage extends AbstractBaseStage {
private Map<Partition, Map<String, String>> _bestPossibleMap;
private Map<Partition, Map<String, String>> _currentStateMap;
private String _topState;
- private boolean _recoveryRebalance;
PartitionPriorityComparator(Map<Partition, Map<String, String>> bestPossibleMap,
- Map<Partition, Map<String, String>> currentStateMap, String topState,
- boolean recoveryRebalance) {
+ Map<Partition, Map<String, String>> currentStateMap, String topState){
_bestPossibleMap = bestPossibleMap;
_currentStateMap = currentStateMap;
_topState = topState;
- _recoveryRebalance = recoveryRebalance;
}
@Override
public int compare(Partition p1, Partition p2) {
- if (_recoveryRebalance) {
- int missTopState1 = getMissTopStateIndex(p1);
- int missTopState2 = getMissTopStateIndex(p2);
- // Highest priority for the partition without top state
- if (missTopState1 != missTopState2) {
- return Integer.compare(missTopState1, missTopState2);
- }
- // Higher priority for the partition with fewer active replicas
- int currentActiveReplicas1 = getCurrentActiveReplicas(p1);
- int currentActiveReplicas2 = getCurrentActiveReplicas(p2);
+ int missTopState1 = getMissTopStateIndex(p1);
+ int missTopState2 = getMissTopStateIndex(p2);
+ // Highest priority for the partition without top state
+ if (missTopState1 != missTopState2) {
+ return Integer.compare(missTopState1, missTopState2);
+ }
+ // Higher priority for the partition with fewer active replicas
+ int currentActiveReplicas1 = getCurrentActiveReplicas(p1);
+ int currentActiveReplicas2 = getCurrentActiveReplicas(p2);
+ if (currentActiveReplicas1 != currentActiveReplicas2) {
return Integer.compare(currentActiveReplicas1, currentActiveReplicas2);
}
// Higher priority for the partition with fewer replicas with states matching with IdealState
[helix] 02/03: Change throttling logic to per message (#1714)
Posted by jx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jxue pushed a commit to branch replica_level_throttle
in repository https://gitbox.apache.org/repos/asf/helix.git
commit a65c37eca0e4fb0c5290a136509d945e4818fc7e
Author: Junkai Xue <jx...@linkedin.com>
AuthorDate: Wed Apr 28 13:20:09 2021 -0700
Change throttling logic to per message (#1714)
Apply the logic for throttling with per message quota charge.
---
.../stages/IntermediateStateCalcStage.java | 115 ++++++---------------
1 file changed, 33 insertions(+), 82 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/IntermediateStateCalcStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/IntermediateStateCalcStage.java
index a840d5e..c46a6ce 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/IntermediateStateCalcStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/IntermediateStateCalcStage.java
@@ -631,104 +631,55 @@ public class IntermediateStateCalcStage extends AbstractBaseStage {
}
/**
- * Check the status on throttling at every level (cluster, resource, instance) and set
- * intermediatePartitionStateMap accordingly per partition.
- * @param throttleController
- * @param resourceName
- * @param partition
- * @param currentStateOutput
- * @param bestPossiblePartitionStateMap
- * @param partitionsThrottled
- * @param intermediatePartitionStateMap
- * @param rebalanceType
- * @param cache
+ * Check the status for a single message on throttling at every level (cluster, resource, replica) and set
+ * intermediatePartitionStateMap accordingly for that replica.
+ * @param throttleController throttle controller object for throttling quota
+ * @param resourceName the resource for throttling check
+ * @param partition the partition for throttling check
+ * @param messageToThrottle the message to be throttled
+ * @param messagesThrottled the cumulative set of messages that have been throttled already. These
+ * messages represent the replicas of this partition that have been throttled.
+ * @param intermediatePartitionStateMap the cumulative partition-state mapping as a result of the throttling step
+ * of IntermediateStateCalcStage
+ * @param rebalanceType the rebalance type to charge quota
+ * @param cache cached cluster metadata required by the throttle controller
*/
- private void throttleStateTransitionsForPartition(
- StateTransitionThrottleController throttleController, String resourceName,
- Partition partition, CurrentStateOutput currentStateOutput,
- PartitionStateMap bestPossiblePartitionStateMap, Set<Partition> partitionsThrottled,
+ private void throttleStateTransitionsForReplica(StateTransitionThrottleController throttleController,
+ String resourceName, Partition partition, Message messageToThrottle, Set<Message> messagesThrottled,
PartitionStateMap intermediatePartitionStateMap, RebalanceType rebalanceType,
ResourceControllerDataProvider cache) {
-
- Map<String, String> currentStateMap =
- currentStateOutput.getCurrentStateMap(resourceName, partition);
- Map<String, String> bestPossibleMap = bestPossiblePartitionStateMap.getPartitionMap(partition);
- Set<String> allInstances = new HashSet<>(currentStateMap.keySet());
- allInstances.addAll(bestPossibleMap.keySet());
- Map<String, String> intermediateMap = new HashMap<>();
-
boolean hasReachedThrottlingLimit = false;
if (throttleController.shouldThrottleForResource(rebalanceType, resourceName)) {
hasReachedThrottlingLimit = true;
if (logger.isDebugEnabled()) {
- LogUtil.logDebug(logger, _eventId,
- String.format("Throttled on partition: %s in resource: %s",
- partition.getPartitionName(), resourceName));
+ LogUtil.logDebug(logger, _eventId, String.format(
+ "Throttled because of cluster/resource quota is full for message {%s} on partition {%s} in resource {%s}",
+ messageToThrottle.getId(), partition.getPartitionName(), resourceName));
}
} else {
- // throttle if any of the instances are not able to accept state transitions
- for (String instance : allInstances) {
- String currentState = currentStateMap.get(instance);
- String bestPossibleState = bestPossibleMap.get(instance);
- if (bestPossibleState != null && !bestPossibleState.equals(currentState)
- && !cache.getDisabledInstancesForPartition(resourceName, partition.getPartitionName())
- .contains(instance)) {
- if (throttleController.shouldThrottleForInstance(rebalanceType, instance)) {
- hasReachedThrottlingLimit = true;
- if (logger.isDebugEnabled()) {
- LogUtil.logDebug(logger, _eventId,
- String.format(
- "Throttled because of instance: %s for partition: %s in resource: %s",
- instance, partition.getPartitionName(), resourceName));
- }
- break;
+ // Since message already generated, we can assume the current state is not null and target state is not null
+ if (!cache.getDisabledInstancesForPartition(resourceName, partition.getPartitionName())
+ .contains(messageToThrottle.getTgtName())) {
+ if (throttleController.shouldThrottleForInstance(rebalanceType, messageToThrottle.getTgtName())) {
+ hasReachedThrottlingLimit = true;
+ if (logger.isDebugEnabled()) {
+ LogUtil.logDebug(logger, _eventId, String.format(
+ "Throttled because of instance level quota is full on instance {%s} for message {%s} of partition {%s} in resource {%s}",
+ messageToThrottle.getId(), messageToThrottle.getTgtName(), partition.getPartitionName(), resourceName));
}
}
}
}
+ // If there is still room for this replica, proceed to charge at the cluster and resource level and set the
+ // intermediate partition-state mapping so that the state transition message can move forward.
if (!hasReachedThrottlingLimit) {
- // This implies that there is room for more state transitions.
- // Find instances with a replica whose current state is different from BestPossibleState and
- // "charge" for it, and bestPossibleStates will become intermediate states
- intermediateMap.putAll(bestPossibleMap);
- boolean shouldChargeForPartition = false;
- for (String instance : allInstances) {
- String currentState = currentStateMap.get(instance);
- String bestPossibleState = bestPossibleMap.get(instance);
- if (bestPossibleState != null && !bestPossibleState.equals(currentState)
- && !cache.getDisabledInstancesForPartition(resourceName, partition.getPartitionName())
- .contains(instance)) {
- throttleController.chargeInstance(rebalanceType, instance);
- shouldChargeForPartition = true;
- }
- }
- if (shouldChargeForPartition) {
- throttleController.chargeCluster(rebalanceType);
- throttleController.chargeResource(rebalanceType, resourceName);
- }
+ throttleController.chargeCluster(rebalanceType);
+ throttleController.chargeResource(rebalanceType, resourceName);
+ intermediatePartitionStateMap.setState(partition, messageToThrottle.getTgtName(), messageToThrottle.getToState());
} else {
- // No more room for more state transitions; current states will just become intermediate
- // states unless the partition is disabled
- // Add this partition to a set of throttled partitions
- for (String instance : allInstances) {
- String currentState = currentStateMap.get(instance);
- String bestPossibleState = bestPossibleMap.get(instance);
- if (bestPossibleState != null && !bestPossibleState.equals(currentState)
- && cache.getDisabledInstancesForPartition(resourceName, partition.getPartitionName())
- .contains(instance)) {
- // Because this partition is disabled, we allow assignment
- intermediateMap.put(instance, bestPossibleState);
- } else {
- // This partition is not disabled, so it must be throttled by just passing on the current
- // state
- if (currentState != null) {
- intermediateMap.put(instance, currentState);
- }
- partitionsThrottled.add(partition);
- }
- }
+ // Intermediate Map is based on current state
+ messagesThrottled.add(messageToThrottle);
}
- intermediatePartitionStateMap.setState(partition, intermediateMap);
}
/**
[helix] 01/03: [Replica Level Throttle] Add per replica rebalance
type compute logic (#1703)
Posted by jx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jxue pushed a commit to branch replica_level_throttle
in repository https://gitbox.apache.org/repos/asf/helix.git
commit d7f6cbd194d15351ba8d0a324827e726ead0ffa8
Author: Junkai Xue <jx...@linkedin.com>
AuthorDate: Fri Apr 23 14:58:48 2021 -0700
[Replica Level Throttle] Add per replica rebalance type compute logic (#1703)
* Add per replica rebalance type compute logic
Three functions added: 1) rebalance type computation required state.
2) rebalance type per message
3) message sorting rules and comparators to determine which message to apply first.
---
.../stages/IntermediateStateCalcStage.java | 82 +++++++++++++++++++++-
1 file changed, 81 insertions(+), 1 deletion(-)
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/IntermediateStateCalcStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/IntermediateStateCalcStage.java
index c0defa9..a840d5e 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/IntermediateStateCalcStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/IntermediateStateCalcStage.java
@@ -29,6 +29,8 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
import org.apache.helix.HelixDefinedState;
import org.apache.helix.HelixException;
import org.apache.helix.HelixManager;
@@ -43,6 +45,7 @@ import org.apache.helix.model.BuiltInStateModelDefinitions;
import org.apache.helix.model.ClusterConfig;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.MaintenanceSignal;
+import org.apache.helix.model.Message;
import org.apache.helix.model.Partition;
import org.apache.helix.model.Resource;
import org.apache.helix.model.StateModelDefinition;
@@ -796,6 +799,54 @@ public class IntermediateStateCalcStage extends AbstractBaseStage {
}
/**
+ * Determine the message rebalance type with message and current states.
+ * @param desiredStates Ideally how may states we needed for guarantee the health of replica
+ * @param message The message to be determined what is the rebalance type
+ * @param derivedCurrentStates Derived from current states with previous messages not be throttled.
+ * @return Rebalance type. Recovery or load.
+ */
+ private RebalanceType getRebalanceTypePerMessage(Map<String, Integer> desiredStates, Message message,
+ Map<String, String> derivedCurrentStates) {
+ Map<String, Integer> desiredStatesSnapshot = new HashMap<>(desiredStates);
+ // Looping existing current states to see whether current states fulfilled all the required states.
+ for (String state : derivedCurrentStates.values()) {
+ if (desiredStatesSnapshot.containsKey(state)) {
+ if (desiredStatesSnapshot.get(state) == 1) {
+ desiredStatesSnapshot.remove(state);
+ } else {
+ desiredStatesSnapshot.put(state, desiredStatesSnapshot.get(state) - 1);
+ }
+ }
+ }
+
+ // If the message contains any "required" state changes, then it is considered recovery rebalance.
+ // Otherwise, it is load balance.
+ return desiredStatesSnapshot.containsKey(message.getToState()) ? RebalanceType.RECOVERY_BALANCE
+ : RebalanceType.LOAD_BALANCE;
+ }
+
+ private Map<String, Integer> getRequiredStates(String resourceName,
+ ResourceControllerDataProvider resourceControllerDataProvider, List<String> preferenceList) {
+
+ // Prepare required inputs: 1) Priority State List 2) required number of replica
+ IdealState idealState = resourceControllerDataProvider.getIdealState(resourceName);
+ StateModelDefinition stateModelDefinition =
+ resourceControllerDataProvider.getStateModelDef(idealState.getStateModelDefRef());
+ int requiredNumReplica = idealState.getMinActiveReplicas() == -1
+ ? idealState.getReplicaCount(preferenceList.size())
+ : idealState.getMinActiveReplicas();
+
+ // Generate a state mapping, state -> required numbers based on the live and enabled instances for this partition
+ // preference list
+ LinkedHashMap<String, Integer> expectedStateCountMap = stateModelDefinition.getStateCountMap(
+ (int) preferenceList.stream()
+ .filter(i -> resourceControllerDataProvider.getEnabledLiveInstances().contains(i))
+ .count(), requiredNumReplica); // StateModelDefinition's counts
+
+ return expectedStateCountMap;
+ }
+
+ /**
* Log rebalancer metadata for debugging purposes.
* @param resource
* @param allPartitions
@@ -874,7 +925,36 @@ public class IntermediateStateCalcStage extends AbstractBaseStage {
}
}
- // Compare partitions according following standard:
+ private class MessagePriorityComparator implements Comparator<Message> {
+ private Map<String, Integer> _preferenceInstanceMap;
+ private Map<String, Integer> _statePriorityMap;
+
+ MessagePriorityComparator(List<String> preferenceList, Map<String, Integer> statePriorityMap) {
+ // Get instance -> priority map.
+ _preferenceInstanceMap = IntStream.range(0, preferenceList.size())
+ .boxed()
+ .collect(Collectors.toMap(preferenceList::get, index -> index));
+ _statePriorityMap = statePriorityMap;
+ }
+
+ @Override
+ public int compare(Message m1, Message m2) {
+ // Compare rules:
+ // 1. Higher target state has higher priority.
+ // 2. If target state is same, range it as preference list order.
+ // 3. Sort by the name of targeted instances just for deterministic ordering.
+ if (m1.getToState().equals(m2.getToState()) && _preferenceInstanceMap.containsKey(m1.getTgtName())
+ && _preferenceInstanceMap.containsKey(m2.getTgtName())) {
+ return _preferenceInstanceMap.get(m1.getTgtName()).compareTo(_preferenceInstanceMap.get(m2.getTgtName()));
+ }
+ if (!m1.getToState().equals(m2.getToState())) {
+ return _statePriorityMap.get(m1.getToState()).compareTo(_statePriorityMap.get(m2.getToState()));
+ }
+ return m1.getTgtName().compareTo(m2.getTgtName());
+ }
+ }
+
+ // Compare partitions according following standard:
// 1) Partition without top state always is the highest priority.
// 2) For partition with top-state, the more number of active replica it has, the less priority.
private class PartitionPriorityComparator implements Comparator<Partition> {