private List<TabletSchedCtx> balanceBackendTablet(ClusterLoadStatistic clusterStat,
                                                      TStorageMedium medium) {
        List<TabletSchedCtx> alternativeTablets = Lists.newArrayList();
        for (BackendLoadStatistic beStat : getValidBeStats(clusterStat, medium)) {
            long beId = beStat.getBeId();
            if (!GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo().checkBackendAvailable(beId)) {
                continue;
            }

            List<RootPathLoadStatistic> pathStats = getValidBePathStats(beStat, medium);
            LOG.debug(
                    "get backend path stats for backend tablet distribution balance. medium: {}, be id: {}, path stats: {}",
                    medium, beId, pathStats);
            if (pathStats.size() <= 1) {
                continue;
            }

            balanceTablet(medium, alternativeTablets, true, null, null, pathStats, beId);
            if (alternativeTablets.size() >= Config.tablet_sched_max_balancing_tablets) {
                break;
            }
        }
        return alternativeTablets;
    }
@Test
    public void testBalanceBackendTablet(@Mocked GlobalStateMgr globalStateMgr) {
        // system info
        long dbId = 10001L;
        long tableId = 10002L;
        long partitionId1 = 10010L;
        long partitionId2 = 10011L;
        long indexId = 10003L;
        long tabletDataSize = 200 * 1024 * 1024L;
        long beId1 = 1L;
        long beId2 = 2L;
        long pathHash10 = 10L;
        long pathHash11 = 11L;
        long pathHash12 = 12L;
        long pathHash13 = 13L;
        long pathHash14 = 14L;
        long pathHash20 = 20L;
        long pathHash21 = 21L;

        Backend be1 = genBackend(beId1, "host1", 2 * tabletDataSize,
                2 * tabletDataSize, 4 * tabletDataSize, pathHash10);
        DiskInfo disk10 = genDiskInfo(2 * tabletDataSize, 2 * tabletDataSize,
                4 * tabletDataSize, "/data10", pathHash10, TStorageMedium.HDD);
        DiskInfo disk11 = genDiskInfo(3 * tabletDataSize, 1 * tabletDataSize,
                4 * tabletDataSize, "/data11", pathHash11, TStorageMedium.HDD);
        DiskInfo disk12 = genDiskInfo(4 * tabletDataSize, 0 * tabletDataSize,
                4 * tabletDataSize, "/data12", pathHash12, TStorageMedium.HDD);
        DiskInfo disk13 = genDiskInfo(2 * tabletDataSize, 2 * tabletDataSize,
                4 * tabletDataSize, "/data13", pathHash13, TStorageMedium.SSD);
        DiskInfo disk14 = genDiskInfo(4 * tabletDataSize, 0 * tabletDataSize,
                4 * tabletDataSize, "/data14", pathHash14, TStorageMedium.SSD);
        Map<String, DiskInfo> diskInfoMap1 = Maps.newHashMap();
        diskInfoMap1.put(disk10.getRootPath(), disk10);
        diskInfoMap1.put(disk11.getRootPath(), disk11);
        diskInfoMap1.put(disk12.getRootPath(), disk12);
        diskInfoMap1.put(disk13.getRootPath(), disk13);
        diskInfoMap1.put(disk14.getRootPath(), disk14);
        be1.setDisks(ImmutableMap.copyOf(diskInfoMap1));

        Backend be2 = genBackend(beId2, "host2", 6 * tabletDataSize,
                2 * tabletDataSize, 8 * tabletDataSize, pathHash20);
        DiskInfo disk20 = genDiskInfo(6 * tabletDataSize, 2 * tabletDataSize,
                8 * tabletDataSize, "/data20", pathHash20, TStorageMedium.HDD);
        DiskInfo disk21 = genDiskInfo(9 * tabletDataSize, 3 * tabletDataSize,
                12 * tabletDataSize, "/data21", pathHash21, TStorageMedium.SSD);
        Map<String, DiskInfo> diskInfoMap2 = Maps.newHashMap();
        diskInfoMap2.put(disk20.getRootPath(), disk20);
        diskInfoMap2.put(disk21.getRootPath(), disk21);
        be2.setDisks(ImmutableMap.copyOf(diskInfoMap2));

        SystemInfoService infoService = new SystemInfoService();
        infoService.addBackend(be1);
        infoService.addBackend(be2);

        // tablet inverted index
        TabletInvertedIndex invertedIndex = new TabletInvertedIndex();
        MaterializedIndex materializedIndex = new MaterializedIndex(indexId, IndexState.NORMAL);
        addTablet(invertedIndex, materializedIndex, TStorageMedium.HDD, dbId, tableId, partitionId1, indexId,
                20001L, 30001L, beId1, tabletDataSize, pathHash10);
        addTablet(invertedIndex, materializedIndex, TStorageMedium.HDD, dbId, tableId, partitionId1, indexId,
                20002L, 30002L, beId1, tabletDataSize, pathHash10);
        addTablet(invertedIndex, materializedIndex, TStorageMedium.HDD, dbId, tableId, partitionId1, indexId,
                20003L, 30003L, beId1, tabletDataSize, pathHash11);
        addTablet(invertedIndex, materializedIndex, TStorageMedium.HDD, dbId, tableId, partitionId1, indexId,
                20004L, 30004L, beId2, tabletDataSize, pathHash20);
        addTablet(invertedIndex, materializedIndex, TStorageMedium.HDD, dbId, tableId, partitionId1, indexId,
                20005L, 30005L, beId2, tabletDataSize, pathHash20);

        addTablet(invertedIndex, materializedIndex, TStorageMedium.SSD, dbId, tableId, partitionId2, indexId,
                20006L, 30006L, beId1, tabletDataSize, pathHash13);
        addTablet(invertedIndex, materializedIndex, TStorageMedium.SSD, dbId, tableId, partitionId2, indexId,
                20007L, 30007L, beId1, tabletDataSize, pathHash13);
        addTablet(invertedIndex, materializedIndex, TStorageMedium.SSD, dbId, tableId, partitionId2, indexId,
                20008L, 30008L, beId2, tabletDataSize, pathHash21);
        addTablet(invertedIndex, materializedIndex, TStorageMedium.SSD, dbId, tableId, partitionId2, indexId,
                20009L, 30009L, beId2, tabletDataSize, pathHash21);
        addTablet(invertedIndex, materializedIndex, TStorageMedium.SSD, dbId, tableId, partitionId2, indexId,
                20010L, 30010L, beId2, tabletDataSize, pathHash21);

        ClusterLoadStatistic clusterLoadStatistic = new ClusterLoadStatistic(infoService, invertedIndex);
        clusterLoadStatistic.init();

        PartitionInfo partitionInfo = new PartitionInfo();
        DataProperty dataProperty1 = new DataProperty(TStorageMedium.HDD);
        partitionInfo.addPartition(partitionId1, dataProperty1, (short) 1, false);
        DataProperty dataProperty2 = new DataProperty(TStorageMedium.SSD);
        partitionInfo.addPartition(partitionId2, dataProperty2, (short) 1, false);
        DistributionInfo distributionInfo = new HashDistributionInfo(6, Lists.newArrayList());
        Partition partition1 = new Partition(partitionId1, "partition1", materializedIndex, distributionInfo);
        Partition partition2 = new Partition(partitionId2, "partition2", materializedIndex, distributionInfo);
        OlapTable table = new OlapTable(tableId, "table", Lists.newArrayList(), KeysType.AGG_KEYS, partitionInfo,
                distributionInfo);
        table.addPartition(partition1);
        table.addPartition(partition2);
        Database database = new Database(dbId, "database");
        database.registerTableUnlocked(table);

        new Expectations() {
            {
                GlobalStateMgr.getCurrentState();
                result = globalStateMgr;
                minTimes = 0;

                GlobalStateMgr.getCurrentState().getLocalMetastore().getDbIdsIncludeRecycleBin();
                result = Lists.newArrayList(dbId);
                minTimes = 0;

                GlobalStateMgr.getCurrentState().getLocalMetastore().getDbIncludeRecycleBin(dbId);
                result = database;
                minTimes = 0;

                GlobalStateMgr.getCurrentState().getLocalMetastore().getTableIncludeRecycleBin((Database) any, anyLong);
                result = table;
                minTimes = 0;

                GlobalStateMgr.getCurrentState().getLocalMetastore().getTablesIncludeRecycleBin((Database) any);
                result = Lists.newArrayList(table);
                minTimes = 0;

                GlobalStateMgr.getCurrentState().getLocalMetastore().getPartitionIncludeRecycleBin((OlapTable) any, partitionId1);
                result = partition1;
                minTimes = 0;

                GlobalStateMgr.getCurrentState().getLocalMetastore().getPartitionIncludeRecycleBin((OlapTable) any, partitionId2);
                result = partition2;
                minTimes = 0;

                GlobalStateMgr.getCurrentState().getLocalMetastore().getAllPartitionsIncludeRecycleBin((OlapTable) any);
                result = Lists.newArrayList(partition1, partition2);
                minTimes = 0;

                GlobalStateMgr.getCurrentState().getLocalMetastore()
                        .getReplicationNumIncludeRecycleBin((PartitionInfo) any, anyLong);
                result = (short) 1;
                minTimes = 0;

                GlobalStateMgr.getCurrentState().getLocalMetastore()
                        .getDataPropertyIncludeRecycleBin((PartitionInfo) any, partitionId1);
                result = dataProperty1;
                minTimes = 0;

                GlobalStateMgr.getCurrentState().getLocalMetastore()
                        .getDataPropertyIncludeRecycleBin((PartitionInfo) any, partitionId2);
                result = dataProperty2;
                minTimes = 0;

                GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo();
                result = infoService;
                minTimes = 0;

                GlobalStateMgr.getCurrentState().getTabletInvertedIndex();
                result = invertedIndex;
                minTimes = 0;
            }
        };

        Rebalancer rebalancer = new DiskAndTabletLoadReBalancer();
        rebalancer.updateLoadStatistic(clusterLoadStatistic);

        // set Config.balance_load_disk_safe_threshold to 0.4 to trigger backend disk balance
        Config.tablet_sched_balance_load_disk_safe_threshold = 0.4;
        Config.storage_usage_soft_limit_reserve_bytes = 1;
        List<TabletSchedCtx> tablets = rebalancer.selectAlternativeTablets();
        Assert.assertEquals(2, tablets.size());
        Assert.assertTrue(tablets.stream().allMatch(t -> (t.getDestBackendId() == beId1)));
        Assert.assertTrue(tablets.stream().allMatch(t -> (t.getSrcBackendId() == beId1)));
        Assert.assertTrue(tablets.stream().anyMatch(t -> (t.getDestPathHash() == pathHash12)));
        Assert.assertTrue(tablets.stream().anyMatch(t -> (t.getDestPathHash() == pathHash14)));
        Assert.assertTrue(tablets.stream().anyMatch(t -> (t.getSrcPathHash() == pathHash10)));
        Assert.assertTrue(tablets.stream().anyMatch(t -> (t.getSrcPathHash() == pathHash13)));

        // set Config.balance_load_disk_safe_threshold to 0.9 to trigger backend tablet distribution balance
        Config.tablet_sched_balance_load_disk_safe_threshold = 0.9;
        tablets = rebalancer.selectAlternativeTablets();
        Assert.assertEquals(2, tablets.size());
        Assert.assertTrue(tablets.stream().allMatch(t -> (t.getDestBackendId() == beId1)));
        Assert.assertTrue(tablets.stream().allMatch(t -> (t.getSrcBackendId() == beId1)));
        Assert.assertTrue(tablets.stream().anyMatch(t -> (t.getDestPathHash() == pathHash12)));
        Assert.assertTrue(tablets.stream().anyMatch(t -> (t.getDestPathHash() == pathHash14)));
        Assert.assertTrue(tablets.stream().anyMatch(t -> (t.getSrcPathHash() == pathHash10)));
        Assert.assertTrue(tablets.stream().anyMatch(t -> (t.getSrcPathHash() == pathHash13)));

        // set table state to schema_change, balance should be ignored
        table.setState(OlapTable.OlapTableState.SCHEMA_CHANGE);
        Assert.assertEquals(0, rebalancer.selectAlternativeTablets().size());
    }