// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

package org.apache.doris.master;


import org.apache.doris.catalog.BinlogConfig;
import org.apache.doris.catalog.ColocateGroupSchema;
import org.apache.doris.catalog.ColocateTableIndex;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.Index;
import org.apache.doris.catalog.MaterializedIndex;
import org.apache.doris.catalog.MaterializedIndex.IndexState;
import org.apache.doris.catalog.MaterializedIndexMeta;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Partition;
import org.apache.doris.catalog.Replica;
import org.apache.doris.catalog.Replica.ReplicaState;
import org.apache.doris.catalog.ReplicaAllocation;
import org.apache.doris.catalog.Resource;
import org.apache.doris.catalog.Resource.ResourceType;
import org.apache.doris.catalog.Table;
import org.apache.doris.catalog.Tablet;
import org.apache.doris.catalog.Tablet.TabletStatus;
import org.apache.doris.catalog.TabletInvertedIndex;
import org.apache.doris.catalog.TabletMeta;
import org.apache.doris.clone.TabletSchedCtx;
import org.apache.doris.common.Config;
import org.apache.doris.common.MetaNotFoundException;
import org.apache.doris.common.Pair;
import org.apache.doris.common.util.Daemon;
import org.apache.doris.common.util.NetUtils;
import org.apache.doris.common.util.TimeUtils;
import org.apache.doris.cooldown.CooldownConf;
import org.apache.doris.metric.GaugeMetric;
import org.apache.doris.metric.Metric.MetricUnit;
import org.apache.doris.metric.MetricRepo;
import org.apache.doris.persist.BackendReplicasInfo;
import org.apache.doris.persist.BackendTabletsInfo;
import org.apache.doris.persist.ReplicaPersistInfo;
import org.apache.doris.policy.Policy;
import org.apache.doris.policy.PolicyTypeEnum;
import org.apache.doris.policy.StoragePolicy;
import org.apache.doris.system.Backend;
import org.apache.doris.system.Backend.BackendStatus;
import org.apache.doris.system.SystemInfoService;
import org.apache.doris.task.AgentBatchTask;
import org.apache.doris.task.AgentTask;
import org.apache.doris.task.AgentTaskExecutor;
import org.apache.doris.task.AgentTaskQueue;
import org.apache.doris.task.ClearTransactionTask;
import org.apache.doris.task.CreateReplicaTask;
import org.apache.doris.task.DropReplicaTask;
import org.apache.doris.task.MasterTask;
import org.apache.doris.task.PublishVersionTask;
import org.apache.doris.task.PushCooldownConfTask;
import org.apache.doris.task.PushStoragePolicyTask;
import org.apache.doris.task.StorageMediaMigrationTask;
import org.apache.doris.task.UpdateTabletMetaInfoTask;
import org.apache.doris.thrift.TBackend;
import org.apache.doris.thrift.TDisk;
import org.apache.doris.thrift.TMasterResult;
import org.apache.doris.thrift.TPartitionVersionInfo;
import org.apache.doris.thrift.TReportRequest;
import org.apache.doris.thrift.TStatus;
import org.apache.doris.thrift.TStatusCode;
import org.apache.doris.thrift.TStorageMedium;
import org.apache.doris.thrift.TStoragePolicy;
import org.apache.doris.thrift.TStorageResource;
import org.apache.doris.thrift.TStorageType;
import org.apache.doris.thrift.TTablet;
import org.apache.doris.thrift.TTabletInfo;
import org.apache.doris.thrift.TTabletMetaInfo;
import org.apache.doris.thrift.TTaskType;

import com.google.common.base.Preconditions;
import com.google.common.collect.LinkedListMultimap;
import com.google.common.collect.ListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Queues;
import com.google.common.collect.Sets;
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.thrift.TException;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.stream.Collectors;

public class ReportHandler extends Daemon {
    private static final Logger LOG = LogManager.getLogger(ReportHandler.class);

    private BlockingQueue<ReportTask> reportQueue = Queues.newLinkedBlockingQueue();

    private enum ReportType {
        UNKNOWN,
        TASK,
        DISK,
        TABLET
    }

    public ReportHandler() {
        super("report-thread");
        GaugeMetric<Long> gauge = new GaugeMetric<Long>(
                "report_queue_size", MetricUnit.NOUNIT, "report queue size") {
            @Override
            public Long getValue() {
                return (long) reportQueue.size();
            }
        };
        MetricRepo.DORIS_METRIC_REGISTER.addMetrics(gauge);
    }

    public TMasterResult handleReport(TReportRequest request) throws TException {
        TMasterResult result = new TMasterResult();
        TStatus tStatus = new TStatus(TStatusCode.OK);
        result.setStatus(tStatus);

        // get backend
        TBackend tBackend = request.getBackend();
        String host = tBackend.getHost();
        int bePort = tBackend.getBePort();
        Backend backend = Env.getCurrentSystemInfo().getBackendWithBePort(host, bePort);
        if (backend == null) {
            tStatus.setStatusCode(TStatusCode.INTERNAL_ERROR);
            List<String> errorMsgs = Lists.newArrayList();
            errorMsgs.add("backend[" + NetUtils
                    .getHostPortInAccessibleFormat(host, bePort) + "] does not exist.");
            tStatus.setErrorMsgs(errorMsgs);
            return result;
        }

        long beId = backend.getId();
        Map<TTaskType, Set<Long>> tasks = null;
        Map<String, TDisk> disks = null;
        Map<Long, TTablet> tablets = null;
        long reportVersion = -1;

        ReportType reportType = ReportType.UNKNOWN;

        if (request.isSetTasks()) {
            tasks = request.getTasks();
            reportType = ReportType.TASK;
        }

        if (request.isSetDisks()) {
            disks = request.getDisks();
            reportType = ReportType.DISK;
        }

        if (request.isSetTablets()) {
            tablets = request.getTablets();
            reportVersion = request.getReportVersion();
            reportType = ReportType.TABLET;
        } else if (request.isSetTabletList()) {
            // the 'tablets' member will be deprecated in future.
            tablets = buildTabletMap(request.getTabletList());
            reportVersion = request.getReportVersion();
            reportType = ReportType.TABLET;
        }

        if (request.isSetTabletMaxCompactionScore()) {
            backend.setTabletMaxCompactionScore(request.getTabletMaxCompactionScore());
        }

        ReportTask reportTask = new ReportTask(beId, tasks, disks, tablets, reportVersion,
                request.getStoragePolicy(), request.getResource(), request.getNumCores(),
                request.getPipelineExecutorSize());
        try {
            putToQueue(reportTask);
        } catch (Exception e) {
            tStatus.setStatusCode(TStatusCode.INTERNAL_ERROR);
            List<String> errorMsgs = Lists.newArrayList();
            errorMsgs.add("failed to put report task to queue. queue size: " + reportQueue.size());
            errorMsgs.add("err: " + e.getMessage());
            tStatus.setErrorMsgs(errorMsgs);
            return result;
        }
        LOG.info("receive report from be {}. type: {}, current queue size: {}",
                backend.getId(), reportType, reportQueue.size());
        return result;
    }

    private void putToQueue(ReportTask reportTask) throws Exception {
        int currentSize = reportQueue.size();
        if (currentSize > Config.report_queue_size) {
            LOG.warn("the report queue size exceeds the limit: {}. current: {}", Config.report_queue_size, currentSize);
            throw new Exception(
                    "the report queue size exceeds the limit: "
                            + Config.report_queue_size + ". current: " + currentSize);
        }
        reportQueue.put(reportTask);
    }

    private Map<Long, TTablet> buildTabletMap(List<TTablet> tabletList) {
        Map<Long, TTablet> tabletMap = Maps.newHashMap();
        for (TTablet tTablet : tabletList) {
            if (tTablet.getTabletInfos().isEmpty()) {
                continue;
            }

            tabletMap.put(tTablet.getTabletInfos().get(0).getTabletId(), tTablet);
        }
        return tabletMap;
    }

    private class ReportTask extends MasterTask {

        private long beId;
        private Map<TTaskType, Set<Long>> tasks;
        private Map<String, TDisk> disks;
        private Map<Long, TTablet> tablets;
        private long reportVersion;

        private List<TStoragePolicy> storagePolicies;
        private List<TStorageResource> storageResources;
        private int cpuCores;
        private int pipelineExecutorSize;

        public ReportTask(long beId, Map<TTaskType, Set<Long>> tasks,
                Map<String, TDisk> disks,
                Map<Long, TTablet> tablets, long reportVersion,
                List<TStoragePolicy> storagePolicies, List<TStorageResource> storageResources, int cpuCores,
                int pipelineExecutorSize) {
            this.beId = beId;
            this.tasks = tasks;
            this.disks = disks;
            this.tablets = tablets;
            this.reportVersion = reportVersion;
            this.storagePolicies = storagePolicies;
            this.storageResources = storageResources;
            this.cpuCores = cpuCores;
            this.pipelineExecutorSize = pipelineExecutorSize;
        }

        @Override
        protected void exec() {
            if (tasks != null) {
                ReportHandler.taskReport(beId, tasks);
            }
            if (disks != null) {
                ReportHandler.diskReport(beId, disks);
                ReportHandler.cpuReport(beId, cpuCores, pipelineExecutorSize);
            }
            if (Config.enable_storage_policy && storagePolicies != null && storageResources != null) {
                storagePolicyReport(beId, storagePolicies, storageResources);
            }

            if (tablets != null) {
                long backendReportVersion = Env.getCurrentSystemInfo().getBackendReportVersion(beId);
                if (reportVersion < backendReportVersion) {
                    LOG.warn("out of date report version {} from backend[{}]. current report version[{}]",
                            reportVersion, beId, backendReportVersion);
                } else {
                    ReportHandler.tabletReport(beId, tablets, reportVersion);
                }
            }
        }
    }

    private static void handlePushCooldownConf(long backendId, List<CooldownConf> cooldownConfToPush) {
        final int PUSH_BATCH_SIZE = 1024;
        AgentBatchTask batchTask = new AgentBatchTask();
        for (int start = 0; start < cooldownConfToPush.size(); start += PUSH_BATCH_SIZE) {
            PushCooldownConfTask task = new PushCooldownConfTask(backendId,
                    cooldownConfToPush.subList(start, Math.min(start + PUSH_BATCH_SIZE, cooldownConfToPush.size())));
            batchTask.addTask(task);
        }
        AgentTaskExecutor.submit(batchTask);
    }

    private static void handlePushStoragePolicy(long backendId, List<Policy> policyToPush,
                                                List<Resource> resourceToPush, List<Long> policyToDrop) {
        AgentBatchTask batchTask = new AgentBatchTask();
        PushStoragePolicyTask pushStoragePolicyTask = new PushStoragePolicyTask(backendId, policyToPush,
                resourceToPush, policyToDrop);
        batchTask.addTask(pushStoragePolicyTask);
        AgentTaskExecutor.submit(batchTask);
    }

    private static void storagePolicyReport(long backendId,
                                            List<TStoragePolicy> storagePoliciesInBe,
                                            List<TStorageResource> storageResourcesInBe) {
        LOG.info("backend[{}] reports policies {}, report resources: {}",
                backendId, storagePoliciesInBe, storageResourcesInBe);
        // do the diff. find out (intersection) / (be - meta) / (meta - be)
        List<Policy> policiesInFe = Env.getCurrentEnv().getPolicyMgr().getCopiedPoliciesByType(PolicyTypeEnum.STORAGE);
        List<Resource> resourcesInFe = Env.getCurrentEnv().getResourceMgr().getResource(ResourceType.S3);
        resourcesInFe.addAll(Env.getCurrentEnv().getResourceMgr().getResource(ResourceType.HDFS));

        List<Resource> resourceToPush = new ArrayList<>();
        List<Policy> policyToPush = new ArrayList<>();
        List<Long> policyToDrop = new ArrayList<>();

        diffPolicy(storagePoliciesInBe, policiesInFe, policyToPush, policyToDrop);
        diffResource(storageResourcesInBe, resourcesInFe, resourceToPush);

        if (policyToPush.isEmpty() && resourceToPush.isEmpty() && policyToDrop.isEmpty()) {
            return;
        }
        LOG.info("after diff policy, policyToPush {}, policyToDrop {}, and resourceToPush {}",
                policyToPush.stream()
                        .map(p -> "StoragePolicy(name=" + p.getPolicyName() + " id=" + p.getId() + " version="
                                + p.getVersion()).collect(Collectors.toList()),
                policyToDrop, resourceToPush.stream()
                        .map(r -> "Resource(name=" + r.getName() + " id=" + r.getId() + " version="
                                + r.getVersion()).collect(Collectors.toList()));
        // send push rpc
        handlePushStoragePolicy(backendId, policyToPush, resourceToPush, policyToDrop);
    }

    private static void diffPolicy(List<TStoragePolicy> storagePoliciesInBe, List<Policy> policiesInFe,
            List<Policy> policyToPush, List<Long> policyToDrop) {
        // fe - be
        for (Policy policy : policiesInFe) {
            if (policy.getId() <= 0 || ((StoragePolicy) policy).getStorageResource() == null) {
                continue; // ignore policy with invalid id or storage resource
            }
            boolean beHasIt = false;
            for (TStoragePolicy tStoragePolicy : storagePoliciesInBe) {
                if (policy.getId() == tStoragePolicy.getId()) {
                    beHasIt = true;
                    // find id eq
                    if (policy.getVersion() == tStoragePolicy.getVersion()) {
                        // find version eq
                    } else if (policy.getVersion() > tStoragePolicy.getVersion()) {
                        // need to add
                        policyToPush.add(policy);
                    } else {
                        // impossible
                        LOG.warn("fe policy version {} litter than be {}, impossible",
                                policy.getVersion(), tStoragePolicy.getVersion());
                    }
                    break;
                }
            }
            if (!beHasIt) {
                policyToPush.add(policy);
            }
        }

        // be - fe
        for (TStoragePolicy tStoragePolicy : storagePoliciesInBe) {
            boolean feHasIt = false;
            for (Policy policy : policiesInFe) {
                if (policy.getId() == tStoragePolicy.getId()) {
                    feHasIt = true;
                    // find id eq
                    break;
                }
            }
            if (!feHasIt) {
                policyToDrop.add(tStoragePolicy.getId());
            }
        }
    }

    private static void diffResource(List<TStorageResource> storageResourcesInBe, List<Resource> resourcesInFe,
                                   List<Resource> resourceToPush) {
        // fe - be
        for (Resource resource : resourcesInFe) {
            if (resource.getId() <= 0) {
                continue; // ignore resource with invalid id
            }
            boolean beHasIt = false;
            for (TStorageResource tStorageResource : storageResourcesInBe) {
                if (resource.getId() == tStorageResource.getId()) {
                    beHasIt = true;
                    // find id eq
                    if (resource.getVersion() == tStorageResource.getVersion()) {
                        // find version eq
                    } else if (resource.getVersion() > tStorageResource.getVersion()) {
                        // need to add
                        resourceToPush.add(resource);
                    } else {
                        // impossible
                        LOG.warn("fe resource version {} litter than be {}, impossible",
                                resource.getVersion(), tStorageResource.getVersion());
                    }
                    break;
                }
            }
            if (!beHasIt) {
                resourceToPush.add(resource);
            }
        }
    }

    // public for fe ut
    public static void tabletReport(long backendId, Map<Long, TTablet> backendTablets, long backendReportVersion) {
        long start = System.currentTimeMillis();
        LOG.info("backend[{}] reports {} tablet(s). report version: {}",
                backendId, backendTablets.size(), backendReportVersion);

        // storage medium map
        HashMap<Long, TStorageMedium> storageMediumMap = Config.disable_storage_medium_check
                ? Maps.newHashMap() : Env.getCurrentEnv().getPartitionIdToStorageMediumMap();

        // db id -> tablet id
        ListMultimap<Long, Long> tabletSyncMap = LinkedListMultimap.create();
        // db id -> tablet id
        ListMultimap<Long, Long> tabletDeleteFromMeta = LinkedListMultimap.create();
        // tablet ids which both in fe and be
        Set<Long> tabletFoundInMeta = Sets.newConcurrentHashSet();
        // storage medium -> tablet id
        ListMultimap<TStorageMedium, Long> tabletMigrationMap = LinkedListMultimap.create();

        // dbid -> txn id -> [partition info]
        Map<Long, ListMultimap<Long, TPartitionVersionInfo>> transactionsToPublish = Maps.newHashMap();
        ListMultimap<Long, Long> transactionsToClear = LinkedListMultimap.create();

        // db id -> tablet id
        ListMultimap<Long, Long> tabletRecoveryMap = LinkedListMultimap.create();

        List<TTabletMetaInfo> tabletToUpdate = Lists.newArrayList();

        List<CooldownConf> cooldownConfToPush = new LinkedList<>();
        List<CooldownConf> cooldownConfToUpdate = new LinkedList<>();

        // 1. do the diff. find out (intersection) / (be - meta) / (meta - be)
        Env.getCurrentInvertedIndex().tabletReport(backendId, backendTablets, storageMediumMap,
                tabletSyncMap,
                tabletDeleteFromMeta,
                tabletFoundInMeta,
                tabletMigrationMap,
                transactionsToPublish,
                transactionsToClear,
                tabletRecoveryMap,
                tabletToUpdate,
                cooldownConfToPush,
                cooldownConfToUpdate);

        // 2. sync
        if (!tabletSyncMap.isEmpty()) {
            sync(backendTablets, tabletSyncMap, backendId, backendReportVersion);
        }

        // 3. delete (meta - be)
        // BE will automatically drop defective tablets. these tablets should also be dropped in catalog
        if (!tabletDeleteFromMeta.isEmpty()) {
            deleteFromMeta(tabletDeleteFromMeta, backendId, backendReportVersion);
        }

        // 4. handle (be - meta)
        if (tabletFoundInMeta.size() != backendTablets.size()) {
            deleteFromBackend(backendTablets, tabletFoundInMeta, backendId);
        }

        // 5. migration (ssd <-> hdd)
        if (!Config.disable_storage_medium_check && !tabletMigrationMap.isEmpty()) {
            handleMigration(tabletMigrationMap, backendId);
        }

        // 6. send clear transactions to be
        if (!transactionsToClear.isEmpty()) {
            handleClearTransactions(transactionsToClear, backendId);
        }

        // 7. send publish version request to be
        if (!transactionsToPublish.isEmpty()) {
            handleRepublishVersionInfo(transactionsToPublish, backendId);
        }

        // 8. send recover request to be
        if (!tabletRecoveryMap.isEmpty()) {
            handleRecoverTablet(tabletRecoveryMap, backendTablets, backendId);
        }

        // 9. send tablet meta to be for updating
        if (!tabletToUpdate.isEmpty()) {
            handleUpdateTabletMeta(backendId, tabletToUpdate);
        }

        // handle cooldown conf
        if (!cooldownConfToPush.isEmpty()) {
            handlePushCooldownConf(backendId, cooldownConfToPush);
        }
        if (!cooldownConfToUpdate.isEmpty()) {
            Env.getCurrentEnv().getCooldownConfHandler().addCooldownConfToUpdate(cooldownConfToUpdate);
        }

        final SystemInfoService currentSystemInfo = Env.getCurrentSystemInfo();
        Backend reportBackend = currentSystemInfo.getBackend(backendId);
        if (reportBackend != null) {
            BackendStatus backendStatus = reportBackend.getBackendStatus();
            backendStatus.lastSuccessReportTabletsTime = TimeUtils.longToTimeString(start);
        }

        long end = System.currentTimeMillis();
        LOG.info("finished to handle tablet report from backend[{}] cost: {} ms", backendId, (end - start));
    }

    private static void taskReport(long backendId, Map<TTaskType, Set<Long>> runningTasks) {
        if (LOG.isDebugEnabled()) {
            LOG.debug("begin to handle task report from backend {}", backendId);
        }
        long start = System.currentTimeMillis();

        if (LOG.isDebugEnabled()) {
            for (TTaskType type : runningTasks.keySet()) {
                Set<Long> taskSet = runningTasks.get(type);
                if (!taskSet.isEmpty()) {
                    String signatures = StringUtils.join(taskSet, ", ");
                    if (LOG.isDebugEnabled()) {
                        LOG.debug("backend task[{}]: {}", type.name(), signatures);
                    }
                }
            }
        }

        List<AgentTask> diffTasks = AgentTaskQueue.getDiffTasks(backendId, runningTasks);

        AgentBatchTask batchTask = new AgentBatchTask();
        long taskReportTime = System.currentTimeMillis();
        for (AgentTask task : diffTasks) {
            // these tasks no need to do diff
            // 1. CREATE
            // 2. SYNC DELETE
            // 3. CHECK_CONSISTENCY
            // 4. STORAGE_MDEIUM_MIGRATE
            if (task.getTaskType() == TTaskType.CREATE
                    || task.getTaskType() == TTaskType.CHECK_CONSISTENCY
                    || task.getTaskType() == TTaskType.STORAGE_MEDIUM_MIGRATE) {
                continue;
            }

            // to escape sending duplicate agent task to be
            if (task.shouldResend(taskReportTime)) {
                batchTask.addTask(task);
            }

        }

        if (LOG.isDebugEnabled()) {
            LOG.debug("get {} diff task(s) to resend", batchTask.getTaskNum());
        }
        if (batchTask.getTaskNum() > 0) {
            AgentTaskExecutor.submit(batchTask);
        }
        LOG.info("finished to handle task report from backend {}, diff task num: {}. cost: {} ms",
                backendId, batchTask.getTaskNum(), (System.currentTimeMillis() - start));
    }

    private static void diskReport(long backendId, Map<String, TDisk> backendDisks) {
        LOG.info("begin to handle disk report from backend {}", backendId);
        long start = System.currentTimeMillis();
        Backend backend = Env.getCurrentSystemInfo().getBackend(backendId);
        if (backend == null) {
            LOG.warn("backend doesn't exist. id: " + backendId);
            return;
        }
        List<String> badDisks = backendDisks.values().stream().filter(disk -> !disk.isUsed())
                .map(disk -> "path=" + disk.getRootPath() + ", path hash=" + disk.getPathHash())
                .collect(Collectors.toList());
        backend.updateDisks(backendDisks);
        LOG.info("finished to handle disk report from backend: {}, disk size: {}, bad disk: {}, cost: {} ms",
                backendId, backendDisks.size(), badDisks, (System.currentTimeMillis() - start));
    }

    private static void cpuReport(long backendId, int cpuCores, int pipelineExecutorSize) {
        LOG.info("begin to handle cpu report from backend {}", backendId);
        long start = System.currentTimeMillis();
        Backend backend = Env.getCurrentSystemInfo().getBackend(backendId);
        if (backend == null) {
            LOG.warn("backend doesn't exist. id: " + backendId);
            return;
        }
        if (backend.updateCpuInfo(cpuCores, pipelineExecutorSize)) {
            // cpu info is changed
            LOG.info("new cpu info. backendId: {}, cpucores: {}, pipelineExecutorSize: {}", backendId, cpuCores,
                    pipelineExecutorSize);
            // log change
            Env.getCurrentEnv().getEditLog().logBackendStateChange(backend);
        }
        LOG.info("finished to handle cpu report from backend {}, cost: {} ms",
                backendId, (System.currentTimeMillis() - start));
    }

    private static void sync(Map<Long, TTablet> backendTablets, ListMultimap<Long, Long> tabletSyncMap,
                             long backendId, long backendReportVersion) {
        TabletInvertedIndex invertedIndex = Env.getCurrentInvertedIndex();
        OUTER:
        for (Long dbId : tabletSyncMap.keySet()) {
            Database db = Env.getCurrentInternalCatalog().getDbNullable(dbId);
            if (db == null) {
                continue;
            }
            int syncCounter = 0;
            List<Long> tabletIds = tabletSyncMap.get(dbId);
            LOG.info("before sync tablets in db[{}]. report num: {}. backend[{}]",
                    dbId, tabletIds.size(), backendId);
            List<TabletMeta> tabletMetaList = invertedIndex.getTabletMetaList(tabletIds);
            for (int i = 0; i < tabletMetaList.size(); i++) {
                TabletMeta tabletMeta = tabletMetaList.get(i);
                if (tabletMeta == TabletInvertedIndex.NOT_EXIST_TABLET_META) {
                    continue;
                }
                long tabletId = tabletIds.get(i);
                long tableId = tabletMeta.getTableId();
                OlapTable olapTable = (OlapTable) db.getTableNullable(tableId);
                if (olapTable == null || !olapTable.writeLockIfExist()) {
                    continue;
                }

                try {
                    if (backendReportVersion < Env.getCurrentSystemInfo().getBackendReportVersion(backendId)) {
                        break OUTER;
                    }

                    long partitionId = tabletMeta.getPartitionId();
                    Partition partition = olapTable.getPartition(partitionId);
                    if (partition == null) {
                        continue;
                    }

                    long indexId = tabletMeta.getIndexId();
                    MaterializedIndex index = partition.getIndex(indexId);
                    if (index == null) {
                        continue;
                    }
                    int schemaHash = olapTable.getSchemaHashByIndexId(indexId);

                    Tablet tablet = index.getTablet(tabletId);
                    if (tablet == null) {
                        continue;
                    }

                    Replica replica = tablet.getReplicaByBackendId(backendId);
                    if (replica == null) {
                        continue;
                    }

                    // yiguolei: it is very important here, if the replica is under schema change or rollup
                    // should ignore the report.
                    // eg.
                    // original replica import successfully, but the dest schema change replica failed
                    // the fe will sync the replica with the original replica, but ignore the schema change replica.
                    // if the last failed version is changed, then fe will think schema change successfully.
                    // this is an fatal error.
                    if (replica.getState() == ReplicaState.NORMAL) {
                        long metaVersion = replica.getVersion();
                        long backendVersion = -1L;
                        long rowCount = -1L;
                        long dataSize = -1L;
                        long remoteDataSize = -1L;
                        // schema change maybe successfully in fe, but not inform be,
                        // then be will report two schema hash
                        // just select the dest schema hash
                        for (TTabletInfo tabletInfo : backendTablets.get(tabletId).getTabletInfos()) {
                            if (tabletInfo.getSchemaHash() == schemaHash) {
                                backendVersion = tabletInfo.getVersion();
                                rowCount = tabletInfo.getRowCount();
                                dataSize = tabletInfo.getDataSize();
                                remoteDataSize = tabletInfo.getRemoteDataSize();
                                break;
                            }
                        }
                        if (backendVersion == -1L) {
                            continue;
                        }

                        boolean needSync = false;
                        if (metaVersion < backendVersion) {
                            needSync = true;
                        } else if (metaVersion == backendVersion) {
                            if (replica.isBad()) {
                                needSync = true;
                            }
                            if (replica.getVersion() >= partition.getCommittedVersion()
                                    && replica.getLastFailedVersion() > partition.getCommittedVersion()) {
                                LOG.info("sync replica {} of tablet {} in backend {} in db {}. replica last failed"
                                        + " version change to -1 because last failed version > replica's committed"
                                        + " version {}",
                                        replica, tabletId, backendId, dbId, partition.getCommittedVersion());
                                replica.updateLastFailedVersion(-1L);
                                needSync = true;
                            }
                        }

                        if (needSync) {
                            // happens when
                            // 1. PUSH finished in BE but failed or not yet report to FE
                            // 2. repair for VERSION_INCOMPLETE finished in BE, but failed or not yet report to FE
                            replica.updateVersionInfo(backendVersion, dataSize, remoteDataSize, rowCount);

                            if (replica.getLastFailedVersion() < 0) {
                                if (replica.setBad(false)) {
                                    LOG.info("sync replica {} of tablet {} in backend {} in db {}. "
                                            + "replica change from bad to good.",
                                            replica, tabletId, backendId, dbId);
                                }

                                // last failed version < 0 means this replica becomes health after sync,
                                // so we write an edit log to sync this operation
                                ReplicaPersistInfo info = ReplicaPersistInfo.createForClone(dbId, tableId,
                                        partitionId, indexId, tabletId, backendId, replica.getId(),
                                        replica.getVersion(), schemaHash,
                                        dataSize, remoteDataSize, rowCount,
                                        replica.getLastFailedVersion(),
                                        replica.getLastSuccessVersion());
                                Env.getCurrentEnv().getEditLog().logUpdateReplica(info);
                            }

                            ++syncCounter;
                            if (LOG.isDebugEnabled()) {
                                LOG.debug("sync replica {} of tablet {} in backend {} in db {}. report version: {}",
                                        replica.getId(), tabletId, backendId, dbId, backendReportVersion);
                            }
                        } else {
                            if (LOG.isDebugEnabled()) {
                                LOG.debug("replica {} of tablet {} in backend {} version is changed"
                                                + " between check and real sync. meta[{}]. backend[{}]",
                                        replica.getId(), tabletId, backendId, metaVersion,
                                        backendVersion);
                            }
                        }
                    }
                } finally {
                    olapTable.writeUnlock();
                }
            }
            LOG.info("sync {} tablets in db[{}]. backend[{}]", syncCounter, dbId, backendId);
        } // end for dbs
    }

    private static void deleteFromMeta(ListMultimap<Long, Long> tabletDeleteFromMeta, long backendId,
                                       long backendReportVersion) {
        AgentBatchTask createReplicaBatchTask = new AgentBatchTask();
        TabletInvertedIndex invertedIndex = Env.getCurrentInvertedIndex();
        for (Long dbId : tabletDeleteFromMeta.keySet()) {
            Database db = Env.getCurrentInternalCatalog().getDbNullable(dbId);
            if (db == null) {
                continue;
            }

            int deleteCounter = 0;
            List<Long> tabletIds = tabletDeleteFromMeta.get(dbId);
            List<TabletMeta> tabletMetaList = invertedIndex.getTabletMetaList(tabletIds);
            for (int i = 0; i < tabletMetaList.size(); i++) {
                TabletMeta tabletMeta = tabletMetaList.get(i);
                if (tabletMeta == TabletInvertedIndex.NOT_EXIST_TABLET_META) {
                    continue;
                }
                long tabletId = tabletIds.get(i);
                long tableId = tabletMeta.getTableId();
                OlapTable olapTable = (OlapTable) db.getTableNullable(tableId);
                if (olapTable == null || !olapTable.writeLockIfExist()) {
                    continue;
                }
                try {
                    long partitionId = tabletMeta.getPartitionId();
                    Partition partition = olapTable.getPartition(partitionId);
                    if (partition == null) {
                        continue;
                    }

                    short replicationNum = olapTable.getPartitionInfo()
                            .getReplicaAllocation(partition.getId()).getTotalReplicaNum();

                    long indexId = tabletMeta.getIndexId();
                    MaterializedIndex index = partition.getIndex(indexId);
                    if (index == null) {
                        continue;
                    }
                    if (index.getState() == IndexState.SHADOW) {
                        // This index is under schema change or rollup, tablet may not be created on BE.
                        // ignore it.
                        continue;
                    }

                    Tablet tablet = index.getTablet(tabletId);
                    if (tablet == null) {
                        continue;
                    }

                    Replica replica = tablet.getReplicaByBackendId(backendId);
                    if (replica == null) {
                        continue;
                    }

                    // check report version again
                    long currentBackendReportVersion = Env.getCurrentSystemInfo()
                            .getBackendReportVersion(backendId);
                    if (backendReportVersion < currentBackendReportVersion) {
                        continue;
                    }

                    BinlogConfig binlogConfig = new BinlogConfig(olapTable.getBinlogConfig());

                    ReplicaState state = replica.getState();
                    if (state == ReplicaState.NORMAL || state == ReplicaState.SCHEMA_CHANGE) {
                        // if state is PENDING / ROLLUP / CLONE
                        // it's normal that the replica is not created in BE but exists in meta.
                        // so we do not delete it.
                        List<Replica> replicas = tablet.getReplicas();
                        if (replicas.size() <= 1) {
                            LOG.error("backend [{}] invalid situation. tablet[{}] has few replica[{}], "
                                            + "replica num setting is [{}]",
                                    backendId, tabletId, replicas.size(), replicationNum);
                            // there is a replica in FE, but not in BE and there is only one replica in this tablet
                            // in this case, it means data is lost.
                            // should generate a create replica request to BE to create a replica forcibly.
                            if (replicas.size() == 1) {
                                if (Config.recover_with_empty_tablet) {
                                    // only create this task if force recovery is true
                                    LOG.warn("tablet {} has only one replica {} on backend {}"
                                                    + " and it is lost. create an empty replica to recover it",
                                            tabletId, replica.getId(), backendId);
                                    MaterializedIndexMeta indexMeta = olapTable.getIndexMetaByIndexId(indexId);
                                    Set<String> bfColumns = olapTable.getCopiedBfColumns();
                                    double bfFpp = olapTable.getBfFpp();
                                    List<Index> indexes = indexId == olapTable.getBaseIndexId()
                                                            ? olapTable.getCopiedIndexes() : null;
                                    CreateReplicaTask createReplicaTask = new CreateReplicaTask(backendId, dbId,
                                            tableId, partitionId, indexId, tabletId, replica.getId(),
                                            indexMeta.getShortKeyColumnCount(),
                                            indexMeta.getSchemaHash(), partition.getVisibleVersion(),
                                            indexMeta.getKeysType(),
                                            TStorageType.COLUMN,
                                            TStorageMedium.HDD, indexMeta.getSchema(), bfColumns, bfFpp, null,
                                            indexes,
                                            olapTable.isInMemory(),
                                            olapTable.getPartitionInfo().getTabletType(partitionId),
                                            null,
                                            olapTable.getCompressionType(),
                                            olapTable.getEnableUniqueKeyMergeOnWrite(), olapTable.getStoragePolicy(),
                                            olapTable.disableAutoCompaction(),
                                            olapTable.enableSingleReplicaCompaction(),
                                            olapTable.skipWriteIndexOnLoad(), olapTable.getCompactionPolicy(),
                                            olapTable.getTimeSeriesCompactionGoalSizeMbytes(),
                                            olapTable.getTimeSeriesCompactionFileCountThreshold(),
                                            olapTable.getTimeSeriesCompactionTimeThresholdSeconds(),
                                            olapTable.getTimeSeriesCompactionEmptyRowsetsThreshold(),
                                            olapTable.getTimeSeriesCompactionLevelThreshold(),
                                            olapTable.storeRowColumn(),
                                            binlogConfig);

                                    createReplicaTask.setIsRecoverTask(true);
                                    createReplicaBatchTask.addTask(createReplicaTask);
                                } else {
                                    // just set this replica as bad
                                    if (replica.setBad(true)) {
                                        LOG.warn("tablet {} has only one replica {} on backend {}"
                                                        + " and it is lost, set it as bad",
                                                tabletId, replica.getId(), backendId);
                                        BackendTabletsInfo tabletsInfo = new BackendTabletsInfo(backendId);
                                        tabletsInfo.setBad(true);
                                        ReplicaPersistInfo replicaPersistInfo = ReplicaPersistInfo.createForReport(dbId,
                                                tableId, partitionId, indexId, tabletId, backendId, replica.getId());
                                        tabletsInfo.addReplicaInfo(replicaPersistInfo);
                                        Env.getCurrentEnv().getEditLog().logBackendTabletsInfo(tabletsInfo);
                                    }
                                }
                            }
                            continue;
                        }

                        tablet.deleteReplicaByBackendId(backendId);
                        ++deleteCounter;

                        // remove replica related tasks
                        AgentTaskQueue.removeReplicaRelatedTasks(backendId, tabletId);

                        // write edit log
                        ReplicaPersistInfo info = ReplicaPersistInfo.createForDelete(dbId, tableId, partitionId,
                                indexId, tabletId, backendId);

                        Env.getCurrentEnv().getEditLog().logDeleteReplica(info);
                        LOG.warn("delete replica[{}] in tablet[{}] from meta. backend[{}], report version: {}"
                                        + ", current report version: {}",
                                replica.getId(), tabletId, backendId, backendReportVersion,
                                currentBackendReportVersion);

                        // check for clone
                        replicas = tablet.getReplicas();
                        if (replicas.size() == 0) {
                            LOG.error("invalid situation. tablet[{}] is empty", tabletId);
                        }
                    }
                } finally {
                    olapTable.writeUnlock();
                }
            }
            LOG.info("delete {} replica(s) from catalog in db[{}]", deleteCounter, dbId);
        } // end for dbs

        if (Config.recover_with_empty_tablet && createReplicaBatchTask.getTaskNum() > 0) {
            // must add to queue, so that when task finish report, the task can be found in queue.
            // the task will be eventually removed from queue by task report, so no need to worry
            // about the residuals.
            AgentTaskQueue.addBatchTask(createReplicaBatchTask);
            AgentTaskExecutor.submit(createReplicaBatchTask);
        }
    }

    private static void deleteFromBackend(Map<Long, TTablet> backendTablets,
                                          Set<Long> tabletFoundInMeta,
                                          long backendId) {
        int deleteFromBackendCounter = 0;
        int addToMetaCounter = 0;
        AgentBatchTask batchTask = new AgentBatchTask();
        TabletInvertedIndex invertedIndex = Env.getCurrentInvertedIndex();
        for (Long tabletId : backendTablets.keySet()) {
            TTablet backendTablet = backendTablets.get(tabletId);
            TTabletInfo backendTabletInfo = backendTablet.getTabletInfos().get(0);
            boolean needDelete = false;
            TabletMeta tabletMeta = invertedIndex.getTabletMeta(tabletId);
            if (LOG.isDebugEnabled()) {
                LOG.debug("process tablet [{}], backend[{}]", tabletId, backendId);
            }
            if (!tabletFoundInMeta.contains(tabletId)) {
                if (isBackendReplicaHealthy(backendTabletInfo)) {
                    // if this tablet meta is still in invertedIndex. try to add it.
                    // if add failed. delete this tablet from backend.
                    if (tabletMeta != null && addReplica(tabletId, tabletMeta, backendTabletInfo, backendId)) {
                        // update counter
                        ++addToMetaCounter;
                        if (LOG.isDebugEnabled()) {
                            LOG.debug("add to meta. tablet[{}], backend[{}]", tabletId, backendId);
                        }
                    } else {
                        LOG.info("failed add to meta. tablet[{}], backend[{}]", tabletId, backendId);
                        needDelete = true;
                    }
                } else {
                    needDelete = true;
                }
            }

            if (needDelete) {
                // drop replica
                long replicaId = backendTabletInfo.getReplicaId();
                // If no such tablet meta, this indicates that the tablet belongs to a dropped table or partition
                boolean isDropTableOrPartition = tabletMeta == null;
                DropReplicaTask task = new DropReplicaTask(backendId, tabletId, replicaId,
                        backendTabletInfo.getSchemaHash(), isDropTableOrPartition);
                batchTask.addTask(task);
                LOG.info("delete tablet[{}] from backend[{}] because not found in meta", tabletId, backendId);
                ++deleteFromBackendCounter;
            }
        } // end for backendTabletIds

        if (batchTask.getTaskNum() != 0) {
            AgentTaskExecutor.submit(batchTask);
        }

        LOG.info("delete {} tablet(s) and add {} replica(s) to meta from backend[{}]",
                deleteFromBackendCounter, addToMetaCounter, backendId);
    }

    // replica is used and no version missing
    private static boolean isBackendReplicaHealthy(TTabletInfo backendTabletInfo) {
        if (backendTabletInfo.isSetUsed() && !backendTabletInfo.isUsed()) {
            return false;
        }
        if (backendTabletInfo.isSetVersionMiss() && backendTabletInfo.isVersionMiss()) {
            return false;
        }
        return true;
    }

    private static void handleMigration(ListMultimap<TStorageMedium, Long> tabletMetaMigrationMap,
                                        long backendId) {
        TabletInvertedIndex invertedIndex = Env.getCurrentInvertedIndex();
        SystemInfoService infoService = Env.getCurrentSystemInfo();
        Backend be = infoService.getBackend(backendId);
        if (be == null) {
            return;
        }
        AgentBatchTask batchTask = new AgentBatchTask();
        for (TStorageMedium storageMedium : tabletMetaMigrationMap.keySet()) {
            List<Long> tabletIds = tabletMetaMigrationMap.get(storageMedium);
            if (!be.hasSpecifiedStorageMedium(storageMedium)) {
                LOG.warn("no specified storage medium {} on backend {}, skip storage migration."
                        + " sample tablet id: {}", storageMedium, backendId, tabletIds.isEmpty()
                        ? "-1" : tabletIds.get(0));
                continue;
            }
            List<TabletMeta> tabletMetaList = invertedIndex.getTabletMetaList(tabletIds);
            for (int i = 0; i < tabletMetaList.size(); i++) {
                long tabletId = tabletIds.get(i);
                TabletMeta tabletMeta = tabletMetaList.get(i);
                if (tabletMeta == TabletInvertedIndex.NOT_EXIST_TABLET_META) {
                    continue;
                }
                // always get old schema hash(as effective one)
                int effectiveSchemaHash = tabletMeta.getOldSchemaHash();
                StorageMediaMigrationTask task = new StorageMediaMigrationTask(backendId, tabletId,
                        effectiveSchemaHash, storageMedium);
                batchTask.addTask(task);
            }
        }

        AgentTaskExecutor.submit(batchTask);
    }

    private static void handleRepublishVersionInfo(
            Map<Long, ListMultimap<Long, TPartitionVersionInfo>> transactionsToPublish, long backendId) {
        AgentBatchTask batchTask = new AgentBatchTask();
        long createPublishVersionTaskTime = System.currentTimeMillis();
        for (Long dbId : transactionsToPublish.keySet()) {
            ListMultimap<Long, TPartitionVersionInfo> map = transactionsToPublish.get(dbId);
            for (long txnId : map.keySet()) {
                PublishVersionTask task = new PublishVersionTask(backendId, txnId, dbId,
                        map.get(txnId), createPublishVersionTaskTime);
                batchTask.addTask(task);
                // add to AgentTaskQueue for handling finish report.
                AgentTaskQueue.addTask(task);
            }
        }
        AgentTaskExecutor.submit(batchTask);
    }

    private static void handleRecoverTablet(ListMultimap<Long, Long> tabletRecoveryMap,
                                            Map<Long, TTablet> backendTablets, long backendId) {
        // print a warn log here to indicate the exceptions on the backend
        LOG.warn("find {} tablets on backend {} which is bad or misses versions that need clone or force recovery",
                tabletRecoveryMap.size(), backendId);

        TabletInvertedIndex invertedIndex = Env.getCurrentInvertedIndex();
        BackendReplicasInfo backendReplicasInfo = new BackendReplicasInfo(backendId);
        for (Long dbId : tabletRecoveryMap.keySet()) {
            Database db = Env.getCurrentInternalCatalog().getDbNullable(dbId);
            if (db == null) {
                continue;
            }
            List<Long> tabletIds = tabletRecoveryMap.get(dbId);
            List<TabletMeta> tabletMetaList = invertedIndex.getTabletMetaList(tabletIds);
            for (int i = 0; i < tabletMetaList.size(); i++) {
                TabletMeta tabletMeta = tabletMetaList.get(i);
                if (tabletMeta == TabletInvertedIndex.NOT_EXIST_TABLET_META) {
                    continue;
                }
                long tabletId = tabletIds.get(i);
                long tableId = tabletMeta.getTableId();
                OlapTable olapTable = (OlapTable) db.getTableNullable(tableId);
                if (olapTable == null || !olapTable.writeLockIfExist()) {
                    continue;
                }
                try {
                    long partitionId = tabletMeta.getPartitionId();
                    Partition partition = olapTable.getPartition(partitionId);
                    if (partition == null) {
                        continue;
                    }

                    long indexId = tabletMeta.getIndexId();
                    MaterializedIndex index = partition.getIndex(indexId);
                    if (index == null) {
                        continue;
                    }

                    int schemaHash = olapTable.getSchemaHashByIndexId(indexId);

                    Tablet tablet = index.getTablet(tabletId);
                    if (tablet == null) {
                        continue;
                    }

                    Replica replica = tablet.getReplicaByBackendId(backendId);
                    if (replica == null) {
                        continue;
                    }

                    for (TTabletInfo tTabletInfo : backendTablets.get(tabletId).getTabletInfos()) {
                        if (tTabletInfo.getSchemaHash() == schemaHash) {
                            if (tTabletInfo.isSetUsed() && !tTabletInfo.isUsed()) {
                                if (replica.setBad(true)) {
                                    LOG.warn("set bad for replica {} of tablet {} on backend {}",
                                            replica.getId(), tabletId, backendId);
                                    backendReplicasInfo.addBadReplica(tabletId);
                                }
                                break;
                            }

                            if ((tTabletInfo.isSetVersionMiss() && tTabletInfo.isVersionMiss())
                                    || replica.checkVersionRegressive(tTabletInfo.getVersion())) {
                                // If the origin last failed version is larger than 0, not change it.
                                // Otherwise, we set last failed version to replica'version + 1.
                                // Because last failed version should always larger than replica's version.
                                long newLastFailedVersion = replica.getLastFailedVersion();
                                if (newLastFailedVersion < 0) {
                                    newLastFailedVersion = replica.getVersion() + 1;
                                    replica.updateLastFailedVersion(newLastFailedVersion);
                                    LOG.warn("set missing version for replica {} of tablet {} on backend {}, "
                                            + "version in fe {}, version in be {}, be missing {}",
                                            replica.getId(), tabletId, backendId, replica.getVersion(),
                                            tTabletInfo.getVersion(), tTabletInfo.isVersionMiss());
                                }
                                backendReplicasInfo.addMissingVersionReplica(tabletId, newLastFailedVersion);
                                break;
                            }

                            break;
                        }
                    }
                } finally {
                    olapTable.writeUnlock();
                }
            }
        } // end for recovery map

        if (!backendReplicasInfo.isEmpty()) {
            // need to write edit log the sync the bad info to other FEs
            Env.getCurrentEnv().getEditLog().logBackendReplicasInfo(backendReplicasInfo);
        }
    }

    private static void handleUpdateTabletMeta(long backendId, List<TTabletMetaInfo> tabletToUpdate) {
        final int updateBatchSize = 4096;
        AgentBatchTask batchTask = new AgentBatchTask();
        for (int start = 0; start < tabletToUpdate.size(); start += updateBatchSize) {
            UpdateTabletMetaInfoTask task = new UpdateTabletMetaInfoTask(backendId,
                    tabletToUpdate.subList(start, Math.min(start + updateBatchSize, tabletToUpdate.size())));
            batchTask.addTask(task);
        }
        AgentTaskExecutor.submit(batchTask);
    }

    private static void handleClearTransactions(ListMultimap<Long, Long> transactionsToClear, long backendId) {
        AgentBatchTask batchTask = new AgentBatchTask();
        for (Long transactionId : transactionsToClear.keySet()) {
            ClearTransactionTask clearTransactionTask = new ClearTransactionTask(backendId,
                    transactionId, transactionsToClear.get(transactionId));
            batchTask.addTask(clearTransactionTask);
        }
        AgentTaskExecutor.submit(batchTask);
    }

    // return false if add replica failed
    private static boolean addReplica(long tabletId, TabletMeta tabletMeta, TTabletInfo backendTabletInfo,
            long backendId) {
        long dbId = tabletMeta.getDbId();
        long tableId = tabletMeta.getTableId();
        long partitionId = tabletMeta.getPartitionId();
        long indexId = tabletMeta.getIndexId();

        int schemaHash = backendTabletInfo.getSchemaHash();
        long version = backendTabletInfo.getVersion();
        long dataSize = backendTabletInfo.getDataSize();
        long remoteDataSize = backendTabletInfo.getRemoteDataSize();
        long rowCount = backendTabletInfo.getRowCount();

        Database db;
        OlapTable olapTable;
        try {
            db = Env.getCurrentInternalCatalog().getDbOrMetaException(dbId);
            olapTable = (OlapTable) db.getTableOrMetaException(tableId, Table.TableType.OLAP);
            olapTable.writeLockOrMetaException();
        } catch (MetaNotFoundException e) {
            LOG.warn(e);
            return false;
        }

        try {
            Partition partition = olapTable.getPartition(partitionId);
            if (partition == null) {
                LOG.warn("partition[{}] does not exist", partitionId);
                return false;
            }
            ReplicaAllocation replicaAlloc = olapTable.getPartitionInfo().getReplicaAllocation(partition.getId());

            MaterializedIndex materializedIndex = partition.getIndex(indexId);
            if (materializedIndex == null) {
                LOG.warn("index[{}] does not exist", indexId);
                return false;
            }

            Tablet tablet = materializedIndex.getTablet(tabletId);
            if (tablet == null) {
                LOG.warn("tablet[{}] does not exist", tabletId);
                return false;
            }

            // check replica id
            long replicaId = backendTabletInfo.getReplicaId();
            if (replicaId <= 0) {
                LOG.warn("replica id is invalid");
                return false;
            }

            long visibleVersion = partition.getVisibleVersion();

            // check replica version
            if (version < visibleVersion) {
                LOG.warn("version is invalid. tablet[{}], visible[{}]", version, visibleVersion);
                return false;
            }

            // check schema hash
            if (schemaHash != olapTable.getSchemaHashByIndexId(indexId)) {
                LOG.warn("schema hash is diff[{}-{}]", schemaHash, olapTable.getSchemaHashByIndexId(indexId));
                return false;
            }

            // colocate table will delete Replica in meta when balance
            // but we need to rely on MetaNotFoundException to decide whether delete the tablet in backend
            // if the tablet is healthy, delete it.
            boolean isColocateBackend = false;
            ColocateTableIndex colocateTableIndex = Env.getCurrentColocateIndex();
            if (colocateTableIndex.isColocateTable(olapTable.getId())) {
                ColocateTableIndex.GroupId groupId = colocateTableIndex.getGroup(tableId);
                Preconditions.checkState(groupId != null,
                        "can not get colocate group for %s", tableId);
                int tabletOrderIdx = materializedIndex.getTabletOrderIdx(tabletId);
                Preconditions.checkState(tabletOrderIdx != -1, "get tablet materializedIndex for %s fail", tabletId);
                Set<Long> backendsSet = colocateTableIndex.getTabletBackendsByGroup(groupId, tabletOrderIdx);
                ColocateGroupSchema groupSchema = colocateTableIndex.getGroupSchema(groupId);
                if (groupSchema != null) {
                    replicaAlloc = groupSchema.getReplicaAlloc();
                }
                TabletStatus status =
                        tablet.getColocateHealthStatus(visibleVersion, replicaAlloc, backendsSet);
                if (status == TabletStatus.HEALTHY) {
                    return false;
                }

                if (backendsSet.contains(backendId)) {
                    isColocateBackend = true;
                }
            }

            SystemInfoService infoService = Env.getCurrentSystemInfo();
            List<Long> aliveBeIds = infoService.getAllBackendIds(true);
            Pair<TabletStatus, TabletSchedCtx.Priority> status = tablet.getHealthStatusWithPriority(infoService,
                    visibleVersion, replicaAlloc, aliveBeIds);

            // FORCE_REDUNDANT is a specific missing case.
            // So it can add replica when it's in FORCE_REDUNDANT.
            // But must be careful to avoid: delete a replica then add it back, then repeat forever.
            // If this replica is sched available and existing another replica is sched unavailable,
            // it's safe to add this replica.
            // Because if the tablet scheduler want to delete a replica, it will choose the sched
            // unavailable replica and avoid the repeating loop as above.
            boolean canAddForceRedundant = status.first == TabletStatus.FORCE_REDUNDANT
                    && infoService.checkBackendScheduleAvailable(backendId)
                    && tablet.getReplicas().stream().anyMatch(
                            r -> !infoService.checkBackendScheduleAvailable(r.getBackendId()));

            if (isColocateBackend
                    || canAddForceRedundant
                    || status.first == TabletStatus.VERSION_INCOMPLETE
                    || status.first == TabletStatus.REPLICA_MISSING
                    || status.first == TabletStatus.UNRECOVERABLE) {
                long lastFailedVersion = -1L;

                // For some partition created by old version's Doris
                // The init partition's version in FE is (1-0), the tablet's version in BE is (2-0)
                // If the BE report version is (2-0) and partition's version is (1-0),
                // we should add the tablet to meta.
                // But old version doris is too old, we should not consider them any more,
                // just throw exception in this case
                if (version > partition.getNextVersion() - 1) {
                    // this is a fatal error
                    LOG.warn("version is invalid. tablet[{}], partition's max version [{}]", version,
                            partition.getNextVersion() - 1);
                    return false;
                } else if (version < partition.getCommittedVersion()) {
                    lastFailedVersion = partition.getCommittedVersion();
                }

                if (backendTabletInfo.isSetCooldownMetaId()) {
                    // replica has cooldowned data
                    do {
                        Pair<Long, Long> cooldownConf = tablet.getCooldownConf();
                        if (backendTabletInfo.getCooldownTerm() > cooldownConf.second) {
                            // should not be here
                            LOG.warn("report cooldownTerm({}) > cooldownTerm in TabletMeta({}), tabletId={}",
                                    backendTabletInfo.getCooldownTerm(), cooldownConf.second, tabletId);
                            return false;
                        }
                        if (backendTabletInfo.getReplicaId() == cooldownConf.first) {
                            // this replica is true cooldown replica, so replica's cooldowned data must not be deleted
                            break;
                        }
                        List<Replica> replicas = Env.getCurrentInvertedIndex().getReplicas(tabletId);
                        if (backendTabletInfo.getCooldownTerm() <= 0) {
                            if (replicas.stream().anyMatch(
                                    r -> backendTabletInfo.getCooldownMetaId().equals(r.getCooldownMetaId()))) {
                                // this backend is just restarted, and shares same cooldowned data with others replica,
                                // so replica's cooldowned data must not be deleted
                                break;
                            }
                        }
                        long minCooldownTerm = Long.MAX_VALUE;
                        for (Replica r : replicas) {
                            minCooldownTerm = Math.min(r.getCooldownTerm(), minCooldownTerm);
                        }
                        if (backendTabletInfo.getCooldownTerm() >= minCooldownTerm) {
                            if (replicas.stream().anyMatch(
                                    r -> backendTabletInfo.getCooldownMetaId().equals(r.getCooldownMetaId()))) {
                                // this replica shares same cooldowned data with others replica, and won't follow data
                                // of lower cooldown term, so replica's cooldowned data must not be deleted
                                break;
                            }
                        }
                        LOG.warn("replica's cooldowned data may have been deleted. tabletId={}, replicaId={}", tabletId,
                                replicaId);
                        return false;
                    } while (false);
                }

                // use replicaId reported by BE to maintain replica meta consistent between FE and BE
                Replica replica = new Replica(replicaId, backendId, version, schemaHash,
                        dataSize, remoteDataSize, rowCount, ReplicaState.NORMAL,
                        lastFailedVersion, version);
                tablet.addReplica(replica);

                // write edit log
                ReplicaPersistInfo info = ReplicaPersistInfo.createForAdd(dbId, tableId, partitionId, indexId,
                        tabletId, backendId, replicaId,
                        version, schemaHash, dataSize, remoteDataSize, rowCount,
                        lastFailedVersion,
                        version);

                Env.getCurrentEnv().getEditLog().logAddReplica(info);

                LOG.info("add replica[{}-{}] to catalog. backend[{}], tablet status {}, tablet size {}, "
                        + "is colocate backend {}",
                        tabletId, replicaId, backendId, status.first.name(), tablet.getReplicas().size(),
                        isColocateBackend);
                return true;
            } else {
                // replica is enough. check if this tablet is already in meta
                // (status changed between 'tabletReport()' and 'addReplica()')
                for (Replica replica : tablet.getReplicas()) {
                    if (replica.getBackendId() == backendId) {
                        // tablet is already in meta. return true
                        return true;
                    }
                }
                LOG.warn("no add replica [{}-{}] cause it is enough[{}-{}], tablet status {}",
                        tabletId, replicaId, tablet.getReplicas().size(), replicaAlloc.toCreateStmt(),
                        status.first.name());
                return false;
            }
        } finally {
            olapTable.writeUnlock();
        }
    }

    @Override
    protected void runOneCycle() {
        while (true) {
            ReportTask task = null;
            try {
                task = reportQueue.take();
                task.exec();
            } catch (InterruptedException e) {
                LOG.warn("got interupted exception when executing report", e);
            }
        }
    }
}