private List<Map.Entry<Long, Long>> getSortedBackendReplicaNumPairs(List<Long> allAvailBackendIds,
                                                                        Set<Long> unavailBackendIds,
                                                                        ClusterLoadStatistic statistic,
                                                                        List<Long> flatBackendsPerBucketSeq) {
        // backend id -> replica num, and sorted by replica num, descending.
        Map<Long, Long> backendToReplicaNum = flatBackendsPerBucketSeq.stream()
                .collect(Collectors.groupingBy(Function.identity(), Collectors.counting()));
        // remove unavailable backend
        for (Long backendId : unavailBackendIds) {
            backendToReplicaNum.remove(backendId);
        }
        // add backends which are not in flatBackendsPerBucketSeq, with replication number 0
        for (Long backendId : allAvailBackendIds) {
            if (!backendToReplicaNum.containsKey(backendId)) {
                backendToReplicaNum.put(backendId, 0L);
            }
        }

        List<Map.Entry<Long, Long>> entries = new ArrayList<>(backendToReplicaNum.entrySet());
        if (!FeConstants.runningUnitTest) {
            // to randomize the relative order of entries with the same number of replicas
            Collections.shuffle(entries);
        }
        entries.sort((entry1, entry2) -> {
            if (!entry1.getValue().equals(entry2.getValue())) {
                return (int) (entry2.getValue() - entry1.getValue());
            }
            BackendLoadStatistic beStat1 = statistic.getBackendLoadStatistic(entry1.getKey());
            BackendLoadStatistic beStat2 = statistic.getBackendLoadStatistic(entry2.getKey());
            if (beStat1 == null || beStat2 == null) {
                return 0;
            }
            double loadScore1 = beStat1.getMixLoadScore();
            double loadScore2 = beStat2.getMixLoadScore();
            if (Math.abs(loadScore1 - loadScore2) < 1e-6) {
                return 0;
            } else if (loadScore2 > loadScore1) {
                return 1;
            } else {
                return -1;
            }
        });

        return entries;
    }
@Test
    public void testGetSortedBackendReplicaNumPairs(@Mocked ClusterLoadStatistic statistic) {
        new Expectations() {
            {
                statistic.getBackendLoadStatistic(anyLong);
                result = new Delegate<BackendLoadStatistic>() {
                    BackendLoadStatistic delegate(Long beId) {
                        return new FakeBackendLoadStatistic(beId, null, null, null);
                    }
                };
                minTimes = 0;
            }
        };

        GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo().getIdToBackend();
        // all buckets are on different be
        List<Long> allAvailBackendIds = Lists.newArrayList(1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L);
        Set<Long> unavailBackendIds = Sets.newHashSet(9L);
        List<Long> flatBackendsPerBucketSeq = Lists.newArrayList(1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L, 9L);
        List<Map.Entry<Long, Long>> backends = Deencapsulation.invoke(balancer, "getSortedBackendReplicaNumPairs",
                allAvailBackendIds, unavailBackendIds, statistic, flatBackendsPerBucketSeq);
        long[] backendIds = backends.stream().mapToLong(Map.Entry::getKey).toArray();
        Assert.assertArrayEquals(new long[] {7L, 8L, 6L, 2L, 3L, 5L, 4L, 1L}, backendIds);

        // 0,1 bucket on same be and 5, 6 on same be
        flatBackendsPerBucketSeq = Lists.newArrayList(1L, 1L, 3L, 4L, 5L, 6L, 7L, 7L, 9L);
        backends = Deencapsulation
                .invoke(balancer, "getSortedBackendReplicaNumPairs", allAvailBackendIds, unavailBackendIds,
                        statistic, flatBackendsPerBucketSeq);
        backendIds = backends.stream().mapToLong(Map.Entry::getKey).toArray();
        Assert.assertArrayEquals(new long[] {7L, 1L, 6L, 3L, 5L, 4L, 8L, 2L}, backendIds);
    }