public void checkAndSetBinlogAvailableVersion(Database db, OlapTable table, long tabletId, long beId) {
        lock.writeLock().lock();
        try {
            Locker locker = new Locker();
            try {
                // check if partitions has changed
                locker.lockDatabase(db, LockType.READ);
                Set<Long> partitions = tableIdToPartitions.computeIfAbsent(table.getId(), key -> new HashSet<>());
                boolean isPartitionChanged = false;
                // if partitions is empty indicates that the tablet is
                // the first tablet of the table to be reported,
                // no need to check whether the partitions have been changed
                if (!partitions.isEmpty()) {
                    isPartitionChanged = checkIsPartitionChanged(partitions, table.getAllPhysicalPartitions());
                }

                if (isPartitionChanged) {
                    // for the sake of simplicity, if the partition have been changed, re-statistics
                    partitions.clear();
                    partitions.addAll(table.getAllPhysicalPartitions().stream().
                            map(partition -> partition.getId()).
                            collect(Collectors.toSet()));
                    tableIdToReportedNum.clear();
                    tableIdToBinlogVersion.clear();
                    tableIdToReplicaCount.clear();
                    tabletStatistics.clear();
                }

                Map<Long, Set<Long>> allTabletIds = tabletStatistics.computeIfAbsent(table.getId(), key -> new HashMap<>());

                // the previous version binlog config was not completed,
                // and a new version was generated,
                // the tablets counted need to be cleared
                if (tableIdToBinlogVersion.containsKey(table.getId()) &&
                        tableIdToBinlogVersion.get(table.getId()) != table.getBinlogVersion()) {
                    allTabletIds.clear();
                    tableIdToReportedNum.remove(table.getId());
                }
                tableIdToBinlogVersion.put(table.getId(), table.getBinlogVersion());
                if (!tableIdToReplicaCount.containsKey(table.getId())) {
                    long totalReplicaCount = table.getAllPhysicalPartitions().stream().
                            map(partition -> partition.getBaseIndex().getReplicaCount()).
                            reduce(0L, (acc, n) -> acc + n);
                    tableIdToReplicaCount.put(table.getId(), totalReplicaCount);
                }
                long num = tableIdToReportedNum.computeIfAbsent(table.getId(), key -> 0L);

                Set<Long> allBeIds = allTabletIds.computeIfAbsent(tabletId, key -> new HashSet<>());
                if (!allBeIds.contains(beId)) {
                    num++;
                    tableIdToReportedNum.put(table.getId(), num);
                    allBeIds.add(beId);
                }
            } finally {
                locker.unLockDatabase(db, LockType.READ);
            }

            if (tableIdToReportedNum.get(table.getId()).equals(tableIdToReplicaCount.get(table.getId()))) {
                locker.lockDatabase(db, LockType.WRITE);
                try {
                    // check again if all replicas have been reported
                    long totalReplicaCount = table.getAllPhysicalPartitions().stream().
                            map(partition -> partition.getBaseIndex().getReplicaCount()).
                            reduce(0L, (acc, n) -> acc + n);

                    if (totalReplicaCount != tableIdToReplicaCount.get(table.getId())) {
                        tableIdToBinlogVersion.put(table.getId(), totalReplicaCount);
                        return;
                    }

                    long binlogTxnId = table.getBinlogTxnId();
                    if (binlogTxnId == INVALID) {
                        // the binlog config takes effect in all tablets of the table in BE for now
                        Long nextTxnId = GlobalStateMgr.getCurrentState().getGlobalTransactionMgr().
                                getTransactionIDGenerator().getNextTransactionId();
                        table.setBinlogTxnId(binlogTxnId);
                        binlogTxnId = nextTxnId;
                    }
                    // check whether the concurrent imports when binlog is enabled have completed
                    List<Long> tableList = new ArrayList<>();
                    tableList.add(table.getId());
                    boolean isFinished = GlobalStateMgr.getCurrentState().getGlobalTransactionMgr().
                            getDatabaseTransactionMgr(db.getId()).isPreviousTransactionsFinished(
                                    binlogTxnId, tableList);

                    if (isFinished) {
                        Map<String, String> properties = table.buildBinlogAvailableVersion();
                        ModifyTablePropertyOperationLog log = new ModifyTablePropertyOperationLog(db.getId(),
                                table.getId(), properties);
                        GlobalStateMgr.getCurrentState().getEditLog().logModifyBinlogAvailableVersion(log);
                        table.setBinlogAvailableVersion(properties);
                        LOG.info("set binlog available version tableName : {}, partitions : {}",
                                table.getName(), properties.toString());
                        tabletStatistics.remove(table.getId());
                        tableIdToReportedNum.remove(table.getId());
                        tableIdToReplicaCount.remove(table.getId());
                        tableIdToPartitions.remove(table.getId());
                    }
                } catch (AnalysisException e) {
                    LOG.warn("Failed to execute", e);
                } finally {
                    locker.unLockDatabase(db, LockType.WRITE);
                }
            }
        } finally {
            lock.writeLock().unlock();
        }
    }
@Test
    public void testCheckAndSetBinlogAvailableVersion() {
        Database db = GlobalStateMgr.getCurrentState().getDb("test");
        OlapTable table = (OlapTable) db.getTable("binlog_test");
        table.setBinlogTxnId(2);
        long totalNum = GlobalStateMgr.getCurrentState().getTabletInvertedIndex().getTabletIdsByBackendId(10001).size();
        binlogManager.checkAndSetBinlogAvailableVersion(db, table, 1, 10002);
        Assert.assertFalse(binlogManager.isBinlogAvailable(db.getId(), table.getId()));

        binlogManager.checkAndSetBinlogAvailableVersion(db, table, 2, 10002);
        Assert.assertTrue(binlogManager.isBinlogAvailable(db.getId(), table.getId()));
    }