@Test public void testFindReplicaToSwapWith() { 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(); Broker b2 = clusterModel.broker(2); Replica r = b2.replica(T0P1); assertNull(goal.findReplicaToSwapWith(r, sortedReplicaAscend(clusterModel.broker(1)), 30, 10, 90, clusterModel)); findReplicaToSwapWithAndVerify(Arrays.asList(-1.0, 5.0, 10.0, 20.0, 21.0, 60.0, 100.0), Arrays.asList(T0P0, T0P0, T0P0, T0P0, T1P1, T1P1, T1P1), 9, 90, r, 3, clusterModel, goal); findReplicaToSwapWithAndVerify(Arrays.asList(-1.0, 5.0, 10.0, 20.0, 21.0, 60.0, 100.0), Arrays.asList(T1P1, T1P1, T1P1, T1P1, T1P1, T1P1, T1P1), 10, 31, r, 3, clusterModel, goal); findReplicaToSwapWithAndVerify(Arrays.asList(-1.0, 5.0, 10.0, 20.0, 21.0, 60.0, 100.0), Arrays.asList(T0P0, T0P0, T0P0, T0P0, T0P0, T0P0, T0P0), 9, 30, r, 3, clusterModel, goal); findReplicaToSwapWithAndVerify(Arrays.asList(-1.0, 5.0, 10.0, 20.0, 21.0, 60.0, 100.0), Arrays.asList(null, null, null, null, null, null, null), 10, 30, r, 3, clusterModel, goal); }
Replica findReplicaToSwapWith(Replica replica, NavigableSet<ReplicaWrapper> sortedReplicasToSearch, double targetSize, double minSize, double maxSize, ClusterModel clusterModel) { if (minSize > maxSize) { return null; } NavigableSet<ReplicaWrapper> candidates = sortedReplicasToSearch.subSet(ReplicaWrapper.greaterThan(minSize), false, ReplicaWrapper.lessThan(maxSize), false); if (candidates.isEmpty()) { return null; } Iterator<ReplicaWrapper> ascendingLargerIter = null; Iterator<ReplicaWrapper> descendingLessIter = null; if (targetSize <= minSize) { ascendingLargerIter = candidates.iterator(); } else if (targetSize >= maxSize) { descendingLessIter = candidates.descendingIterator(); } else { ascendingLargerIter = candidates.tailSet(ReplicaWrapper.greaterThanOrEqualsTo(targetSize), true).iterator(); descendingLessIter = candidates.headSet(ReplicaWrapper.lessThanOrEqualsTo(targetSize), true).descendingIterator(); } ReplicaWrapper low = null; ReplicaWrapper high = null; ReplicaWrapper candidateReplica = null; while (true) { if (candidateReplica == high) { high = ascendingLargerIter != null && ascendingLargerIter.hasNext() ? ascendingLargerIter.next() : null; } if (candidateReplica == low) { low = descendingLessIter != null && descendingLessIter.hasNext() ? descendingLessIter.next() : null; } if (high == null && low == null) { return null; } else if (high == null) { candidateReplica = low; } else if (low == null) { candidateReplica = high; } else { double lowDiff = targetSize - low.size(); double highDiff = high.size() - targetSize; candidateReplica = lowDiff <= highDiff ? low : high; } if (canSwap(replica, candidateReplica.replica(), clusterModel)) { return candidateReplica.replica(); } } }
KafkaAssignerDiskUsageDistributionGoal implements Goal { Replica findReplicaToSwapWith(Replica replica, NavigableSet<ReplicaWrapper> sortedReplicasToSearch, double targetSize, double minSize, double maxSize, ClusterModel clusterModel) { if (minSize > maxSize) { return null; } NavigableSet<ReplicaWrapper> candidates = sortedReplicasToSearch.subSet(ReplicaWrapper.greaterThan(minSize), false, ReplicaWrapper.lessThan(maxSize), false); if (candidates.isEmpty()) { return null; } Iterator<ReplicaWrapper> ascendingLargerIter = null; Iterator<ReplicaWrapper> descendingLessIter = null; if (targetSize <= minSize) { ascendingLargerIter = candidates.iterator(); } else if (targetSize >= maxSize) { descendingLessIter = candidates.descendingIterator(); } else { ascendingLargerIter = candidates.tailSet(ReplicaWrapper.greaterThanOrEqualsTo(targetSize), true).iterator(); descendingLessIter = candidates.headSet(ReplicaWrapper.lessThanOrEqualsTo(targetSize), true).descendingIterator(); } ReplicaWrapper low = null; ReplicaWrapper high = null; ReplicaWrapper candidateReplica = null; while (true) { if (candidateReplica == high) { high = ascendingLargerIter != null && ascendingLargerIter.hasNext() ? ascendingLargerIter.next() : null; } if (candidateReplica == low) { low = descendingLessIter != null && descendingLessIter.hasNext() ? descendingLessIter.next() : null; } if (high == null && low == null) { return null; } else if (high == null) { candidateReplica = low; } else if (low == null) { candidateReplica = high; } else { double lowDiff = targetSize - low.size(); double highDiff = high.size() - targetSize; candidateReplica = lowDiff <= highDiff ? low : high; } if (canSwap(replica, candidateReplica.replica(), clusterModel)) { return candidateReplica.replica(); } } } }
KafkaAssignerDiskUsageDistributionGoal implements Goal { Replica findReplicaToSwapWith(Replica replica, NavigableSet<ReplicaWrapper> sortedReplicasToSearch, double targetSize, double minSize, double maxSize, ClusterModel clusterModel) { if (minSize > maxSize) { return null; } NavigableSet<ReplicaWrapper> candidates = sortedReplicasToSearch.subSet(ReplicaWrapper.greaterThan(minSize), false, ReplicaWrapper.lessThan(maxSize), false); if (candidates.isEmpty()) { return null; } Iterator<ReplicaWrapper> ascendingLargerIter = null; Iterator<ReplicaWrapper> descendingLessIter = null; if (targetSize <= minSize) { ascendingLargerIter = candidates.iterator(); } else if (targetSize >= maxSize) { descendingLessIter = candidates.descendingIterator(); } else { ascendingLargerIter = candidates.tailSet(ReplicaWrapper.greaterThanOrEqualsTo(targetSize), true).iterator(); descendingLessIter = candidates.headSet(ReplicaWrapper.lessThanOrEqualsTo(targetSize), true).descendingIterator(); } ReplicaWrapper low = null; ReplicaWrapper high = null; ReplicaWrapper candidateReplica = null; while (true) { if (candidateReplica == high) { high = ascendingLargerIter != null && ascendingLargerIter.hasNext() ? ascendingLargerIter.next() : null; } if (candidateReplica == low) { low = descendingLessIter != null && descendingLessIter.hasNext() ? descendingLessIter.next() : null; } if (high == null && low == null) { return null; } else if (high == null) { candidateReplica = low; } else if (low == null) { candidateReplica = high; } else { double lowDiff = targetSize - low.size(); double highDiff = high.size() - targetSize; candidateReplica = lowDiff <= highDiff ? low : high; } if (canSwap(replica, candidateReplica.replica(), clusterModel)) { return candidateReplica.replica(); } } } KafkaAssignerDiskUsageDistributionGoal();   KafkaAssignerDiskUsageDistributionGoal(BalancingConstraint constraint); }
KafkaAssignerDiskUsageDistributionGoal implements Goal { Replica findReplicaToSwapWith(Replica replica, NavigableSet<ReplicaWrapper> sortedReplicasToSearch, double targetSize, double minSize, double maxSize, ClusterModel clusterModel) { if (minSize > maxSize) { return null; } NavigableSet<ReplicaWrapper> candidates = sortedReplicasToSearch.subSet(ReplicaWrapper.greaterThan(minSize), false, ReplicaWrapper.lessThan(maxSize), false); if (candidates.isEmpty()) { return null; } Iterator<ReplicaWrapper> ascendingLargerIter = null; Iterator<ReplicaWrapper> descendingLessIter = null; if (targetSize <= minSize) { ascendingLargerIter = candidates.iterator(); } else if (targetSize >= maxSize) { descendingLessIter = candidates.descendingIterator(); } else { ascendingLargerIter = candidates.tailSet(ReplicaWrapper.greaterThanOrEqualsTo(targetSize), true).iterator(); descendingLessIter = candidates.headSet(ReplicaWrapper.lessThanOrEqualsTo(targetSize), true).descendingIterator(); } ReplicaWrapper low = null; ReplicaWrapper high = null; ReplicaWrapper candidateReplica = null; while (true) { if (candidateReplica == high) { high = ascendingLargerIter != null && ascendingLargerIter.hasNext() ? ascendingLargerIter.next() : null; } if (candidateReplica == low) { low = descendingLessIter != null && descendingLessIter.hasNext() ? descendingLessIter.next() : null; } if (high == null && low == null) { return null; } else if (high == null) { candidateReplica = low; } else if (low == null) { candidateReplica = high; } else { double lowDiff = targetSize - low.size(); double highDiff = high.size() - targetSize; candidateReplica = lowDiff <= highDiff ? low : high; } if (canSwap(replica, candidateReplica.replica(), clusterModel)) { return candidateReplica.replica(); } } } 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 { Replica findReplicaToSwapWith(Replica replica, NavigableSet<ReplicaWrapper> sortedReplicasToSearch, double targetSize, double minSize, double maxSize, ClusterModel clusterModel) { if (minSize > maxSize) { return null; } NavigableSet<ReplicaWrapper> candidates = sortedReplicasToSearch.subSet(ReplicaWrapper.greaterThan(minSize), false, ReplicaWrapper.lessThan(maxSize), false); if (candidates.isEmpty()) { return null; } Iterator<ReplicaWrapper> ascendingLargerIter = null; Iterator<ReplicaWrapper> descendingLessIter = null; if (targetSize <= minSize) { ascendingLargerIter = candidates.iterator(); } else if (targetSize >= maxSize) { descendingLessIter = candidates.descendingIterator(); } else { ascendingLargerIter = candidates.tailSet(ReplicaWrapper.greaterThanOrEqualsTo(targetSize), true).iterator(); descendingLessIter = candidates.headSet(ReplicaWrapper.lessThanOrEqualsTo(targetSize), true).descendingIterator(); } ReplicaWrapper low = null; ReplicaWrapper high = null; ReplicaWrapper candidateReplica = null; while (true) { if (candidateReplica == high) { high = ascendingLargerIter != null && ascendingLargerIter.hasNext() ? ascendingLargerIter.next() : null; } if (candidateReplica == low) { low = descendingLessIter != null && descendingLessIter.hasNext() ? descendingLessIter.next() : null; } if (high == null && low == null) { return null; } else if (high == null) { candidateReplica = low; } else if (low == null) { candidateReplica = high; } else { double lowDiff = targetSize - low.size(); double highDiff = high.size() - targetSize; candidateReplica = lowDiff <= highDiff ? low : high; } if (canSwap(replica, candidateReplica.replica(), clusterModel)) { return candidateReplica.replica(); } } } 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();  }