private static void handleSetTabletBinlogConfig(long backendId, Map<Long, TTablet> backendTablets) {
        List<Pair<Long, BinlogConfig>> tabletToBinlogConfig = Lists.newArrayList();

        TabletInvertedIndex invertedIndex = GlobalStateMgr.getCurrentState().getTabletInvertedIndex();
        for (TTablet backendTablet : backendTablets.values()) {
            for (TTabletInfo tabletInfo : backendTablet.tablet_infos) {
                if (!tabletInfo.isSetBinlog_config_version()) {
                    continue;
                }
                long tabletId = tabletInfo.getTablet_id();
                long beBinlogConfigVersion = tabletInfo.binlog_config_version;
                TabletMeta tabletMeta = invertedIndex.getTabletMeta(tabletId);
                long dbId = tabletMeta != null ? tabletMeta.getDbId() : TabletInvertedIndex.NOT_EXIST_VALUE;
                long tableId = tabletMeta != null ? tabletMeta.getTableId() : TabletInvertedIndex.NOT_EXIST_VALUE;

                Database db = GlobalStateMgr.getCurrentState().getDb(dbId);
                if (db == null) {
                    continue;
                }

                boolean needToCheck = false;
                OlapTable olapTable = (OlapTable) db.getTable(tableId);
                if (olapTable == null) {
                    continue;
                }
                Locker locker = new Locker();
                locker.lockTablesWithIntensiveDbLock(db, Lists.newArrayList(olapTable.getId()), LockType.READ);
                try {
                    BinlogConfig binlogConfig = olapTable.getCurBinlogConfig();
                    // backward compatible
                    if (binlogConfig == null) {
                        continue;
                    }
                    Long feBinlogConfigVersion = binlogConfig.getVersion();
                    if (beBinlogConfigVersion < feBinlogConfigVersion) {
                        tabletToBinlogConfig.add(new Pair<>(tabletId, olapTable.getCurBinlogConfig()));
                    } else if (beBinlogConfigVersion == feBinlogConfigVersion) {
                        if (olapTable.isBinlogEnabled() && olapTable.getBinlogAvailableVersion().isEmpty()) {
                            // not to check here is that the function may need to get the write db lock
                            needToCheck = true;
                        }
                    } else {
                        LOG.warn("table {} binlog config version of tabletId: {}, BeId: {}, is {} " +
                                        "greater than version of FE, which is {}", olapTable.getName(), tabletId, backendId,
                                beBinlogConfigVersion, feBinlogConfigVersion);
                    }
                } finally {
                    locker.unLockTablesWithIntensiveDbLock(db, Lists.newArrayList(olapTable.getId()), LockType.READ);
                }

                if (needToCheck) {
                    GlobalStateMgr.getCurrentState().getBinlogManager().checkAndSetBinlogAvailableVersion(db,
                            olapTable, tabletId, backendId);
                }
            }
        }

        LOG.debug("find [{}] tablets need set binlog config ", tabletToBinlogConfig.size());
        if (!tabletToBinlogConfig.isEmpty()) {
            AgentBatchTask batchTask = new AgentBatchTask();
            TabletMetadataUpdateAgentTask task = TabletMetadataUpdateAgentTaskFactory.createBinlogConfigUpdateTask(
                    backendId, tabletToBinlogConfig);
            batchTask.addTask(task);
            AgentTaskExecutor.submit(batchTask);
        }
    }
@Test
    public void testHandleSetTabletBinlogConfig() {
        Database db = GlobalStateMgr.getCurrentState().getDb("test");
        long dbId = db.getId();
        OlapTable olapTable = (OlapTable) db.getTable("binlog_report_handler_test");
        long backendId = 10001L;
        List<Long> tabletIds = GlobalStateMgr.getCurrentState().getTabletInvertedIndex().getTabletIdsByBackendId(10001);
        Assert.assertFalse(tabletIds.isEmpty());

        Map<Long, TTablet> backendTablets = new HashMap<Long, TTablet>();
        List<TTabletInfo> tabletInfos = Lists.newArrayList();
        TTablet tablet = new TTablet(tabletInfos);
        for (Long tabletId : tabletIds) {
            TTabletInfo tabletInfo = new TTabletInfo();
            tabletInfo.setTablet_id(tabletId);
            tabletInfo.setSchema_hash(60000);
            tabletInfo.setBinlog_config_version(-1);
            tablet.tablet_infos.add(tabletInfo);
        }
        backendTablets.put(backendId, tablet);

        ReportHandler handler = new ReportHandler();
        handler.testHandleSetTabletBinlogConfig(backendId, backendTablets);

        for (Long tabletId : tabletIds) {
            TTabletInfo tabletInfo = new TTabletInfo();
            tabletInfo.setTablet_id(tabletId);
            tabletInfo.setSchema_hash(60000);
            tabletInfo.setBinlog_config_version(0);
            tablet.tablet_infos.add(tabletInfo);
        }
        backendTablets.put(backendId, tablet);

        handler.testHandleSetTabletBinlogConfig(backendId, backendTablets);
        Assert.assertTrue(GlobalStateMgr.getCurrentState().getBinlogManager().isBinlogAvailable(dbId, olapTable.getId()));

    }