Replica findReplicaToSwapWith(Replica replica,
                                NavigableSet<ReplicaWrapper> sortedReplicasToSearch,
                                double targetSize,
                                double minSize,
                                double maxSize,
                                ClusterModel clusterModel) {
    if (minSize > maxSize) {
      return null;
    }

    // Within the candidate replicas, find the replicas whose size falls into [minSize, maxSize], not inclusive.
    NavigableSet<ReplicaWrapper> candidates = sortedReplicasToSearch.subSet(ReplicaWrapper.greaterThan(minSize),
                                                                            false,
                                                                            ReplicaWrapper.lessThan(maxSize),
                                                                            false);
    // No candidate available, just return null.
    if (candidates.isEmpty()) {
      return null;
    }
    // The iterator for replicas from [targetSize (inclusive), maxSize (exclusive)]
    Iterator<ReplicaWrapper> ascendingLargerIter = null;
    // The iterator for replicas from [minSize (exclusive), targetSize (inclusive)]
    Iterator<ReplicaWrapper> descendingLessIter = null;
    // Check if the target size falls in the range or not. This is needed to avoid passing invalid targetSize to
    // the tailSet() or headSet().
    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();
    }

    // Advance the ascending and descending iterator in the ascending order of their distance from target size,
    // return the first replica that can swap. Otherwise return null.
    ReplicaWrapper low = null;
    ReplicaWrapper high = null;
    ReplicaWrapper candidateReplica = null;
    while (true) {
      // The last checked replica is from the high end, advance the ascending iterator.
      if (candidateReplica == high) {
        high = ascendingLargerIter != null && ascendingLargerIter.hasNext() ? ascendingLargerIter.next() : null;
      }
      // The last checked replica is from the low end, advance the descending iterator.
      if (candidateReplica == low) {
        low = descendingLessIter != null && descendingLessIter.hasNext() ? descendingLessIter.next() : null;
      }

      // No more replicas to check, give up.
      if (high == null && low == null) {
        return null;
      } else if (high == null) {
        // Use the lower end
        candidateReplica = low;
      } else if (low == null) {
        // Use the higher end
        candidateReplica = high;
      } else {
        // pick a replica closer to the target.
        double lowDiff = targetSize - low.size();
        double highDiff = high.size() - targetSize;
        candidateReplica = lowDiff <= highDiff ? low : high;
      }
      if (canSwap(replica, candidateReplica.replica(), clusterModel)) {
        return candidateReplica.replica();
      }
    }
  }
@Test
  public void testFindReplicaToSwapWith() {
    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();

    Broker b2 = clusterModel.broker(2);
    Replica r = b2.replica(T0P1);
    assertNull(goal.findReplicaToSwapWith(r, sortedReplicaAscend(clusterModel.broker(1)),
                                          30, 10, 90, clusterModel));

    // The replicas on broker 3 are of the following sizes
    // T0P0: 10
    // T0P2: 20
    // T1P1: 30
    // T2P2: 50
    // T2P1: 60
    // T1P0: 80
    // Only T0P0 and T1P1 are eligible to swap with r.
    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);

  }