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 testHandleMigrationTaskControl() {
        long backendId = 10001L;
        // mock the task execution on BE
        new MockUp<AgentTaskExecutor>() {
            @Mock
            public void submit(AgentBatchTask task) {

            }
        };

        OlapTable olapTable = (OlapTable) GlobalStateMgr.getCurrentState()
                .getDb("test").getTable("binlog_report_handler_test");
        ListMultimap<TStorageMedium, Long> tabletMetaMigrationMap = ArrayListMultimap.create();
        List<Long> allTablets = new ArrayList<>();
        for (MaterializedIndex index : olapTable.getPartition("binlog_report_handler_test")
                .getMaterializedIndices(MaterializedIndex.IndexExtState.ALL)) {
            for (Tablet tablet : index.getTablets()) {
                tabletMetaMigrationMap.put(TStorageMedium.HDD, tablet.getId());
                allTablets.add(tablet.getId());
            }
        }

        Assert.assertEquals(50, tabletMetaMigrationMap.size());

        ReportHandler.handleMigration(tabletMetaMigrationMap, backendId);

        Assert.assertEquals(50, AgentTaskQueue.getTaskNum(backendId, TTaskType.STORAGE_MEDIUM_MIGRATE, false));

        // finish 30 tablets migration
        for (int i = 0; i < 30; i++) {
            AgentTaskQueue.removeTask(backendId, TTaskType.STORAGE_MEDIUM_MIGRATE, allTablets.get(49 - i));
        }
        // limit the batch size to 30
        Config.tablet_sched_max_migration_task_sent_once = 30;
        ReportHandler.handleMigration(tabletMetaMigrationMap, backendId);
        Assert.assertEquals(30, AgentTaskQueue.getTaskNum(backendId, TTaskType.STORAGE_MEDIUM_MIGRATE, false));
    }