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);

    // Depending on whether we need more or less disk usage, using ascending or descending iterator.
    Iterator<ReplicaWrapper> toSwapIter =
        sizeToChange > 0 ? sortedReplicasToSwap.iterator() : sortedReplicasToSwap.descendingIterator();

    while (toSwapIter.hasNext()) {
      Replica replicaToSwap = toSwapIter.next().replica();
      if (excludedTopics.contains(replicaToSwap.topicPartition().topic())) {
        continue;
      }
      // First make sure the replica is possible to be moved to the broker toSwapWith. If this check fails,
      // don't bother to search for replica to swap with.
      if (!possibleToMove(replicaToSwap, toSwapWith.broker(), clusterModel)) {
        continue;
      }
      NavigableSet<ReplicaWrapper> sortedReplicasToSwapWith =
          replicaToSwap.isLeader() ? sortedLeadersToSwapWith : sortedFollowersToSwapWith;
      double sizeToSwap = replicaSize(replicaToSwap);
      // No need to continue if we are trying to reduce the size and the replicas to swap out is of size 0.
      if (sizeToChange < 0 && sizeToSwap == 0) {
        break;
      }
      // when sizeToChange > 0, the broker toSwap needs more disk utilization, the replicaToSwapWith should meet the
      // following requirements:
      // 1. replicaToSwapWith.size() > replicaToSwap.size()
      // 2. After the swap, the disk usage of broker toSwap should not be more than the disk usage of broker
      //    toSwapWith before the swap.
      // 3. After the swap, the disk usage of broker toSwapWith should not be less than the disk usage of broker
      //    toSwap before the swap.
      //
      // When sizeToChange < 0, the broker toSwap needs less disk utilization, the replicaToSwapWith should meet the
      // following requirements:
      // 4. replicaToSwapWith.size < replicaToSwap.size()
      // 5. After the swap, the disk usage of broker toSwap should not be less than the disk usage of broker
      //    toSwapWith before the swap.
      // 6. After the swap, the disk usage of broker toSwapWith should not be more than the disk usage of broker
      //    toSwap before the swap.
      //
      // We do not require the swap to be under the balance upper limit or lower limit. Instead, we just ensure
      // that after the swap, the two replicas are closer to the mean usage.
      double maxSize = Double.MAX_VALUE;
      double minSize = Double.MIN_VALUE;
      if (sizeToChange > 0) {
        // requirement 1
        minSize = sizeToSwap;
        // requirement 2
        double maxSizeOfBrokerToSwap = diskUsage(toSwapWith) * toSwap.broker().capacityFor(DISK);
        double currentSizeOfBrokerToSwap = brokerSize(toSwap);
        // after given out the sizeToSwap, the maximum size the broker toSwap can take in.
        maxSize = Math.min(maxSize, maxSizeOfBrokerToSwap - (currentSizeOfBrokerToSwap - sizeToSwap));
        // requirement 3
        double minSizeOfBrokerToSwapWith = diskUsage(toSwap) * toSwapWith.broker().capacityFor(DISK);
        double currentSizeOfBrokerToSwapWith = brokerSize(toSwapWith);
        // after take in the sizeToSwap, the maximum size the broker toSwapWith can give out.
        maxSize = Math.min(maxSize, (currentSizeOfBrokerToSwapWith + sizeToSwap) - minSizeOfBrokerToSwapWith);
      } else {
        // requirement 4
        maxSize = sizeToSwap;
        // requirement 5
        double minSizeOfBrokerToSwap = diskUsage(toSwapWith) * toSwap.broker().capacityFor(DISK);
        double currentSizeOfBrokerToSwap = brokerSize(toSwap);
        // After give out the sizeToSwap, the minimum size the broker toSwap should take in.
        minSize = Math.max(minSize, minSizeOfBrokerToSwap - (currentSizeOfBrokerToSwap - sizeToSwap));
        // requirement 6
        double maxSizeOfBrokerToSwapWith = diskUsage(toSwap) * toSwapWith.broker().capacityFor(DISK);
        double currentSizeOfBrokerToSwapWith = brokerSize(toSwapWith);
        // after take in the sizeToSwap, the minimum size the broker toSwapWith should give out.
        minSize = Math.max(minSize, (currentSizeOfBrokerToSwapWith + sizeToSwap) - maxSizeOfBrokerToSwapWith);
      }

      // Add the delta to the min and max size.
      minSize += REPLICA_CONVERGENCE_DELTA;
      maxSize -= REPLICA_CONVERGENCE_DELTA;

      // The target size might be negative here. It would still work for our binary search purpose.
      double targetSize = sizeToSwap + sizeToChange;

      // Find a replica that is eligible for swap.
      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;
  }
@Test
  public void testSwapReplicas() {
    Properties props = KafkaCruiseControlUnitTestUtils.getKafkaCruiseControlProperties();
    props.setProperty(AnalyzerConfig.MAX_REPLICAS_PER_BROKER_CONFIG, Long.toString(10L));
    props.setProperty(AnalyzerConfig.OVERPROVISIONED_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()));
  }