public void recover() {
        GlobalStateMgr stateMgr = GlobalStateMgr.getCurrentState();
        if (!checkTabletReportCacheUp(stateMgr.getDominationStartTimeMs())) {
            LOG.warn("check tablet report time failed, wait for next round");
            return;
        }

        List<PartitionVersion> partitionsToRecover = new ArrayList<>();
        List<Long> dbIds = stateMgr.getLocalMetastore().getDbIds();
        for (long dbId : dbIds) {
            Database database = stateMgr.getDb(dbId);
            if (database == null || database.isSystemDatabase()) {
                continue;
            }

            Locker locker = new Locker();
            locker.lockDatabase(database, LockType.READ);
            try {
                for (Table table : database.getTables()) {
                    if (!table.isOlapTableOrMaterializedView()) {
                        continue;
                    }

                    OlapTable olapTable = (OlapTable) table;

                    for (Partition partition : olapTable.getAllPartitions()) {
                        short replicaNum = olapTable.getPartitionInfo().getReplicationNum(partition.getId());

                        for (PhysicalPartition physicalPartition : partition.getSubPartitions()) {
                            long recoverVersion = -1L;
                            // commonVersion is used to record all replica versions on this partition.
                            // key is the version, value is a flag means whether this version is existing on all tablets.
                            Map<Long, Boolean> commonVersion = new HashMap<>();
                            StringBuilder info = new StringBuilder();
                            boolean isFirstTablet = true;
                            boolean foundCommonVersion = true;
                            for (MaterializedIndex idx : physicalPartition.getMaterializedIndices(
                                    MaterializedIndex.IndexExtState.VISIBLE)) {
                                for (Tablet tablet : idx.getTablets()) {
                                    LocalTablet localTablet = (LocalTablet) tablet;
                                    long quorumVersion = localTablet.getQuorumVersion(replicaNum / 2 + 1);
                                    if (quorumVersion == -1L) {
                                        foundCommonVersion = false;
                                        info.append(String.format("cannot find quorum version for tablet: %d; ",
                                                tablet.getId()));
                                        LOG.warn("cannot find quorum version for tablet: {}, " +
                                                "ignore partition: {}-{} table: {}-{} db: {}-{}",
                                                tablet.getId(), physicalPartition.getId(), partition.getName(),
                                                table.getId(), table.getName(), database.getId(), database.getFullName());
                                    }
                                    if (recoverVersion == -1L) {
                                        recoverVersion = quorumVersion;
                                    } else if (recoverVersion != quorumVersion) {
                                        foundCommonVersion = false;
                                        info.append(String.format("found different quorum versions: %d %d; ",
                                                recoverVersion, quorumVersion));
                                        LOG.warn("found different quorum versions: {} {}, " +
                                                        "ignore partition: {}-{} table: {}-{} db: {}-{}",
                                                recoverVersion, quorumVersion, physicalPartition.getId(), partition.getName(),
                                                table.getId(), table.getName(), database.getId(), database.getFullName());
                                    }

                                    Set<Long> replicaVersions = localTablet.getAllReplicaVersions();
                                    if (isFirstTablet) {
                                        for (Long version : replicaVersions) {
                                            commonVersion.put(version, true);
                                        }
                                        isFirstTablet = false;
                                    } else {
                                        for (Map.Entry<Long, Boolean> entry : commonVersion.entrySet()) {
                                            if (!replicaVersions.contains(entry.getKey())) {
                                                entry.setValue(false);
                                            }
                                        }
                                    }
                                }
                            }
                            if (foundCommonVersion && recoverVersion != physicalPartition.getVisibleVersion()) {
                                if (GlobalStateMgr.getCurrentState().getGlobalTransactionMgr()
                                        .hasCommittedTxnOnPartition(dbId, table.getId(), physicalPartition.getId())) {
                                    LOG.warn("There are committed txns waiting to publish, " +
                                                    "ignore partition: {}-{} table: {}-{} db: {}-{}",
                                            physicalPartition.getId(), partition.getName(), table.getId(),
                                            table.getName(), database.getId(), database.getFullName());
                                    UnRecoveredPartition recoveryInfo = new UnRecoveredPartition(database.getFullName(),
                                            table.getName(), partition.getName(), physicalPartition.getId(),
                                            "There are committed txns waiting to publish, ignore this partition");
                                    addToUnRecoveredPartitions(recoveryInfo);
                                } else {
                                    partitionsToRecover.add(new PartitionVersion(dbId, table.getId(),
                                            physicalPartition.getId(), recoverVersion));
                                }
                            } else if (!foundCommonVersion) {
                                long maxCommonVersion = -1L;
                                for (Map.Entry<Long, Boolean> entry : commonVersion.entrySet()) {
                                    if (entry.getValue() && entry.getKey() > maxCommonVersion) {
                                        maxCommonVersion = entry.getKey();
                                    }
                                }
                                UnRecoveredPartition recoveryInfo = new UnRecoveredPartition(database.getFullName(),
                                        table.getName(), partition.getName(), physicalPartition.getId(), info.toString());
                                if (maxCommonVersion != -1L) {
                                    recoveryInfo.setMaxCommonVersion(maxCommonVersion);
                                }
                                addToUnRecoveredPartitions(recoveryInfo);
                            }
                        }
                    }
                }
            } finally {
                locker.unLockDatabase(database, LockType.READ);
            }
        }

        PartitionVersionRecoveryInfo recoveryInfo =
                new PartitionVersionRecoveryInfo(partitionsToRecover, System.currentTimeMillis());

        recoverPartitionVersion(recoveryInfo);

        GlobalStateMgr.getCurrentState().getEditLog()
                .logRecoverPartitionVersion(new PartitionVersionRecoveryInfo(partitionsToRecover, System.currentTimeMillis()));
    }
@Test
    public void testRecover() throws Exception {
        PseudoCluster cluster = PseudoCluster.getInstance();
        String sql = "CREATE TABLE test.`tbl_recover` (\n" +
                "    k1 int,\n" +
                "    k2 VARCHAR NOT NULL\n" +
                ") ENGINE=OLAP\n" +
                "DUPLICATE KEY(`k1`)\n" +
                "COMMENT \"OLAP\"\n" +
                "DISTRIBUTED BY HASH(`k1`) BUCKETS 8\n" +
                "PROPERTIES (\n" +
                "    \"replication_num\" = \"3\",\n" +
                "    \"in_memory\" = \"false\"\n" +
                ");";
        cluster.runSql("test", sql);
        cluster.runSql("test", "insert into test.tbl_recover values (1, 'a'), (2, 'b')");

        // wait insert to finish
        Thread.sleep(2000L);

        Database database = GlobalStateMgr.getCurrentState().getLocalMetastore().getDb("test");
        OlapTable table = (OlapTable) database.getTable("tbl_recover");
        Partition partition = table.getPartition("tbl_recover");
        MaterializedIndex index = partition.getMaterializedIndices(MaterializedIndex.IndexExtState.ALL).get(0);
        for (Tablet tablet : index.getTablets()) {
            for (Replica replica : tablet.getAllReplicas()) {
                Assert.assertEquals(2L, replica.getVersion());
            }
        }

        Assert.assertEquals(2L, partition.getVisibleVersion());
        Assert.assertEquals(3L, partition.getNextVersion());

        // set partition version to a lower value
        partition.setVisibleVersion(1L, System.currentTimeMillis());
        partition.setNextVersion(2L);

        for (Backend backend : GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo().getBackends()) {
            backend.getBackendStatus().lastSuccessReportTabletsTime = TimeUtils
                    .longToTimeString(System.currentTimeMillis());
        }

        // add a committed txn
        TransactionState transactionState = new TransactionState(database.getId(), Lists.newArrayList(table.getId()),
                11111, "xxxx", null, TransactionState.LoadJobSourceType.FRONTEND, null, 2222, 100000);
        TableCommitInfo tableCommitInfo = new TableCommitInfo(table.getId());
        PartitionCommitInfo partitionCommitInfo = new PartitionCommitInfo(partition.getId(), 4, -1L);
        tableCommitInfo.addPartitionCommitInfo(partitionCommitInfo);
        transactionState.putIdToTableCommitInfo(table.getId(), tableCommitInfo);
        transactionState.setTransactionStatus(TransactionStatus.COMMITTED);
        GlobalStateMgr.getCurrentState().getGlobalTransactionMgr()
                .getDatabaseTransactionMgr(database.getId()).replayUpsertTransactionState(transactionState);

        // recover will fail, because there is a committed txn on that partition
        MetaRecoveryDaemon recovery = new MetaRecoveryDaemon();
        recovery.recover();
        Assert.assertEquals(1L, partition.getVisibleVersion());
        BaseProcResult baseProcResult = new BaseProcResult();
        recovery.fetchProcNodeResult(baseProcResult);
        Assert.assertEquals(1, baseProcResult.getRows().size());

        // change the txn state to visible, recover will succeed
        transactionState.setTransactionStatus(TransactionStatus.VISIBLE);
        recovery.recover();
        Assert.assertEquals(2L, partition.getVisibleVersion());
        Assert.assertEquals(3L, partition.getNextVersion());
        baseProcResult = new BaseProcResult();
        recovery.fetchProcNodeResult(baseProcResult);
        Assert.assertEquals(0, baseProcResult.getRows().size());

        // change replica version
        LocalTablet localTablet = (LocalTablet) partition.getMaterializedIndices(MaterializedIndex.IndexExtState.ALL)
                .get(0).getTablets().get(0);
        long version = 3;
        for (Replica replica : localTablet.getAllReplicas()) {
            replica.updateForRestore(++version, 10, 10);
        }
        LocalTablet localTablet2 = (LocalTablet) partition.getMaterializedIndices(MaterializedIndex.IndexExtState.ALL)
                .get(0).getTablets().get(0);
        for (Replica replica : localTablet2.getAllReplicas()) {
            replica.updateForRestore(4, 10, 10);
        }

        // set partition version to a lower value
        partition.setVisibleVersion(1L, System.currentTimeMillis());
        partition.setNextVersion(2L);

        // recover will fail, because there is no common version on tablets.
        recovery.recover();
        Assert.assertEquals(1L, partition.getVisibleVersion());
        Assert.assertEquals(2L, partition.getNextVersion());
        baseProcResult = new BaseProcResult();
        recovery.fetchProcNodeResult(baseProcResult);
        Assert.assertEquals(1, baseProcResult.getRows().size());

        // set replica version back to 2
        for (Replica replica : localTablet.getAllReplicas()) {
            replica.updateForRestore(2, 10, 10);
        }
        for (Replica replica : localTablet2.getAllReplicas()) {
            replica.updateForRestore(2, 10, 10);
        }
        Replica badReplica = localTablet2.getAllReplicas().get(0);
        badReplica.updateForRestore(5, 10, 10);
        Assert.assertFalse(badReplica.isBad());

        // recover will succeed
        recovery.recover();
        Assert.assertEquals(2L, partition.getVisibleVersion());
        Assert.assertEquals(3L, partition.getNextVersion());
        baseProcResult = new BaseProcResult();
        recovery.fetchProcNodeResult(baseProcResult);
        Assert.assertEquals(0, baseProcResult.getRows().size());
        // replica with higher version will be set to bad
        Assert.assertTrue(badReplica.isBad());
    }