protected static void handleMigration(ListMultimap<TStorageMedium, Long> tabletMetaMigrationMap,
                                          long backendId) {
        TabletInvertedIndex invertedIndex = GlobalStateMgr.getCurrentState().getTabletInvertedIndex();
        AgentBatchTask batchTask = new AgentBatchTask();

        OUTER:
        for (TStorageMedium storageMedium : tabletMetaMigrationMap.keySet()) {
            List<Long> tabletIds = tabletMetaMigrationMap.get(storageMedium);
            List<TabletMeta> tabletMetaList = invertedIndex.getTabletMetaList(tabletIds);
            for (int i = 0; i < tabletMetaList.size(); i++) {
                long tabletId = tabletIds.get(i);
                TabletMeta tabletMeta = tabletMetaList.get(i);

                // 1. If size of tabletMigrationMap exceeds (Config.tablet_sched_max_migration_task_sent_once - running_tasks_on_be),
                // dot not send more tasks. The number of tasks running on BE cannot exceed Config.tablet_sched_max_migration_task_sent_once
                if (batchTask.getTaskNum() >=
                        Config.tablet_sched_max_migration_task_sent_once
                                - AgentTaskQueue.getTaskNum(backendId, TTaskType.STORAGE_MEDIUM_MIGRATE, false)) {
                    LOG.debug("size of tabletMigrationMap + size of running tasks on BE is bigger than {}",
                            Config.tablet_sched_max_migration_task_sent_once);
                    break OUTER;
                }

                // 2. If the task already running on BE, do not send again
                if (AgentTaskQueue.getTask(backendId, TTaskType.STORAGE_MEDIUM_MIGRATE, tabletId) != null) {
                    LOG.debug("migrate of tablet:{} is already running on BE", tabletId);
                    continue;
                }

                // 3. There are some limitations for primary table, details in migratableTablet()
                Database db = GlobalStateMgr.getCurrentState().getDb(tabletMeta.getDbId());
                if (db == null) {
                    continue;
                }
                OlapTable table = (OlapTable) db.getTable(tabletMeta.getTableId());
                if (table == null) {
                    continue;
                }

                if (!migratableTablet(db, table, tabletMeta.getPhysicalPartitionId(), tabletMeta.getIndexId(), tabletId)) {
                    continue;
                }

                // always get old schema hash(as effective one)
                int effectiveSchemaHash = tabletMeta.getOldSchemaHash();
                StorageMediaMigrationTask task = new StorageMediaMigrationTask(backendId, tabletId,
                        effectiveSchemaHash, storageMedium);
                batchTask.addTask(task);
            }
        }

        AgentTaskQueue.addBatchTask(batchTask);
        AgentTaskExecutor.submit(batchTask);
    }
@Test
    public void testHandleMigration() throws TException {
        List<Long> tabletIds = GlobalStateMgr.getCurrentState().getTabletInvertedIndex().getTabletIdsByBackendId(10001);
        ListMultimap<TStorageMedium, Long> tabletMetaMigrationMap = ArrayListMultimap.create();
        for (Long tabletId : tabletIds) {
            tabletMetaMigrationMap.put(TStorageMedium.SSD, tabletId);
        }
        ReportHandler.handleMigration(tabletMetaMigrationMap, 10001);

        final SystemInfoService currentSystemInfo = GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo();
        Backend reportBackend = currentSystemInfo.getBackend(10001);
        BackendStatus backendStatus = reportBackend.getBackendStatus();
        backendStatus.lastSuccessReportTabletsTime = TimeUtils.longToTimeString(Long.MAX_VALUE);

        ReportHandler.handleMigration(tabletMetaMigrationMap, 10001);

        TabletInvertedIndex invertedIndex = GlobalStateMgr.getCurrentState().getTabletInvertedIndex();
        List<TabletMeta> tabletMetaList = invertedIndex.getTabletMetaList(tabletIds);
        for (int i = 0; i < tabletMetaList.size(); i++) {
            long tabletId = tabletIds.get(i);
            TabletMeta tabletMeta = tabletMetaList.get(i);
            Database db = GlobalStateMgr.getCurrentState().getDb("test");
            if (db == null) {
                continue;
            }
            OlapTable table = null;
            Locker locker = new Locker();
            locker.lockDatabase(db, LockType.READ);
            try {
                table = (OlapTable) db.getTable(tabletMeta.getTableId());
            } finally {
                locker.unLockDatabase(db, LockType.READ);
            }

            Partition partition = table.getPartition(tabletMeta.getPartitionId());
            MaterializedIndex idx = partition.getIndex(tabletMeta.getIndexId());
            LocalTablet tablet = (LocalTablet) idx.getTablet(tabletId);

            for (Replica replica : tablet.getImmutableReplicas()) {
                replica.setMaxRowsetCreationTime(System.currentTimeMillis() / 1000);
            }
        }
        Config.tablet_sched_max_migration_task_sent_once = 1000000;
        Config.primary_key_disk_schedule_time = 0;
        ReportHandler.handleMigration(tabletMetaMigrationMap, 10001);
    }