@Test public void testSwapReplicas() { Properties props = KafkaCruiseControlUnitTestUtils.getKafkaCruiseControlProperties(); props.setProperty(AnalyzerConfig.MAX_REPLICAS_PER_BROKER_CONFIG, Long.toString(10L)); props.setProperty(AnalyzerConfig.DISK_BALANCE_THRESHOLD_CONFIG, "1.05"); BalancingConstraint balancingConstraint = new BalancingConstraint(new KafkaCruiseControlConfig(props)); KafkaAssignerDiskUsageDistributionGoal goal = new KafkaAssignerDiskUsageDistributionGoal(balancingConstraint); ClusterModel clusterModel = createClusterModel(); Comparator<Replica> replicaComparator = Comparator.comparingDouble((Replica r) -> r.load().expectedUtilizationFor(DISK)) .thenComparing(r -> r); double meanDiskUsage = clusterModel.load().expectedUtilizationFor(DISK) / clusterModel.capacityFor(DISK); assertTrue(goal.swapReplicas(new BrokerAndSortedReplicas(clusterModel.broker(0), replicaComparator), new BrokerAndSortedReplicas(clusterModel.broker(1), replicaComparator), meanDiskUsage, clusterModel, Collections.emptySet())); assertFalse(goal.swapReplicas(new BrokerAndSortedReplicas(clusterModel.broker(0), replicaComparator), new BrokerAndSortedReplicas(clusterModel.broker(2), replicaComparator), meanDiskUsage, clusterModel, Collections.emptySet())); assertTrue(goal.swapReplicas(new BrokerAndSortedReplicas(clusterModel.broker(2), replicaComparator), new BrokerAndSortedReplicas(clusterModel.broker(3), replicaComparator), meanDiskUsage, clusterModel, Collections.emptySet())); }
boolean swapReplicas(BrokerAndSortedReplicas toSwap, BrokerAndSortedReplicas toSwapWith, double meanDiskUsage, ClusterModel clusterModel, Set<String> excludedTopics) { if (LOG.isTraceEnabled()) { LOG.trace("Swapping replicas between broker {}({}) and broker {}({})", toSwap.broker().id(), dWrap(brokerSize(toSwap)), toSwapWith.broker().id(), dWrap(brokerSize(toSwapWith))); } double sizeToChange = toSwap.broker().capacityFor(DISK) * meanDiskUsage - brokerSize(toSwap); NavigableSet<ReplicaWrapper> sortedReplicasToSwap = sortReplicasAscend(toSwap, excludedTopics); NavigableSet<ReplicaWrapper> sortedLeadersToSwapWith = sortReplicasAscend(toSwapWith, excludedTopics); NavigableSet<ReplicaWrapper> sortedFollowersToSwapWith = sortedFollowerReplicas(toSwapWith, excludedTopics); Iterator<ReplicaWrapper> toSwapIter = sizeToChange > 0 ? sortedReplicasToSwap.iterator() : sortedReplicasToSwap.descendingIterator(); while (toSwapIter.hasNext()) { Replica replicaToSwap = toSwapIter.next().replica(); if (excludedTopics.contains(replicaToSwap.topicPartition().topic())) { continue; } if (!possibleToMove(replicaToSwap, toSwapWith.broker(), clusterModel)) { continue; } NavigableSet<ReplicaWrapper> sortedReplicasToSwapWith = replicaToSwap.isLeader() ? sortedLeadersToSwapWith : sortedFollowersToSwapWith; double sizeToSwap = replicaSize(replicaToSwap); if (sizeToChange < 0 && sizeToSwap == 0) { break; } double maxSize = Double.MAX_VALUE; double minSize = Double.MIN_VALUE; if (sizeToChange > 0) { minSize = sizeToSwap; double maxSizeOfBrokerToSwap = diskUsage(toSwapWith) * toSwap.broker().capacityFor(DISK); double currentSizeOfBrokerToSwap = brokerSize(toSwap); maxSize = Math.min(maxSize, maxSizeOfBrokerToSwap - (currentSizeOfBrokerToSwap - sizeToSwap)); double minSizeOfBrokerToSwapWith = diskUsage(toSwap) * toSwapWith.broker().capacityFor(DISK); double currentSizeOfBrokerToSwapWith = brokerSize(toSwapWith); maxSize = Math.min(maxSize, (currentSizeOfBrokerToSwapWith + sizeToSwap) - minSizeOfBrokerToSwapWith); } else { maxSize = sizeToSwap; double minSizeOfBrokerToSwap = diskUsage(toSwapWith) * toSwap.broker().capacityFor(DISK); double currentSizeOfBrokerToSwap = brokerSize(toSwap); minSize = Math.max(minSize, minSizeOfBrokerToSwap - (currentSizeOfBrokerToSwap - sizeToSwap)); double maxSizeOfBrokerToSwapWith = diskUsage(toSwap) * toSwapWith.broker().capacityFor(DISK); double currentSizeOfBrokerToSwapWith = brokerSize(toSwapWith); minSize = Math.max(minSize, (currentSizeOfBrokerToSwapWith + sizeToSwap) - maxSizeOfBrokerToSwapWith); } minSize += REPLICA_CONVERGENCE_DELTA; maxSize -= REPLICA_CONVERGENCE_DELTA; double targetSize = sizeToSwap + sizeToChange; if (LOG.isTraceEnabled()) { LOG.trace("replicaToSwap: {}(size={}), targetSize={}, minSize={}, maxSize={}", replicaToSwap, dWrap(replicaSize(replicaToSwap)), dWrap(targetSize), dWrap(minSize), dWrap(maxSize)); } Replica replicaToSwapWith = sortedReplicasToSwapWith.isEmpty() ? null : findReplicaToSwapWith(replicaToSwap, sortedReplicasToSwapWith, targetSize, minSize, maxSize, clusterModel); if (replicaToSwapWith != null) { if (LOG.isDebugEnabled()) { LOG.debug("Found replica to swap. Swapping {}({}) on broker {}({}) and {}({}) on broker {}({})", replicaToSwap.topicPartition(), dWrap(replicaSize(replicaToSwap)), toSwap.broker().id(), dWrap(brokerSize(toSwap)), replicaToSwapWith.topicPartition(), dWrap(replicaSize(replicaToSwapWith)), toSwapWith.broker().id(), dWrap(brokerSize(toSwapWith))); } clusterModel.relocateReplica(replicaToSwapWith.topicPartition(), toSwapWith.broker().id(), toSwap.broker().id()); clusterModel.relocateReplica(replicaToSwap.topicPartition(), toSwap.broker().id(), toSwapWith.broker().id()); toSwap.sortedReplicas().remove(replicaToSwap); toSwap.sortedReplicas().add(replicaToSwapWith); toSwapWith.sortedReplicas().remove(replicaToSwapWith); toSwapWith.sortedReplicas().add(replicaToSwap); return true; } } LOG.trace("Nothing to swap between broker {} and broker {}", toSwap.broker().id(), toSwapWith.broker().id()); return false; }
KafkaAssignerDiskUsageDistributionGoal implements Goal { boolean swapReplicas(BrokerAndSortedReplicas toSwap, BrokerAndSortedReplicas toSwapWith, double meanDiskUsage, ClusterModel clusterModel, Set<String> excludedTopics) { if (LOG.isTraceEnabled()) { LOG.trace("Swapping replicas between broker {}({}) and broker {}({})", toSwap.broker().id(), dWrap(brokerSize(toSwap)), toSwapWith.broker().id(), dWrap(brokerSize(toSwapWith))); } double sizeToChange = toSwap.broker().capacityFor(DISK) * meanDiskUsage - brokerSize(toSwap); NavigableSet<ReplicaWrapper> sortedReplicasToSwap = sortReplicasAscend(toSwap, excludedTopics); NavigableSet<ReplicaWrapper> sortedLeadersToSwapWith = sortReplicasAscend(toSwapWith, excludedTopics); NavigableSet<ReplicaWrapper> sortedFollowersToSwapWith = sortedFollowerReplicas(toSwapWith, excludedTopics); Iterator<ReplicaWrapper> toSwapIter = sizeToChange > 0 ? sortedReplicasToSwap.iterator() : sortedReplicasToSwap.descendingIterator(); while (toSwapIter.hasNext()) { Replica replicaToSwap = toSwapIter.next().replica(); if (excludedTopics.contains(replicaToSwap.topicPartition().topic())) { continue; } if (!possibleToMove(replicaToSwap, toSwapWith.broker(), clusterModel)) { continue; } NavigableSet<ReplicaWrapper> sortedReplicasToSwapWith = replicaToSwap.isLeader() ? sortedLeadersToSwapWith : sortedFollowersToSwapWith; double sizeToSwap = replicaSize(replicaToSwap); if (sizeToChange < 0 && sizeToSwap == 0) { break; } double maxSize = Double.MAX_VALUE; double minSize = Double.MIN_VALUE; if (sizeToChange > 0) { minSize = sizeToSwap; double maxSizeOfBrokerToSwap = diskUsage(toSwapWith) * toSwap.broker().capacityFor(DISK); double currentSizeOfBrokerToSwap = brokerSize(toSwap); maxSize = Math.min(maxSize, maxSizeOfBrokerToSwap - (currentSizeOfBrokerToSwap - sizeToSwap)); double minSizeOfBrokerToSwapWith = diskUsage(toSwap) * toSwapWith.broker().capacityFor(DISK); double currentSizeOfBrokerToSwapWith = brokerSize(toSwapWith); maxSize = Math.min(maxSize, (currentSizeOfBrokerToSwapWith + sizeToSwap) - minSizeOfBrokerToSwapWith); } else { maxSize = sizeToSwap; double minSizeOfBrokerToSwap = diskUsage(toSwapWith) * toSwap.broker().capacityFor(DISK); double currentSizeOfBrokerToSwap = brokerSize(toSwap); minSize = Math.max(minSize, minSizeOfBrokerToSwap - (currentSizeOfBrokerToSwap - sizeToSwap)); double maxSizeOfBrokerToSwapWith = diskUsage(toSwap) * toSwapWith.broker().capacityFor(DISK); double currentSizeOfBrokerToSwapWith = brokerSize(toSwapWith); minSize = Math.max(minSize, (currentSizeOfBrokerToSwapWith + sizeToSwap) - maxSizeOfBrokerToSwapWith); } minSize += REPLICA_CONVERGENCE_DELTA; maxSize -= REPLICA_CONVERGENCE_DELTA; double targetSize = sizeToSwap + sizeToChange; if (LOG.isTraceEnabled()) { LOG.trace("replicaToSwap: {}(size={}), targetSize={}, minSize={}, maxSize={}", replicaToSwap, dWrap(replicaSize(replicaToSwap)), dWrap(targetSize), dWrap(minSize), dWrap(maxSize)); } Replica replicaToSwapWith = sortedReplicasToSwapWith.isEmpty() ? null : findReplicaToSwapWith(replicaToSwap, sortedReplicasToSwapWith, targetSize, minSize, maxSize, clusterModel); if (replicaToSwapWith != null) { if (LOG.isDebugEnabled()) { LOG.debug("Found replica to swap. Swapping {}({}) on broker {}({}) and {}({}) on broker {}({})", replicaToSwap.topicPartition(), dWrap(replicaSize(replicaToSwap)), toSwap.broker().id(), dWrap(brokerSize(toSwap)), replicaToSwapWith.topicPartition(), dWrap(replicaSize(replicaToSwapWith)), toSwapWith.broker().id(), dWrap(brokerSize(toSwapWith))); } clusterModel.relocateReplica(replicaToSwapWith.topicPartition(), toSwapWith.broker().id(), toSwap.broker().id()); clusterModel.relocateReplica(replicaToSwap.topicPartition(), toSwap.broker().id(), toSwapWith.broker().id()); toSwap.sortedReplicas().remove(replicaToSwap); toSwap.sortedReplicas().add(replicaToSwapWith); toSwapWith.sortedReplicas().remove(replicaToSwapWith); toSwapWith.sortedReplicas().add(replicaToSwap); return true; } } LOG.trace("Nothing to swap between broker {} and broker {}", toSwap.broker().id(), toSwapWith.broker().id()); return false; } }
KafkaAssignerDiskUsageDistributionGoal implements Goal { boolean swapReplicas(BrokerAndSortedReplicas toSwap, BrokerAndSortedReplicas toSwapWith, double meanDiskUsage, ClusterModel clusterModel, Set<String> excludedTopics) { if (LOG.isTraceEnabled()) { LOG.trace("Swapping replicas between broker {}({}) and broker {}({})", toSwap.broker().id(), dWrap(brokerSize(toSwap)), toSwapWith.broker().id(), dWrap(brokerSize(toSwapWith))); } double sizeToChange = toSwap.broker().capacityFor(DISK) * meanDiskUsage - brokerSize(toSwap); NavigableSet<ReplicaWrapper> sortedReplicasToSwap = sortReplicasAscend(toSwap, excludedTopics); NavigableSet<ReplicaWrapper> sortedLeadersToSwapWith = sortReplicasAscend(toSwapWith, excludedTopics); NavigableSet<ReplicaWrapper> sortedFollowersToSwapWith = sortedFollowerReplicas(toSwapWith, excludedTopics); Iterator<ReplicaWrapper> toSwapIter = sizeToChange > 0 ? sortedReplicasToSwap.iterator() : sortedReplicasToSwap.descendingIterator(); while (toSwapIter.hasNext()) { Replica replicaToSwap = toSwapIter.next().replica(); if (excludedTopics.contains(replicaToSwap.topicPartition().topic())) { continue; } if (!possibleToMove(replicaToSwap, toSwapWith.broker(), clusterModel)) { continue; } NavigableSet<ReplicaWrapper> sortedReplicasToSwapWith = replicaToSwap.isLeader() ? sortedLeadersToSwapWith : sortedFollowersToSwapWith; double sizeToSwap = replicaSize(replicaToSwap); if (sizeToChange < 0 && sizeToSwap == 0) { break; } double maxSize = Double.MAX_VALUE; double minSize = Double.MIN_VALUE; if (sizeToChange > 0) { minSize = sizeToSwap; double maxSizeOfBrokerToSwap = diskUsage(toSwapWith) * toSwap.broker().capacityFor(DISK); double currentSizeOfBrokerToSwap = brokerSize(toSwap); maxSize = Math.min(maxSize, maxSizeOfBrokerToSwap - (currentSizeOfBrokerToSwap - sizeToSwap)); double minSizeOfBrokerToSwapWith = diskUsage(toSwap) * toSwapWith.broker().capacityFor(DISK); double currentSizeOfBrokerToSwapWith = brokerSize(toSwapWith); maxSize = Math.min(maxSize, (currentSizeOfBrokerToSwapWith + sizeToSwap) - minSizeOfBrokerToSwapWith); } else { maxSize = sizeToSwap; double minSizeOfBrokerToSwap = diskUsage(toSwapWith) * toSwap.broker().capacityFor(DISK); double currentSizeOfBrokerToSwap = brokerSize(toSwap); minSize = Math.max(minSize, minSizeOfBrokerToSwap - (currentSizeOfBrokerToSwap - sizeToSwap)); double maxSizeOfBrokerToSwapWith = diskUsage(toSwap) * toSwapWith.broker().capacityFor(DISK); double currentSizeOfBrokerToSwapWith = brokerSize(toSwapWith); minSize = Math.max(minSize, (currentSizeOfBrokerToSwapWith + sizeToSwap) - maxSizeOfBrokerToSwapWith); } minSize += REPLICA_CONVERGENCE_DELTA; maxSize -= REPLICA_CONVERGENCE_DELTA; double targetSize = sizeToSwap + sizeToChange; if (LOG.isTraceEnabled()) { LOG.trace("replicaToSwap: {}(size={}), targetSize={}, minSize={}, maxSize={}", replicaToSwap, dWrap(replicaSize(replicaToSwap)), dWrap(targetSize), dWrap(minSize), dWrap(maxSize)); } Replica replicaToSwapWith = sortedReplicasToSwapWith.isEmpty() ? null : findReplicaToSwapWith(replicaToSwap, sortedReplicasToSwapWith, targetSize, minSize, maxSize, clusterModel); if (replicaToSwapWith != null) { if (LOG.isDebugEnabled()) { LOG.debug("Found replica to swap. Swapping {}({}) on broker {}({}) and {}({}) on broker {}({})", replicaToSwap.topicPartition(), dWrap(replicaSize(replicaToSwap)), toSwap.broker().id(), dWrap(brokerSize(toSwap)), replicaToSwapWith.topicPartition(), dWrap(replicaSize(replicaToSwapWith)), toSwapWith.broker().id(), dWrap(brokerSize(toSwapWith))); } clusterModel.relocateReplica(replicaToSwapWith.topicPartition(), toSwapWith.broker().id(), toSwap.broker().id()); clusterModel.relocateReplica(replicaToSwap.topicPartition(), toSwap.broker().id(), toSwapWith.broker().id()); toSwap.sortedReplicas().remove(replicaToSwap); toSwap.sortedReplicas().add(replicaToSwapWith); toSwapWith.sortedReplicas().remove(replicaToSwapWith); toSwapWith.sortedReplicas().add(replicaToSwap); return true; } } LOG.trace("Nothing to swap between broker {} and broker {}", toSwap.broker().id(), toSwapWith.broker().id()); return false; } KafkaAssignerDiskUsageDistributionGoal();   KafkaAssignerDiskUsageDistributionGoal(BalancingConstraint constraint); }
KafkaAssignerDiskUsageDistributionGoal implements Goal { boolean swapReplicas(BrokerAndSortedReplicas toSwap, BrokerAndSortedReplicas toSwapWith, double meanDiskUsage, ClusterModel clusterModel, Set<String> excludedTopics) { if (LOG.isTraceEnabled()) { LOG.trace("Swapping replicas between broker {}({}) and broker {}({})", toSwap.broker().id(), dWrap(brokerSize(toSwap)), toSwapWith.broker().id(), dWrap(brokerSize(toSwapWith))); } double sizeToChange = toSwap.broker().capacityFor(DISK) * meanDiskUsage - brokerSize(toSwap); NavigableSet<ReplicaWrapper> sortedReplicasToSwap = sortReplicasAscend(toSwap, excludedTopics); NavigableSet<ReplicaWrapper> sortedLeadersToSwapWith = sortReplicasAscend(toSwapWith, excludedTopics); NavigableSet<ReplicaWrapper> sortedFollowersToSwapWith = sortedFollowerReplicas(toSwapWith, excludedTopics); Iterator<ReplicaWrapper> toSwapIter = sizeToChange > 0 ? sortedReplicasToSwap.iterator() : sortedReplicasToSwap.descendingIterator(); while (toSwapIter.hasNext()) { Replica replicaToSwap = toSwapIter.next().replica(); if (excludedTopics.contains(replicaToSwap.topicPartition().topic())) { continue; } if (!possibleToMove(replicaToSwap, toSwapWith.broker(), clusterModel)) { continue; } NavigableSet<ReplicaWrapper> sortedReplicasToSwapWith = replicaToSwap.isLeader() ? sortedLeadersToSwapWith : sortedFollowersToSwapWith; double sizeToSwap = replicaSize(replicaToSwap); if (sizeToChange < 0 && sizeToSwap == 0) { break; } double maxSize = Double.MAX_VALUE; double minSize = Double.MIN_VALUE; if (sizeToChange > 0) { minSize = sizeToSwap; double maxSizeOfBrokerToSwap = diskUsage(toSwapWith) * toSwap.broker().capacityFor(DISK); double currentSizeOfBrokerToSwap = brokerSize(toSwap); maxSize = Math.min(maxSize, maxSizeOfBrokerToSwap - (currentSizeOfBrokerToSwap - sizeToSwap)); double minSizeOfBrokerToSwapWith = diskUsage(toSwap) * toSwapWith.broker().capacityFor(DISK); double currentSizeOfBrokerToSwapWith = brokerSize(toSwapWith); maxSize = Math.min(maxSize, (currentSizeOfBrokerToSwapWith + sizeToSwap) - minSizeOfBrokerToSwapWith); } else { maxSize = sizeToSwap; double minSizeOfBrokerToSwap = diskUsage(toSwapWith) * toSwap.broker().capacityFor(DISK); double currentSizeOfBrokerToSwap = brokerSize(toSwap); minSize = Math.max(minSize, minSizeOfBrokerToSwap - (currentSizeOfBrokerToSwap - sizeToSwap)); double maxSizeOfBrokerToSwapWith = diskUsage(toSwap) * toSwapWith.broker().capacityFor(DISK); double currentSizeOfBrokerToSwapWith = brokerSize(toSwapWith); minSize = Math.max(minSize, (currentSizeOfBrokerToSwapWith + sizeToSwap) - maxSizeOfBrokerToSwapWith); } minSize += REPLICA_CONVERGENCE_DELTA; maxSize -= REPLICA_CONVERGENCE_DELTA; double targetSize = sizeToSwap + sizeToChange; if (LOG.isTraceEnabled()) { LOG.trace("replicaToSwap: {}(size={}), targetSize={}, minSize={}, maxSize={}", replicaToSwap, dWrap(replicaSize(replicaToSwap)), dWrap(targetSize), dWrap(minSize), dWrap(maxSize)); } Replica replicaToSwapWith = sortedReplicasToSwapWith.isEmpty() ? null : findReplicaToSwapWith(replicaToSwap, sortedReplicasToSwapWith, targetSize, minSize, maxSize, clusterModel); if (replicaToSwapWith != null) { if (LOG.isDebugEnabled()) { LOG.debug("Found replica to swap. Swapping {}({}) on broker {}({}) and {}({}) on broker {}({})", replicaToSwap.topicPartition(), dWrap(replicaSize(replicaToSwap)), toSwap.broker().id(), dWrap(brokerSize(toSwap)), replicaToSwapWith.topicPartition(), dWrap(replicaSize(replicaToSwapWith)), toSwapWith.broker().id(), dWrap(brokerSize(toSwapWith))); } clusterModel.relocateReplica(replicaToSwapWith.topicPartition(), toSwapWith.broker().id(), toSwap.broker().id()); clusterModel.relocateReplica(replicaToSwap.topicPartition(), toSwap.broker().id(), toSwapWith.broker().id()); toSwap.sortedReplicas().remove(replicaToSwap); toSwap.sortedReplicas().add(replicaToSwapWith); toSwapWith.sortedReplicas().remove(replicaToSwapWith); toSwapWith.sortedReplicas().add(replicaToSwap); return true; } } LOG.trace("Nothing to swap between broker {} and broker {}", toSwap.broker().id(), toSwapWith.broker().id()); return false; } KafkaAssignerDiskUsageDistributionGoal();   KafkaAssignerDiskUsageDistributionGoal(BalancingConstraint constraint); @Override ClusterModelStatsComparator clusterModelStatsComparator(); @Override void configure(Map<String, ?> configs); @Override ModelCompletenessRequirements clusterModelCompletenessRequirements(); @Override boolean optimize(ClusterModel clusterModel, Set<Goal> optimizedGoals, OptimizationOptions optimizationOptions); @Override ActionAcceptance actionAcceptance(BalancingAction action, ClusterModel clusterModel); @Override String name(); @Override void finish(); @Override boolean isHardGoal(); }
KafkaAssignerDiskUsageDistributionGoal implements Goal { boolean swapReplicas(BrokerAndSortedReplicas toSwap, BrokerAndSortedReplicas toSwapWith, double meanDiskUsage, ClusterModel clusterModel, Set<String> excludedTopics) { if (LOG.isTraceEnabled()) { LOG.trace("Swapping replicas between broker {}({}) and broker {}({})", toSwap.broker().id(), dWrap(brokerSize(toSwap)), toSwapWith.broker().id(), dWrap(brokerSize(toSwapWith))); } double sizeToChange = toSwap.broker().capacityFor(DISK) * meanDiskUsage - brokerSize(toSwap); NavigableSet<ReplicaWrapper> sortedReplicasToSwap = sortReplicasAscend(toSwap, excludedTopics); NavigableSet<ReplicaWrapper> sortedLeadersToSwapWith = sortReplicasAscend(toSwapWith, excludedTopics); NavigableSet<ReplicaWrapper> sortedFollowersToSwapWith = sortedFollowerReplicas(toSwapWith, excludedTopics); Iterator<ReplicaWrapper> toSwapIter = sizeToChange > 0 ? sortedReplicasToSwap.iterator() : sortedReplicasToSwap.descendingIterator(); while (toSwapIter.hasNext()) { Replica replicaToSwap = toSwapIter.next().replica(); if (excludedTopics.contains(replicaToSwap.topicPartition().topic())) { continue; } if (!possibleToMove(replicaToSwap, toSwapWith.broker(), clusterModel)) { continue; } NavigableSet<ReplicaWrapper> sortedReplicasToSwapWith = replicaToSwap.isLeader() ? sortedLeadersToSwapWith : sortedFollowersToSwapWith; double sizeToSwap = replicaSize(replicaToSwap); if (sizeToChange < 0 && sizeToSwap == 0) { break; } double maxSize = Double.MAX_VALUE; double minSize = Double.MIN_VALUE; if (sizeToChange > 0) { minSize = sizeToSwap; double maxSizeOfBrokerToSwap = diskUsage(toSwapWith) * toSwap.broker().capacityFor(DISK); double currentSizeOfBrokerToSwap = brokerSize(toSwap); maxSize = Math.min(maxSize, maxSizeOfBrokerToSwap - (currentSizeOfBrokerToSwap - sizeToSwap)); double minSizeOfBrokerToSwapWith = diskUsage(toSwap) * toSwapWith.broker().capacityFor(DISK); double currentSizeOfBrokerToSwapWith = brokerSize(toSwapWith); maxSize = Math.min(maxSize, (currentSizeOfBrokerToSwapWith + sizeToSwap) - minSizeOfBrokerToSwapWith); } else { maxSize = sizeToSwap; double minSizeOfBrokerToSwap = diskUsage(toSwapWith) * toSwap.broker().capacityFor(DISK); double currentSizeOfBrokerToSwap = brokerSize(toSwap); minSize = Math.max(minSize, minSizeOfBrokerToSwap - (currentSizeOfBrokerToSwap - sizeToSwap)); double maxSizeOfBrokerToSwapWith = diskUsage(toSwap) * toSwapWith.broker().capacityFor(DISK); double currentSizeOfBrokerToSwapWith = brokerSize(toSwapWith); minSize = Math.max(minSize, (currentSizeOfBrokerToSwapWith + sizeToSwap) - maxSizeOfBrokerToSwapWith); } minSize += REPLICA_CONVERGENCE_DELTA; maxSize -= REPLICA_CONVERGENCE_DELTA; double targetSize = sizeToSwap + sizeToChange; if (LOG.isTraceEnabled()) { LOG.trace("replicaToSwap: {}(size={}), targetSize={}, minSize={}, maxSize={}", replicaToSwap, dWrap(replicaSize(replicaToSwap)), dWrap(targetSize), dWrap(minSize), dWrap(maxSize)); } Replica replicaToSwapWith = sortedReplicasToSwapWith.isEmpty() ? null : findReplicaToSwapWith(replicaToSwap, sortedReplicasToSwapWith, targetSize, minSize, maxSize, clusterModel); if (replicaToSwapWith != null) { if (LOG.isDebugEnabled()) { LOG.debug("Found replica to swap. Swapping {}({}) on broker {}({}) and {}({}) on broker {}({})", replicaToSwap.topicPartition(), dWrap(replicaSize(replicaToSwap)), toSwap.broker().id(), dWrap(brokerSize(toSwap)), replicaToSwapWith.topicPartition(), dWrap(replicaSize(replicaToSwapWith)), toSwapWith.broker().id(), dWrap(brokerSize(toSwapWith))); } clusterModel.relocateReplica(replicaToSwapWith.topicPartition(), toSwapWith.broker().id(), toSwap.broker().id()); clusterModel.relocateReplica(replicaToSwap.topicPartition(), toSwap.broker().id(), toSwapWith.broker().id()); toSwap.sortedReplicas().remove(replicaToSwap); toSwap.sortedReplicas().add(replicaToSwapWith); toSwapWith.sortedReplicas().remove(replicaToSwapWith); toSwapWith.sortedReplicas().add(replicaToSwap); return true; } } LOG.trace("Nothing to swap between broker {} and broker {}", toSwap.broker().id(), toSwapWith.broker().id()); return false; } KafkaAssignerDiskUsageDistributionGoal();   KafkaAssignerDiskUsageDistributionGoal(BalancingConstraint constraint); @Override ClusterModelStatsComparator clusterModelStatsComparator(); @Override void configure(Map<String, ?> configs); @Override ModelCompletenessRequirements clusterModelCompletenessRequirements(); @Override boolean optimize(ClusterModel clusterModel, Set<Goal> optimizedGoals, OptimizationOptions optimizationOptions); @Override ActionAcceptance actionAcceptance(BalancingAction action, ClusterModel clusterModel); @Override String name(); @Override void finish(); @Override boolean isHardGoal();  }