protected List<Long> chooseTablets() {
        GlobalStateMgr globalStateMgr = GlobalStateMgr.getCurrentState();
        MetaObject chosenOne;

        List<Long> chosenTablets = Lists.newArrayList();

        // sort dbs
        List<Long> dbIds = globalStateMgr.getLocalMetastore().getDbIds();
        if (dbIds.isEmpty()) {
            return chosenTablets;
        }
        Queue<MetaObject> dbQueue = new PriorityQueue<>(dbIds.size(), COMPARATOR);
        for (Long dbId : dbIds) {
            if (dbId == 0L) {
                // skip 'information_schema' database
                continue;
            }
            Database db = globalStateMgr.getDb(dbId);
            if (db == null) {
                continue;
            }
            dbQueue.add(db);
        }

        // must lock jobsLock first to obey the lock order rule
        this.jobsLock.readLock().lock();
        try {
            while ((chosenOne = dbQueue.poll()) != null) {
                Database db = (Database) chosenOne;
                Locker locker = new Locker();
                locker.lockDatabase(db, LockType.READ);
                long startTime = System.currentTimeMillis();
                try {
                    // sort tables
                    List<Table> tables = db.getTables();
                    Queue<MetaObject> tableQueue = new PriorityQueue<>(Math.max(tables.size(), 1), COMPARATOR);
                    for (Table table : tables) {
                        // Only check the OLAP table who is in NORMAL state.
                        // Because some tablets of the not NORMAL table may just a temporary presence in memory,
                        // if we check those tablets and log FinishConsistencyCheck to bdb,
                        // it will throw NullPointerException when replaying the log.
                        if (!table.isOlapTableOrMaterializedView() || ((OlapTable) table).getState() != OlapTableState.NORMAL) {
                            continue;
                        }
                        tableQueue.add(table);
                    }

                    while ((chosenOne = tableQueue.poll()) != null) {
                        OlapTable table = (OlapTable) chosenOne;

                        // sort partitions
                        Queue<MetaObject> partitionQueue =
                                    new PriorityQueue<>(Math.max(table.getAllPhysicalPartitions().size(), 1), COMPARATOR);
                        for (PhysicalPartition partition : table.getPhysicalPartitions()) {
                            // check partition's replication num. if 1 replication. skip
                            if (table.getPartitionInfo().getReplicationNum(partition.getParentId()) == (short) 1) {
                                LOG.debug("partition[{}]'s replication num is 1. ignore", partition.getParentId());
                                continue;
                            }

                            // check if this partition has no data
                            if (partition.getVisibleVersion() == Partition.PARTITION_INIT_VERSION) {
                                LOG.debug("partition[{}]'s version is {}. ignore", partition.getId(),
                                            Partition.PARTITION_INIT_VERSION);
                                continue;
                            }
                            if (partition instanceof Partition) {
                                partitionQueue.add((Partition) partition);
                            } else if (partition instanceof PhysicalPartitionImpl) {
                                partitionQueue.add((PhysicalPartitionImpl) partition);
                            }
                        }

                        while ((chosenOne = partitionQueue.poll()) != null) {
                            PhysicalPartition partition = (PhysicalPartition) chosenOne;

                            // sort materializedIndices
                            List<MaterializedIndex> visibleIndexes =
                                        partition.getMaterializedIndices(IndexExtState.VISIBLE);
                            Queue<MetaObject> indexQueue =
                                        new PriorityQueue<>(Math.max(visibleIndexes.size(), 1), COMPARATOR);
                            indexQueue.addAll(visibleIndexes);

                            while ((chosenOne = indexQueue.poll()) != null) {
                                MaterializedIndex index = (MaterializedIndex) chosenOne;

                                // sort tablets
                                Queue<MetaObject> tabletQueue =
                                            new PriorityQueue<>(Math.max(index.getTablets().size(), 1), COMPARATOR);
                                tabletQueue.addAll(index.getTablets());

                                while ((chosenOne = tabletQueue.poll()) != null) {
                                    LocalTablet tablet = (LocalTablet) chosenOne;
                                    long chosenTabletId = tablet.getId();

                                    if (this.jobs.containsKey(chosenTabletId)) {
                                        continue;
                                    }

                                    // check if version has already been checked
                                    if (partition.getVisibleVersion() == tablet.getCheckedVersion()) {
                                        if (tablet.isConsistent()) {
                                            LOG.debug("tablet[{}]'s version[{}-{}] has been checked. ignore",
                                                        chosenTabletId, tablet.getCheckedVersion(),
                                                        partition.getVisibleVersion());
                                        }
                                    } else {
                                        LOG.info("chose tablet[{}-{}-{}-{}-{}] to check consistency", db.getId(),
                                                    table.getId(), partition.getId(), index.getId(), chosenTabletId);

                                        chosenTablets.add(chosenTabletId);
                                    }
                                } // end while tabletQueue
                            } // end while indexQueue

                            if (chosenTablets.size() >= MAX_JOB_NUM) {
                                return chosenTablets;
                            }
                        } // end while partitionQueue
                    } // end while tableQueue
                } finally {
                    // Since only at most `MAX_JOB_NUM` tablet are chosen, we don't need to release the db read lock
                    // from time to time, just log the time cost here.
                    LOG.info("choose tablets from db[{}-{}](with read lock held) took {}ms",
                                db.getFullName(), db.getId(), System.currentTimeMillis() - startTime);
                    locker.unLockDatabase(db, LockType.READ);
                }
            } // end while dbQueue
        } finally {
            jobsLock.readLock().unlock();
        }

        return chosenTablets;
    }
@Test
    public void testChooseTablets(@Mocked GlobalStateMgr globalStateMgr) {
        long dbId = 1L;
        long tableId = 2L;
        long partitionId = 3L;
        long indexId = 4L;
        long tabletId = 5L;
        long replicaId = 6L;
        long backendId = 7L;
        TStorageMedium medium = TStorageMedium.HDD;

        MaterializedIndex materializedIndex = new MaterializedIndex(indexId, MaterializedIndex.IndexState.NORMAL);
        Replica replica = new Replica(replicaId, backendId, 2L, 1111,
                10, 1000, Replica.ReplicaState.NORMAL, -1, 2);

        TabletMeta tabletMeta = new TabletMeta(dbId, tableId, partitionId, indexId, 1111, medium);
        LocalTablet tablet = new LocalTablet(tabletId, Lists.newArrayList(replica));
        materializedIndex.addTablet(tablet, tabletMeta, false);
        PartitionInfo partitionInfo = new PartitionInfo();
        DataProperty dataProperty = new DataProperty(medium);
        partitionInfo.addPartition(partitionId, dataProperty, (short) 3, false);
        DistributionInfo distributionInfo = new HashDistributionInfo(1, Lists.newArrayList());
        Partition partition = new Partition(partitionId, "partition", materializedIndex, distributionInfo);
        partition.setVisibleVersion(2L, System.currentTimeMillis());
        OlapTable table = new OlapTable(tableId, "table", Lists.newArrayList(), KeysType.AGG_KEYS, partitionInfo,
                distributionInfo);
        table.addPartition(partition);
        Database database = new Database(dbId, "database");
        database.registerTableUnlocked(table);

        new Expectations() {
            {
                GlobalStateMgr.getCurrentState();
                result = globalStateMgr;
                minTimes = 0;

                globalStateMgr.getLocalMetastore().getDbIds();
                result = Lists.newArrayList(dbId);
                minTimes = 0;

                globalStateMgr.getDb(dbId);
                result = database;
                minTimes = 0;
            }
        };

        Assert.assertEquals(1, new ConsistencyChecker().chooseTablets().size());

        // set table state to RESTORE, we will make sure checker will not choose its tablets.
        table.setState(OlapTable.OlapTableState.RESTORE);
        Assert.assertEquals(0, new ConsistencyChecker().chooseTablets().size());
    }