From 1a8e0773eb3b49e40d3d21d5df2ac90485e364c1 Mon Sep 17 00:00:00 2001 From: Kei Peralta Date: Wed, 16 Sep 2026 10:22:25 -0700 Subject: [PATCH] Port TopicLeaderDistributionGoal and ExecutionTaskPlanner changes Port Liftoff changes made to 2.4.20-liftoff4, originally written by Andrey. This includes: - Replica movement scheduling improvements introduced in #4. - Addition of the TopicLeaderDistributionGoal (#5, #6, c509c47f). Some changes in the 3.0.4 version: - Replace leadersOfTopicInBroker and numLeadersOfTopicInBroker with leadersFor and the new built-in numLeadersFor, which both leverage the replicasOfTopicInBroker, which already handles the null case for _topicReplicas.get(topic). - ClusterModel.numTopicLeaders(topic) now directly calls broker.numLeadersFor(topic) for each broker in the cluster instead of going through racks. - Updated ClusterModelStats.numForAvgTopicLeaders to: - Take brokers, topics, and aliveBrokers as parameters to avoid re-computing the same in the method. - Merge the variance computation into the brokers loop, gated on broker.isAlive(), instead of iterating through aliveBrokers separately. - Uses the double primitive instead of Double for BalancingConstraint.topicLeaderBalancePercentage, matching the refactor in the other methods of the class. - Made some style changes so that checkstyle passes. - Added test changes to ExecutionTaskPlannerTest, to reflect the change we made in ExecutionTaskPlanner. --- .../analyzer/BalancingConstraint.java | 10 + .../goals/TopicLeaderDistributionGoal.java | 589 ++++++++++++++++++ .../config/constants/AnalyzerConfig.java | 14 + .../executor/ExecutionTaskPlanner.java | 4 +- .../kafka/cruisecontrol/model/Broker.java | 10 + .../cruisecontrol/model/ClusterModel.java | 13 + .../model/ClusterModelStats.java | 53 ++ .../analyzer/DeterministicClusterTest.java | 2 + .../analyzer/RandomClusterTest.java | 2 + .../executor/ExecutionTaskPlannerTest.java | 12 +- 10 files changed, 700 insertions(+), 9 deletions(-) create mode 100644 cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/analyzer/goals/TopicLeaderDistributionGoal.java diff --git a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/analyzer/BalancingConstraint.java b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/analyzer/BalancingConstraint.java index f087efe9ec..f5db254ad3 100644 --- a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/analyzer/BalancingConstraint.java +++ b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/analyzer/BalancingConstraint.java @@ -28,6 +28,7 @@ public class BalancingConstraint { private final double _topicReplicaBalancePercentage; private final int _topicReplicaBalanceMinGap; private final int _topicReplicaBalanceMaxGap; + private final double _topicLeaderBalancePercentage; private final double _goalViolationDistributionThresholdMultiplier; private final Map _capacityThreshold; private final Map _lowUtilizationThreshold; @@ -76,6 +77,7 @@ public BalancingConstraint(KafkaCruiseControlConfig config) { _replicaBalancePercentage = config.getDouble(AnalyzerConfig.REPLICA_COUNT_BALANCE_THRESHOLD_CONFIG); _leaderReplicaBalancePercentage = config.getDouble(AnalyzerConfig.LEADER_REPLICA_COUNT_BALANCE_THRESHOLD_CONFIG); _topicReplicaBalancePercentage = config.getDouble(AnalyzerConfig.TOPIC_REPLICA_COUNT_BALANCE_THRESHOLD_CONFIG); + _topicLeaderBalancePercentage = config.getDouble(AnalyzerConfig.TOPIC_LEADER_COUNT_BALANCE_THRESHOLD_CONFIG); _topicReplicaBalanceMinGap = config.getInt(AnalyzerConfig.TOPIC_REPLICA_COUNT_BALANCE_MIN_GAP_CONFIG); _topicReplicaBalanceMaxGap = config.getInt(AnalyzerConfig.TOPIC_REPLICA_COUNT_BALANCE_MAX_GAP_CONFIG); _goalViolationDistributionThresholdMultiplier = config.getDouble(AnalyzerConfig.GOAL_VIOLATION_DISTRIBUTION_THRESHOLD_MULTIPLIER_CONFIG); @@ -122,6 +124,7 @@ Properties setProps(Properties props) { props.put(AnalyzerConfig.TOPIC_REPLICA_COUNT_BALANCE_THRESHOLD_CONFIG, Double.toString(_topicReplicaBalancePercentage)); props.put(AnalyzerConfig.TOPIC_REPLICA_COUNT_BALANCE_MIN_GAP_CONFIG, Integer.toString(_topicReplicaBalanceMinGap)); props.put(AnalyzerConfig.TOPIC_REPLICA_COUNT_BALANCE_MAX_GAP_CONFIG, Integer.toString(_topicReplicaBalanceMaxGap)); + props.put(AnalyzerConfig.TOPIC_LEADER_COUNT_BALANCE_THRESHOLD_CONFIG, Double.toString(_topicLeaderBalancePercentage)); props.put(AnalyzerConfig.GOAL_VIOLATION_DISTRIBUTION_THRESHOLD_MULTIPLIER_CONFIG, Double.toString(_goalViolationDistributionThresholdMultiplier)); props.put(AnalyzerConfig.TOPICS_WITH_MIN_LEADERS_PER_BROKER_CONFIG, _topicsWithMinLeadersPerBrokerPattern.pattern()); props.put(AnalyzerConfig.MIN_TOPIC_LEADERS_PER_BROKER_CONFIG, Integer.toString(_minTopicLeadersPerBroker)); @@ -197,6 +200,13 @@ public int topicReplicaBalanceMaxGap() { return _topicReplicaBalanceMaxGap; } + /** + * @return Topic leader replica balance percentage for {@link com.linkedin.kafka.cruisecontrol.analyzer.goals.TopicLeaderDistributionGoal}. + */ + public double topicLeaderBalancePercentage() { + return _topicLeaderBalancePercentage; + } + /** * @return Goal violation distribution threshold multiplier to be used in detection and fixing goal violations. */ diff --git a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/analyzer/goals/TopicLeaderDistributionGoal.java b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/analyzer/goals/TopicLeaderDistributionGoal.java new file mode 100644 index 0000000000..263b8523dc --- /dev/null +++ b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/analyzer/goals/TopicLeaderDistributionGoal.java @@ -0,0 +1,589 @@ +/* + * Copyright 2017 LinkedIn Corp. Licensed under the BSD 2-Clause License (the "License"). See License in the project root for license information. + * + */ + +package com.linkedin.kafka.cruisecontrol.analyzer.goals; + +import com.linkedin.kafka.cruisecontrol.analyzer.OptimizationOptions; +import com.linkedin.kafka.cruisecontrol.analyzer.ActionAcceptance; +import com.linkedin.kafka.cruisecontrol.analyzer.ActionType; +import com.linkedin.kafka.cruisecontrol.analyzer.AnalyzerUtils; +import com.linkedin.kafka.cruisecontrol.analyzer.BalancingConstraint; +import com.linkedin.kafka.cruisecontrol.analyzer.BalancingAction; +import com.linkedin.kafka.cruisecontrol.common.Statistic; +import com.linkedin.kafka.cruisecontrol.exception.OptimizationFailureException; +import com.linkedin.kafka.cruisecontrol.model.Broker; +import com.linkedin.kafka.cruisecontrol.model.ClusterModel; +import com.linkedin.kafka.cruisecontrol.model.ClusterModelStats; +import com.linkedin.kafka.cruisecontrol.model.Replica; +import com.linkedin.kafka.cruisecontrol.monitor.ModelCompletenessRequirements; +import org.apache.kafka.common.TopicPartition; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.SortedSet; +import java.util.TreeSet; +import java.util.stream.Collectors; + +import static com.linkedin.kafka.cruisecontrol.analyzer.ActionAcceptance.ACCEPT; +import static com.linkedin.kafka.cruisecontrol.analyzer.ActionAcceptance.REPLICA_REJECT; +import static com.linkedin.kafka.cruisecontrol.analyzer.AnalyzerUtils.EPSILON; +import static com.linkedin.kafka.cruisecontrol.analyzer.goals.GoalUtils.MIN_NUM_VALID_WINDOWS_FOR_SELF_HEALING; +import static com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaDistributionAbstractGoal.ChangeType.ADD; +import static com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaDistributionAbstractGoal.ChangeType.REMOVE; + + +/** + * Soft goal to balance the number of leader replicas of each topic. + *
    + *
  • Under: (the average number of topic leader replicas per broker) * (1 + topic leader replica count balance percentage)
    • + *
  • Above: (the average number of topic leader replicas per broker) * Math.max(0, 1 - topic leader replica count balance percentage)
    • + *
+ * Also see: {@link com.linkedin.kafka.cruisecontrol.config.constants.AnalyzerConfig#TOPIC_LEADER_COUNT_BALANCE_THRESHOLD_CONFIG}, + * {@link com.linkedin.kafka.cruisecontrol.config.constants.AnalyzerConfig#GOAL_VIOLATION_DISTRIBUTION_THRESHOLD_MULTIPLIER_CONFIG}, + * and {@link #balancePercentageWithMargin(OptimizationOptions)}. + */ +public class TopicLeaderDistributionGoal extends AbstractGoal { + private static final Logger LOG = LoggerFactory.getLogger(TopicLeaderDistributionGoal.class); + private static final double BALANCE_MARGIN = 0.9; + // Flag to indicate whether the self healing failed to relocate all offline replicas away from dead brokers or broken + // disks in its initial attempt and currently omitting the replica balance limit to relocate remaining replicas. + private boolean _fixOfflineReplicasOnly; + + private final Map> _brokerIdsAboveBalanceUpperLimitByTopic; + private final Map> _brokerIdsUnderBalanceLowerLimitByTopic; + // Must contain only the topics to be rebalanced. + private final Map _avgTopicReplicasOnAliveBroker; + // Must contain all topics to ensure that the lower priority goals work w/o an NPE. + private final Map _balanceUpperLimitByTopic; + private final Map _balanceLowerLimitByTopic; + + /** + * A soft goal to balance collocations of leader replicas of the same topic. + */ + public TopicLeaderDistributionGoal() { + _brokerIdsAboveBalanceUpperLimitByTopic = new HashMap<>(); + _brokerIdsUnderBalanceLowerLimitByTopic = new HashMap<>(); + _avgTopicReplicasOnAliveBroker = new HashMap<>(); + _balanceUpperLimitByTopic = new HashMap<>(); + _balanceLowerLimitByTopic = new HashMap<>(); + } + + public TopicLeaderDistributionGoal(BalancingConstraint balancingConstraint) { + this(); + _balancingConstraint = balancingConstraint; + } + + /** + * To avoid churns, we add a balance margin to the user specified rebalance threshold. e.g. when user sets the + * threshold to be {@link BalancingConstraint#topicLeaderBalancePercentage()}, we use + * ({@link BalancingConstraint#topicLeaderBalancePercentage()}-1)*{@link #BALANCE_MARGIN} instead. + * + * @param optimizationOptions Options to adjust balance percentage with margin in case goal optimization is triggered + * by goal violation detector. + * @return The rebalance threshold with a margin. + */ + private double balancePercentageWithMargin(OptimizationOptions optimizationOptions) { + double balancePercentage = optimizationOptions.isTriggeredByGoalViolation() + ? _balancingConstraint.topicLeaderBalancePercentage() + * _balancingConstraint.goalViolationDistributionThresholdMultiplier() + : _balancingConstraint.topicLeaderBalancePercentage(); + + return (balancePercentage - 1) * BALANCE_MARGIN; + } + + /** + * @param topic Topic for which the upper limit is requested. + * @param optimizationOptions Options to adjust balance upper limit in case goal optimization is triggered by goal + * violation detector. + * @return The topic replica balance upper threshold in number of topic replicas. + */ + private int balanceUpperLimit(String topic, OptimizationOptions optimizationOptions) { + return (int) Math.ceil(_avgTopicReplicasOnAliveBroker.get(topic) + * (1 + balancePercentageWithMargin(optimizationOptions))); + } + + /** + * @param topic Topic for which the lower limit is requested. + * @param optimizationOptions Options to adjust balance lower limit in case goal optimization is triggered by goal + * violation detector. + * @return The replica balance lower threshold in number of topic replicas. + */ + private int balanceLowerLimit(String topic, OptimizationOptions optimizationOptions) { + return (int) Math.floor(_avgTopicReplicasOnAliveBroker.get(topic) + * Math.max(0, (1 - balancePercentageWithMargin(optimizationOptions)))); + } + + /** + * Check whether the given action is acceptable by this goal. An action is acceptable if the number of topic leader replicas at + * (1) the source broker does not go under the allowed limit. + * (2) the destination broker does not go over the allowed limit. + * + * @param action Action to be checked for acceptance. + * @param clusterModel The state of the cluster. + * @return {@link ActionAcceptance#ACCEPT} if the action is acceptable by this goal, + * {@link ActionAcceptance#REPLICA_REJECT} otherwise. + */ + @Override + public ActionAcceptance actionAcceptance(BalancingAction action, ClusterModel clusterModel) { + Broker sourceBroker = clusterModel.broker(action.sourceBrokerId()); + Replica sourceReplica = sourceBroker.replica(action.topicPartition()); + Broker destinationBroker = clusterModel.broker(action.destinationBrokerId()); + String sourceTopic = action.topic(); + + boolean accept = true; + + switch (action.balancingAction()) { + case INTER_BROKER_REPLICA_SWAP: + String destinationTopic = action.destinationTopic(); + Replica destinationReplica = destinationBroker.replica(action.destinationTopicPartition()); + if (sourceTopic.equals(destinationTopic) && sourceReplica.isLeader() == destinationReplica.isLeader()) { + break; + } + if (sourceReplica.isLeader()) { + accept &= isLeaderMovementSatisfiable(sourceTopic, sourceBroker, destinationBroker); + } + if (destinationReplica.isLeader()) { + accept &= isLeaderMovementSatisfiable(destinationTopic, destinationBroker, sourceBroker); + } + break; + case INTER_BROKER_REPLICA_MOVEMENT: + if (sourceReplica.isLeader()) { + accept = isLeaderMovementSatisfiable(sourceTopic, sourceBroker, destinationBroker); + } + break; + case LEADERSHIP_MOVEMENT: + accept = isLeaderMovementSatisfiable(sourceTopic, sourceBroker, destinationBroker); + break; + default: + throw new IllegalArgumentException("Unsupported balancing action " + action.balancingAction() + " is provided."); + } + + return accept ? ACCEPT : REPLICA_REJECT; + } + + private boolean isLeaderMovementSatisfiable(String topic, Broker sourceBroker, Broker destinationBroker) { + return isReplicaCountUnderBalanceUpperLimitAfterChange(topic, destinationBroker, ADD) + && isReplicaCountAboveBalanceLowerLimitAfterChange(topic, sourceBroker, REMOVE); + } + + private boolean isReplicaCountUnderBalanceUpperLimitAfterChange(String topic, + Broker broker, + ReplicaDistributionGoal.ChangeType changeType) { + int numTopicReplicas = broker.numLeadersFor(topic); + int brokerBalanceUpperLimit = broker.isAlive() ? _balanceUpperLimitByTopic.get(topic) : 0; + + return changeType == ADD ? numTopicReplicas + 1 <= brokerBalanceUpperLimit : numTopicReplicas - 1 <= brokerBalanceUpperLimit; + } + + private boolean isReplicaCountAboveBalanceLowerLimitAfterChange(String topic, + Broker broker, + ReplicaDistributionGoal.ChangeType changeType) { + int numTopicReplicas = broker.numLeadersFor(topic); + int brokerBalanceLowerLimit = broker.isAlive() ? _balanceLowerLimitByTopic.get(topic) : 0; + + return changeType == ADD ? numTopicReplicas + 1 >= brokerBalanceLowerLimit : numTopicReplicas - 1 >= brokerBalanceLowerLimit; + } + + @Override + public ClusterModelStatsComparator clusterModelStatsComparator() { + return new TopicLeaderDistrGoalStatsComparator(); + } + + @Override + public ModelCompletenessRequirements clusterModelCompletenessRequirements() { + return new ModelCompletenessRequirements(MIN_NUM_VALID_WINDOWS_FOR_SELF_HEALING, 0.0, true); + } + + /** + * Get the name of this goal. Name of a goal provides an identification for the goal in human readable format. + */ + @Override + public String name() { + return TopicLeaderDistributionGoal.class.getSimpleName(); + } + + @Override + public boolean isHardGoal() { + return false; + } + + /** + * Get brokers that the rebalance process will go over to apply balancing actions to replicas they contain. + * + * @param clusterModel The state of the cluster. + * @return A collection of brokers that the rebalance process will go over to apply balancing actions to replicas + * they contain. + */ + @Override + protected SortedSet brokersToBalance(ClusterModel clusterModel) { + return clusterModel.brokers(); + } + + /** + * Get the set of topics to rebalance. If there are self healing eligible replicas, gets only their topics. + * Otherwise gets all topics except excludedTopics. + * + * @param clusterModel The state of the cluster. + * @param excludedTopics Topics to exclude from the rebalance. + * @return The set of topics to rebalance. + */ + private Set topicsToRebalance(ClusterModel clusterModel, Set excludedTopics) { + Set topicsToRebalance; + if (!clusterModel.selfHealingEligibleReplicas().isEmpty()) { + topicsToRebalance = new HashSet<>(); + for (Replica replica : clusterModel.selfHealingEligibleReplicas()) { + topicsToRebalance.add(replica.topicPartition().topic()); + } + } else { + topicsToRebalance = new HashSet<>(clusterModel.topics()); + topicsToRebalance.removeAll(excludedTopics); + } + + if (topicsToRebalance.isEmpty()) { + LOG.warn("All topics are excluded from {}.", name()); + } + + return topicsToRebalance; + } + + /** + * Initiates this goal. + * + * @param clusterModel The state of the cluster. + * @param optimizationOptions Options to take into account during optimization. + */ + @Override + protected void initGoalState(ClusterModel clusterModel, OptimizationOptions optimizationOptions) { + Set excludedTopics = optimizationOptions.excludedTopics(); + Set topicsToRebalance = topicsToRebalance(clusterModel, excludedTopics); + + // Initialize the average leader replicas on an alive broker. + for (String topic : clusterModel.topics()) { + int numTopicLeaders = clusterModel.numTopicLeaders(topic); + _avgTopicReplicasOnAliveBroker.put(topic, (numTopicLeaders / (double) clusterModel.aliveBrokers().size())); + _balanceUpperLimitByTopic.put(topic, balanceUpperLimit(topic, optimizationOptions)); + _balanceLowerLimitByTopic.put(topic, balanceLowerLimit(topic, optimizationOptions)); + // Retain only the topics to rebalance in _avgTopicReplicasOnAliveBroker + if (!topicsToRebalance.contains(topic)) { + _avgTopicReplicasOnAliveBroker.remove(topic); + } + } + _fixOfflineReplicasOnly = false; + } + + /** + * Check if requirements of this goal are not violated if this proposal is applied to the given cluster state, + * false otherwise. + * + * @param clusterModel The state of the cluster. + * @param action Action containing information about potential modification to the given cluster model. Assumed to be + * of type {@link ActionType#INTER_BROKER_REPLICA_MOVEMENT} or {@link ActionType#INTER_BROKER_REPLICA_SWAP}. + * @return True if requirements of this goal are not violated if this proposal is applied to the given cluster state, + * false otherwise. + */ + @Override + protected boolean selfSatisfied(ClusterModel clusterModel, BalancingAction action) { + if (action.balancingAction() == ActionType.INTER_BROKER_REPLICA_SWAP) { + return true; + } + + Broker sourceBroker = clusterModel.broker(action.sourceBrokerId()); + // The action must be executed if currently fixing offline replicas only and the offline source replica is proposed + // to be moved to another broker. + if (_fixOfflineReplicasOnly && sourceBroker.replica(action.topicPartition()).isCurrentOffline()) { + return action.balancingAction() == ActionType.INTER_BROKER_REPLICA_MOVEMENT; + } + + //Check that destination and source would not become unbalanced. + Broker destinationBroker = clusterModel.broker(action.destinationBrokerId()); + String sourceTopic = action.topic(); + + return isReplicaCountUnderBalanceUpperLimitAfterChange(sourceTopic, destinationBroker, ADD) + && isReplicaCountAboveBalanceLowerLimitAfterChange(sourceTopic, sourceBroker, REMOVE); + } + + /** + * Update goal state after one round of self-healing / rebalance. + * @param clusterModel The state of the cluster. + * @param optimizationOptions Options to take into account during optimization. + */ + @Override + protected void updateGoalState(ClusterModel clusterModel, OptimizationOptions optimizationOptions) + throws OptimizationFailureException { + if (!_brokerIdsAboveBalanceUpperLimitByTopic.isEmpty()) { + _brokerIdsAboveBalanceUpperLimitByTopic.clear(); + _succeeded = false; + } + if (!_brokerIdsUnderBalanceLowerLimitByTopic.isEmpty()) { + _brokerIdsUnderBalanceLowerLimitByTopic.clear(); + _succeeded = false; + } + // Sanity check: No self-healing eligible replica should remain at a dead broker/disk. + try { + GoalUtils.ensureNoOfflineReplicas(clusterModel, name()); + } catch (OptimizationFailureException ofe) { + if (_fixOfflineReplicasOnly) { + throw ofe; + } + _fixOfflineReplicasOnly = true; + LOG.info("Ignoring topic replica balance limit to move replicas from dead brokers/disks."); + return; + } + // Sanity check: No replica should be moved to a broker, which used to host any replica of the same partition on its broken disk. + GoalUtils.ensureReplicasMoveOffBrokersWithBadDisks(clusterModel, name()); + finish(); + } + + @Override + public void finish() { + _finished = true; + } + + private static boolean skipBrokerRebalance(Broker broker, + ClusterModel clusterModel, + Collection replicas, + boolean requireLessReplicas, + boolean requireMoreReplicas, + boolean hasOfflineTopicReplicas, + boolean moveImmigrantReplicaOnly) { + boolean hasImmigrantTopicReplicas = replicas.stream().anyMatch(replica -> broker.immigrantReplicas().contains(replica)); + if (broker.isAlive() && !requireMoreReplicas && !requireLessReplicas) { + LOG.trace("Skip rebalance: Broker {} is already within the limit for replicas {}.", broker, replicas); + return true; + } else if (!clusterModel.newBrokers().isEmpty() && !broker.isNew() && !requireLessReplicas) { + LOG.trace("Skip rebalance: Cluster has new brokers and this broker {} is not new, but does not require less load " + + "for replicas {}. Hence, it does not have any offline replicas.", broker, replicas); + return true; + } else if (!clusterModel.selfHealingEligibleReplicas().isEmpty() && requireLessReplicas + && !hasOfflineTopicReplicas && !hasImmigrantTopicReplicas) { + LOG.trace("Skip rebalance: Cluster is in self-healing mode and the broker {} requires less load, but none of its " + + "current offline or immigrant replicas are from the topic being balanced {}.", broker, replicas); + return true; + } else if (moveImmigrantReplicaOnly && requireLessReplicas && !hasImmigrantTopicReplicas) { + LOG.trace("Skip rebalance: Only immigrant replicas can be moved, but none of broker {}'s " + + "current immigrant replicas are from the topic being balanced {}.", broker, replicas); + return true; + } + + return false; + } + + private static Set retainCurrentOfflineBrokerReplicas(Broker broker, Collection replicas) { + Set offlineReplicas = new HashSet<>(replicas); + offlineReplicas.retainAll(broker.currentOfflineReplicas()); + + return offlineReplicas; + } + + private boolean isTopicExcludedFromRebalance(String topic) { + return _avgTopicReplicasOnAliveBroker.get(topic) == null; + } + + /** + * Rebalance the given broker without violating the constraints of the current goal and optimized goals. + * + * @param broker Broker to be balanced. + * @param clusterModel The state of the cluster. + * @param optimizedGoals Optimized goals. + * @param optimizationOptions Options to take into account during optimization. + */ + @Override + protected void rebalanceForBroker(Broker broker, + ClusterModel clusterModel, + Set optimizedGoals, + OptimizationOptions optimizationOptions) { + LOG.debug("Rebalancing broker {} [limits] lower: {} upper: {}.", broker.id(), _balanceLowerLimitByTopic, + _balanceUpperLimitByTopic); + + for (String topic : broker.topics()) { + if (isTopicExcludedFromRebalance(topic)) { + continue; + } + + Collection leaderReplicas = broker.leadersFor(topic); + int numLeaderReplicas = leaderReplicas.size(); + int numOfflineTopicReplicas = retainCurrentOfflineBrokerReplicas(broker, leaderReplicas).size(); + + boolean requireLessReplicas = numOfflineTopicReplicas > 0 || numLeaderReplicas > _balanceUpperLimitByTopic.get(topic); + boolean requireMoreReplicas = broker.isAlive() && numLeaderReplicas - numOfflineTopicReplicas < _balanceLowerLimitByTopic.get(topic); + + if (skipBrokerRebalance(broker, clusterModel, leaderReplicas, requireLessReplicas, requireMoreReplicas, numOfflineTopicReplicas > 0, + optimizationOptions.onlyMoveImmigrantReplicas())) { + continue; + } + + // Update broker ids over the balance limit for logging purposes. + if (requireLessReplicas && rebalanceByMovingLeadershipOut(broker, topic, clusterModel, optimizedGoals, optimizationOptions)) { + _brokerIdsAboveBalanceUpperLimitByTopic.computeIfAbsent(topic, t -> new HashSet<>()).add(broker.id()); + LOG.debug("Failed to sufficiently decrease leaders of topic {} in broker {} with leadership movements. Leaders: {}.", + topic, broker.id(), broker.numLeadersFor(topic)); + } + if (requireMoreReplicas && rebalanceByMovingLeadershipIn(broker, topic, clusterModel, optimizedGoals, optimizationOptions)) { + _brokerIdsUnderBalanceLowerLimitByTopic.computeIfAbsent(topic, t -> new HashSet<>()).add(broker.id()); + LOG.debug("Failed to sufficiently increase leaders of topic {} in broker {} with leadership movements. Leaders: {}.", + topic, broker.id(), broker.numLeadersFor(topic)); + } + if (!_brokerIdsAboveBalanceUpperLimitByTopic.getOrDefault(topic, Collections.emptySet()).contains(broker.id()) + && !_brokerIdsUnderBalanceLowerLimitByTopic.getOrDefault(topic, Collections.emptySet()).contains(broker.id())) { + LOG.debug("Successfully balanced leaders of topic {} in broker {} by moving leadership. Leaders: {}", + topic, broker.id(), broker.numLeadersFor(topic)); + } + } + } + + /** + * Attempt to decrease the number of topic partitions led by the broker first by moving the leadership to follower + * replicas, and second by swapping leader replicas with follower replicas of partitions not hosted by the broker + * currently. + * + * @param broker Broker that leads too many partitions of the given topic. + * @param topic Topic to rebalance. + * @param clusterModel The state of the cluster. + * @param optimizedGoals Optimized goals. + * @param optimizationOptions Options to take into account during optimization. + * @return true if rebalancing was not successful. + */ + private boolean rebalanceByMovingLeadershipOut(Broker broker, + String topic, + ClusterModel clusterModel, + Set optimizedGoals, + OptimizationOptions optimizationOptions) { + if (!clusterModel.deadBrokers().isEmpty()) { + return true; + } + + // Try to rebalance by moving leadership to follower replicas. + for (Replica leader : broker.leadersFor(topic)) { + final Set candidateBrokers = clusterModel.partition(leader.topicPartition()).partitionBrokers().stream() + .filter(b -> b != broker && !b.replica(leader.topicPartition()).isCurrentOffline()) + .filter(b -> b.numLeadersFor(topic) < _balanceUpperLimitByTopic.get(topic)) + .collect(Collectors.toSet()); + Broker b = maybeApplyBalancingAction(clusterModel, + leader, + candidateBrokers, + ActionType.LEADERSHIP_MOVEMENT, + optimizedGoals, + optimizationOptions); + if (b != null && broker.numLeadersFor(topic) <= _balanceUpperLimitByTopic.get(topic)) { + return false; + } + } + + // Try to rebalance by swapping one of the broker follower replicas with a leader replica from another broker. + Set brokerPartitions = broker.replicasOfTopicInBroker(topic).stream() + .map(Replica::topicPartition) + .collect(Collectors.toSet()); + for (Replica leader : broker.leadersFor(topic)) { + Iterable candidateBrokers = clusterModel.brokers().stream() + .filter(b -> b.numLeadersFor(topic) < _balanceUpperLimitByTopic.get(topic)) + .collect(Collectors.toList()); + for (Broker candidateBroker : candidateBrokers) { + List candidateReplicas = candidateBroker.replicasOfTopicInBroker(topic).stream() + .filter(r -> !r.isLeader() && !r.isCurrentOffline() && !brokerPartitions.contains(r.topicPartition())) + .collect(Collectors.toList()); + Replica r = maybeApplySwapAction(clusterModel, leader, new TreeSet<>(candidateReplicas), optimizedGoals, optimizationOptions); + if (r != null && broker.numLeadersFor(topic) <= _balanceUpperLimitByTopic.get(topic)) { + return false; + } + } + } + + return true; + } + + /** + * Attempt to increase the number of topic partitions led by the broker first by moving the leadership to follower + * replicas hosted by the broker, and second by swapping follower replicas hosted by the broker with leader replicas + * of partitions not currently hosted by the broker. + * + * @param broker Broker that leads too few partitions of the given topic. + * @param topic Topic to rebalance. + * @param clusterModel The state of the cluster. + * @param optimizedGoals Optimized goals. + * @param optimizationOptions Options to take into account during optimization. + * @return true if rebalancing was not successful. + */ + private boolean rebalanceByMovingLeadershipIn(Broker broker, + String topic, + ClusterModel clusterModel, + Set optimizedGoals, + OptimizationOptions optimizationOptions) { + if (!clusterModel.deadBrokers().isEmpty() + || optimizationOptions.excludedBrokersForLeadership().contains(broker.id())) { + return true; + } + + // Try to rebalance by making one of the replicas the broker is already hosting a leader. + Set candidateBrokers = Collections.singleton(broker); + for (Replica replica : broker.replicasOfTopicInBroker(topic)) { + if (replica.isLeader() || replica.isCurrentOffline()) { + continue; + } + Broker b = maybeApplyBalancingAction(clusterModel, + clusterModel.partition(replica.topicPartition()).leader(), + candidateBrokers, + ActionType.LEADERSHIP_MOVEMENT, + optimizedGoals, + optimizationOptions); + if (b != null && broker.numLeadersFor(topic) >= _balanceLowerLimitByTopic.get(topic)) { + return false; + } + } + + // Try to rebalance by swapping one of the broker non-leader replicas with a leader replica from another broker. + Set brokerPartitions = broker.replicasOfTopicInBroker(topic).stream() + .map(Replica::topicPartition) + .collect(Collectors.toSet()); + candidateBrokers = clusterModel.brokers().stream() + .filter(b -> b.numLeadersFor(topic) > _balanceLowerLimitByTopic.get(topic)) + .collect(Collectors.toSet()); + for (Broker candidateBroker : candidateBrokers) { + Iterable leaders = candidateBroker.leaderReplicas().stream() + .filter(r -> r.topicPartition().topic().equals(topic) && !r.isCurrentOffline()) + .filter(r -> !brokerPartitions.contains(r.topicPartition())) + .collect(Collectors.toList()); + for (Replica leader : leaders) { + List candidateReplicas = broker.replicasOfTopicInBroker(topic).stream() + .filter(r -> !r.isLeader()) + .collect(Collectors.toList()); + Replica r = maybeApplySwapAction(clusterModel, leader, new TreeSet<>(candidateReplicas), optimizedGoals, optimizationOptions); + if (r != null && broker.numLeadersFor(topic) >= _balanceLowerLimitByTopic.get(topic)) { + return false; + } + } + } + + return true; + } + + private class TopicLeaderDistrGoalStatsComparator implements ClusterModelStatsComparator { + private String _reasonForLastNegativeResult; + + @Override + public int compare(ClusterModelStats stats1, ClusterModelStats stats2) { + // Standard deviation of number of topic leader replicas over brokers in the current must be less than the + // pre-optimized stats. + double stdDev1 = stats1.topicLeaderStats().get(Statistic.ST_DEV).doubleValue(); + double stdDev2 = stats2.topicLeaderStats().get(Statistic.ST_DEV).doubleValue(); + int result = AnalyzerUtils.compare(stdDev2, stdDev1, EPSILON); + if (result < 0) { + _reasonForLastNegativeResult = String.format("Violated %s. [Std Deviation of Topic Leader Replica Distribution] post-" + + "optimization:%.3f pre-optimization:%.3f", name(), stdDev1, stdDev2); + } + return result; + } + + @Override + public String explainLastComparison() { + return _reasonForLastNegativeResult; + } + } +} diff --git a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/config/constants/AnalyzerConfig.java b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/config/constants/AnalyzerConfig.java index 99c94230c2..86a7dd55e7 100644 --- a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/config/constants/AnalyzerConfig.java +++ b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/config/constants/AnalyzerConfig.java @@ -133,6 +133,14 @@ public final class AnalyzerConfig { + " the average replica count for each topic. A balance limit is set via topic.replica.count.balance.threshold config." + " If the difference between the computed limit and the average replica count for the relevant topic is greater than" + " the value specified by this config, the limit is adjusted accordingly."; + /** + * topic.leader.count.balance.threshold + */ + public static final String TOPIC_LEADER_COUNT_BALANCE_THRESHOLD_CONFIG = "topic.leader.count.balance.threshold"; + public static final double DEFAULT_TOPIC_LEADER_COUNT_BALANCE_THRESHOLD = 3.0; + public static final String TOPIC_LEADER_COUNT_BALANCE_THRESHOLD_DOC = "The maximum allowed extent of unbalance for " + + "leader replica distribution from each topic. For example, 1.80 means the highest topic leader replica count of a " + + "broker should not be above 1.80x of average leader replica count of all brokers for the same topic."; /** * cpu.capacity.threshold @@ -524,6 +532,12 @@ public static ConfigDef define(ConfigDef configDef) { atLeast(1), ConfigDef.Importance.MEDIUM, TOPIC_REPLICA_COUNT_BALANCE_MAX_GAP_DOC) + .define(TOPIC_LEADER_COUNT_BALANCE_THRESHOLD_CONFIG, + ConfigDef.Type.DOUBLE, + DEFAULT_TOPIC_LEADER_COUNT_BALANCE_THRESHOLD, + atLeast(1), + ConfigDef.Importance.HIGH, + TOPIC_LEADER_COUNT_BALANCE_THRESHOLD_DOC) .define(CPU_CAPACITY_THRESHOLD_CONFIG, ConfigDef.Type.DOUBLE, DEFAULT_CPU_CAPACITY_THRESHOLD, diff --git a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/executor/ExecutionTaskPlanner.java b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/executor/ExecutionTaskPlanner.java index 2f3f671c08..93cb087642 100644 --- a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/executor/ExecutionTaskPlanner.java +++ b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/executor/ExecutionTaskPlanner.java @@ -5,7 +5,6 @@ package com.linkedin.kafka.cruisecontrol.executor; import com.linkedin.cruisecontrol.common.utils.Utils; -import com.linkedin.kafka.cruisecontrol.KafkaCruiseControlUtils; import com.linkedin.kafka.cruisecontrol.config.KafkaCruiseControlConfig; import com.linkedin.kafka.cruisecontrol.executor.concurrency.ExecutionConcurrencyManager; import com.linkedin.kafka.cruisecontrol.executor.strategy.BaseReplicaMovementStrategy; @@ -398,8 +397,7 @@ public List getInterBrokerReplicaMovementTasks(Map destinationBrokers = task.proposal().replicasToAdd().stream().mapToInt(ReplicaPlacementInfo::brokerId) .boxed().collect(Collectors.toSet()); - if (brokerInvolved.contains(sourceBroker) - || KafkaCruiseControlUtils.containsAny(brokerInvolved, destinationBrokers)) { + if (brokerInvolved.contains(sourceBroker)) { continue; } TopicPartition tp = task.proposal().topicPartition(); diff --git a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/model/Broker.java b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/model/Broker.java index 73cfe34b59..1b6069dc3f 100644 --- a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/model/Broker.java +++ b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/model/Broker.java @@ -23,6 +23,7 @@ import java.util.SortedMap; import java.util.TreeMap; import java.util.function.Function; +import java.util.stream.Collectors; import org.apache.kafka.common.TopicPartition; import static com.linkedin.cruisecontrol.common.utils.Utils.validateNotNull; @@ -203,6 +204,15 @@ public int numLeadersFor(String topicName) { return (int) replicasOfTopicInBroker(topicName).stream().filter(Replica::isLeader).count(); } + /** + * Get leader replicas for topic. + * @param topic Topic of the requested replicas. + * @return Leader replicas in this broker sharing the given topic. + */ + public Collection leadersFor(String topic) { + return replicasOfTopicInBroker(topic).stream().filter(Replica::isLeader).collect(Collectors.toList()); + } + /** * @return {@code true} if the broker is not dead, {@code false} otherwise. */ diff --git a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/model/ClusterModel.java b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/model/ClusterModel.java index 768d999501..671c97a4b7 100644 --- a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/model/ClusterModel.java +++ b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/model/ClusterModel.java @@ -671,6 +671,19 @@ public int numTopicReplicas(String topic) { return _numReplicasByTopic.getOrDefault(topic, 0); } + /** + * Get the number of leader replicas with the given topic name in cluster. + * @param topic Name of the topic for which the number of leader replicas in cluster will be counted. + * @return Number of leader replicas with the given topic name in cluster. + */ + public int numTopicLeaders(String topic) { + int numTopicLeaders = 0; + for (Broker broker : brokers()) { + numTopicLeaders += broker.numLeadersFor(topic); + } + return numTopicLeaders; + } + /** * Get the number of leader replicas in cluster. * diff --git a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/model/ClusterModelStats.java b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/model/ClusterModelStats.java index d6bb2e22b0..b29e03fde0 100644 --- a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/model/ClusterModelStats.java +++ b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/model/ClusterModelStats.java @@ -37,6 +37,7 @@ public class ClusterModelStats { private final Map _replicaStats; private final Map _leaderReplicaStats; private final Map _topicReplicaStats; + private final Map _topicLeaderStats; private int _numBrokers; private int _numReplicasInCluster; private int _numPartitionsWithOfflineReplicas; @@ -62,6 +63,7 @@ public class ClusterModelStats { _replicaStats = new HashMap<>(); _leaderReplicaStats = new HashMap<>(); _topicReplicaStats = new HashMap<>(); + _topicLeaderStats = new HashMap<>(); _numBrokers = 0; _numReplicasInCluster = 0; _numPartitionsWithOfflineReplicas = 0; @@ -94,6 +96,7 @@ ClusterModelStats populate(ClusterModel clusterModel, BalancingConstraint balanc numForReplicas(clusterModel, brokers, aliveBrokers); numForLeaderReplicas(brokers, aliveBrokers); numForAvgTopicReplicas(clusterModel, brokers, topics); + numForAvgTopicLeaders(clusterModel, brokers, topics, aliveBrokers); _utilizationMatrix = clusterModel.utilizationMatrix(); _numSnapshotWindows = clusterModel.load().numWindows(); _monitoredPartitionsRatio = clusterModel.monitoredPartitionsRatio(); @@ -136,6 +139,13 @@ public Map topicReplicaStats() { return Collections.unmodifiableMap(_topicReplicaStats); } + /** + * @return Topic leader stats for the cluster instance that the object was populated with. + */ + public Map topicLeaderStats() { + return Collections.unmodifiableMap(_topicLeaderStats); + } + /** * @return The number of brokers for the cluster instance that the object was populated with. */ @@ -475,6 +485,49 @@ private void numForAvgTopicReplicas(ClusterModel clusterModel, SortedSet _topicReplicaStats.put(Statistic.ST_DEV, _topicReplicaStats.get(Statistic.ST_DEV).doubleValue() / _numTopics); } + /** + * Generate statistics for leader replicas of each topic in the given cluster. + * Average and standard deviation calculations are based on alive brokers. + * @param clusterModel The state of the cluster. + * @param brokers Brokers in the cluster. + * @param topics Topics in the cluster. + * @param aliveBrokers Alive brokers in the cluster. + */ + private void numForAvgTopicLeaders(ClusterModel clusterModel, + SortedSet brokers, + Set topics, + Set aliveBrokers) { + _topicLeaderStats.put(Statistic.AVG, 0.0); + _topicLeaderStats.put(Statistic.MAX, 0); + _topicLeaderStats.put(Statistic.MIN, Integer.MAX_VALUE); + _topicLeaderStats.put(Statistic.ST_DEV, 0.0); + int numAliveBrokers = aliveBrokers.size(); + for (String topic : topics) { + int maxTopicLeadersInBroker = 0; + int minTopicLeadersInBroker = Integer.MAX_VALUE; + double avgTopicLeaders = ((double) clusterModel.numTopicLeaders(topic)) / numAliveBrokers; + double variance = 0.0; + for (Broker broker : brokers) { + int numTopicLeadersInBroker = broker.numLeadersFor(topic); + maxTopicLeadersInBroker = Math.max(maxTopicLeadersInBroker, numTopicLeadersInBroker); + minTopicLeadersInBroker = Math.min(minTopicLeadersInBroker, numTopicLeadersInBroker); + if (broker.isAlive()) { + // Standard deviation of leader replicas in alive brokers. + variance += (Math.pow(numTopicLeadersInBroker - avgTopicLeaders, 2) / numAliveBrokers); + } + } + _topicLeaderStats.put(Statistic.AVG, _topicLeaderStats.get(Statistic.AVG).doubleValue() + avgTopicLeaders); + _topicLeaderStats.put(Statistic.MAX, + Math.max(_topicLeaderStats.get(Statistic.MAX).intValue(), maxTopicLeadersInBroker)); + _topicLeaderStats.put(Statistic.MIN, + Math.min(_topicLeaderStats.get(Statistic.MIN).intValue(), minTopicLeadersInBroker)); + _topicLeaderStats.put(Statistic.ST_DEV, (Double) _topicLeaderStats.get(Statistic.ST_DEV) + Math.sqrt(variance)); + } + + _topicLeaderStats.put(Statistic.AVG, _topicLeaderStats.get(Statistic.AVG).doubleValue() / _numTopics); + _topicLeaderStats.put(Statistic.ST_DEV, _topicLeaderStats.get(Statistic.ST_DEV).doubleValue() / _numTopics); + } + /** * Generate statistics for disks in the given cluster. * For each alive disk on disk broker in the cluster, check whether its utilization percentage is within the range centered diff --git a/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/analyzer/DeterministicClusterTest.java b/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/analyzer/DeterministicClusterTest.java index 2a644ed4d4..0313ea8efa 100644 --- a/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/analyzer/DeterministicClusterTest.java +++ b/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/analyzer/DeterministicClusterTest.java @@ -24,6 +24,7 @@ import com.linkedin.kafka.cruisecontrol.analyzer.goals.RackAwareGoal; import com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaCapacityGoal; import com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaDistributionGoal; +import com.linkedin.kafka.cruisecontrol.analyzer.goals.TopicLeaderDistributionGoal; import com.linkedin.kafka.cruisecontrol.analyzer.goals.TopicReplicaDistributionGoal; import com.linkedin.kafka.cruisecontrol.analyzer.kafkaassigner.KafkaAssignerDiskUsageDistributionGoal; import com.linkedin.kafka.cruisecontrol.analyzer.kafkaassigner.KafkaAssignerEvenRackAwareGoal; @@ -115,6 +116,7 @@ public static Collection data() { LeaderReplicaDistributionGoal.class.getName(), LeaderBytesInDistributionGoal.class.getName(), TopicReplicaDistributionGoal.class.getName(), + TopicLeaderDistributionGoal.class.getName(), PreferredLeaderElectionGoal.class.getName()); List verifications = Arrays.asList(NEW_BROKERS, BROKEN_BROKERS, REGRESSION); diff --git a/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/analyzer/RandomClusterTest.java b/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/analyzer/RandomClusterTest.java index dd3b51bfe5..84f51a62b1 100644 --- a/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/analyzer/RandomClusterTest.java +++ b/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/analyzer/RandomClusterTest.java @@ -25,6 +25,7 @@ import com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaCapacityGoal; import com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaDistributionGoal; import com.linkedin.kafka.cruisecontrol.analyzer.goals.TopicReplicaDistributionGoal; +import com.linkedin.kafka.cruisecontrol.analyzer.goals.TopicLeaderDistributionGoal; import com.linkedin.kafka.cruisecontrol.analyzer.kafkaassigner.KafkaAssignerDiskUsageDistributionGoal; import com.linkedin.kafka.cruisecontrol.analyzer.kafkaassigner.KafkaAssignerEvenRackAwareGoal; import com.linkedin.kafka.cruisecontrol.common.Resource; @@ -118,6 +119,7 @@ public static Collection data(TestConstants.Distribution distribution) LeaderReplicaDistributionGoal.class.getName(), LeaderBytesInDistributionGoal.class.getName(), TopicReplicaDistributionGoal.class.getName(), + TopicLeaderDistributionGoal.class.getName(), PreferredLeaderElectionGoal.class.getName()); List kafkaAssignerGoals = Arrays.asList(KafkaAssignerEvenRackAwareGoal.class.getName(), diff --git a/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/executor/ExecutionTaskPlannerTest.java b/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/executor/ExecutionTaskPlannerTest.java index 3a3a4c24dd..2e66b42a1d 100644 --- a/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/executor/ExecutionTaskPlannerTest.java +++ b/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/executor/ExecutionTaskPlannerTest.java @@ -288,8 +288,8 @@ public void testGetInterBrokerPartitionMovementTasks() { partitionMovementTasks = prioritizeSmallMovementPlanner.getInterBrokerReplicaMovementTasks(readyBrokers, Collections.emptySet(), _defaultPartitionsMaxCap); assertEquals("First task", _partitionMovement0, partitionMovementTasks.get(0).proposal()); - assertEquals("Second task", _partitionMovement2, partitionMovementTasks.get(1).proposal()); - assertEquals("Third task", _partitionMovement3, partitionMovementTasks.get(2).proposal()); + assertEquals("Second task", _partitionMovement3, partitionMovementTasks.get(1).proposal()); + assertEquals("Third task", _partitionMovement2, partitionMovementTasks.get(2).proposal()); assertEquals("Fourth task", _partitionMovement1, partitionMovementTasks.get(3).proposal()); smallUrpMovementPlanner.addExecutionProposals(proposals, strategyOptions, null); @@ -356,8 +356,8 @@ public void testGetInterBrokerPartitionMovementWithMinIsrTasks() { List partitionMovementTasks = prioritizeOneAboveMinIsrMovementPlanner.getInterBrokerReplicaMovementTasks(readyBrokers, Collections.emptySet(), _defaultPartitionsMaxCap); assertEquals("First task", _rf4PartitionMovement2, partitionMovementTasks.get(0).proposal()); - assertEquals("Second task", _rf4PartitionMovement3, partitionMovementTasks.get(1).proposal()); - assertEquals("Third task", _rf4PartitionMovement1, partitionMovementTasks.get(2).proposal()); + assertEquals("Second task", _rf4PartitionMovement1, partitionMovementTasks.get(1).proposal()); + assertEquals("Third task", _rf4PartitionMovement3, partitionMovementTasks.get(2).proposal()); assertEquals("Fourth task", _rf4PartitionMovement0, partitionMovementTasks.get(3).proposal()); } @@ -407,8 +407,8 @@ public void testDynamicConfigReplicaMovementStrategy() { planner.addExecutionProposals(proposals, strategyOptions, new PrioritizeSmallReplicaMovementStrategy()); partitionMovementTasks = planner.getInterBrokerReplicaMovementTasks(readyBrokers, Collections.emptySet(), _defaultPartitionsMaxCap); assertEquals("First task", _partitionMovement0, partitionMovementTasks.get(0).proposal()); - assertEquals("Second task", _partitionMovement2, partitionMovementTasks.get(1).proposal()); - assertEquals("Third task", _partitionMovement3, partitionMovementTasks.get(2).proposal()); + assertEquals("Second task", _partitionMovement3, partitionMovementTasks.get(1).proposal()); + assertEquals("Third task", _partitionMovement2, partitionMovementTasks.get(2).proposal()); assertEquals("Fourth task", _partitionMovement1, partitionMovementTasks.get(3).proposal()); }