public void finishTransactionBatch(TransactionStateBatch stateBatch, Set<Long> errorReplicaIds) {
        Database db = globalStateMgr.getDb(stateBatch.getDbId());
        if (db == null) {
            stateBatch.writeLock();
            try {
                writeLock();
                try {
                    stateBatch.setTransactionStatus(TransactionStatus.ABORTED);
                    LOG.warn("db is dropped during transaction batch, abort transaction {}", stateBatch);
                    unprotectSetTransactionStateBatch(stateBatch, false);
                } finally {
                    writeUnlock();
                }
                if (Config.lock_manager_enable_using_fine_granularity_lock) {
                    long start = System.currentTimeMillis();
                    editLog.logInsertTransactionStateBatch(stateBatch);
                    LOG.debug("insert txn state visible for txnIds batch {}, cost: {}ms",
                            stateBatch.getTxnIds(), System.currentTimeMillis() - start);
                }
                return;
            } finally {
                stateBatch.writeUnlock();
            }
        }

        Locker locker = new Locker();
        Set<Long> tableIds = Sets.newHashSet();
        for (TransactionState transactionState : stateBatch.getTransactionStates()) {
            tableIds.addAll(transactionState.getTableIdList());
        }
        locker.lockTablesWithIntensiveDbLock(db, new ArrayList<>(tableIds), LockType.WRITE);

        try {
            boolean txnOperated = false;
            stateBatch.writeLock();
            try {
                writeLock();
                try {
                    stateBatch.setTransactionVisibleInfo();
                    unprotectSetTransactionStateBatch(stateBatch, false);
                    txnOperated = true;
                } finally {
                    writeUnlock();
                    stateBatch.afterVisible(TransactionStatus.VISIBLE, txnOperated);
                }
                if (Config.lock_manager_enable_using_fine_granularity_lock) {
                    long start = System.currentTimeMillis();
                    editLog.logInsertTransactionStateBatch(stateBatch);
                    LOG.debug("insert txn state visible for txnIds batch {}, cost: {}ms",
                            stateBatch.getTxnIds(), System.currentTimeMillis() - start);
                }

                updateCatalogAfterVisibleBatch(stateBatch, db);
            } finally {
                stateBatch.writeUnlock();
            }
        } finally {
            locker.unLockTablesWithIntensiveDbLock(db, new ArrayList<>(tableIds), LockType.WRITE);
        }

        // do after transaction finish in batch
        for (TransactionState transactionState : stateBatch.getTransactionStates()) {
            GlobalStateMgr.getCurrentState().getOperationListenerBus().onStreamJobTransactionFinish(transactionState);
            GlobalStateMgr.getCurrentState().getLocalMetastore().handleMVRepair(transactionState);
        }

        LOG.info("finish transaction {} batch successfully", stateBatch);
    }
@Test
    public void testFinishTransactionBatch() throws UserException {
        FakeGlobalStateMgr.setGlobalStateMgr(masterGlobalStateMgr);
        DatabaseTransactionMgr masterDbTransMgr = masterTransMgr.getDatabaseTransactionMgr(GlobalStateMgrTestUtil.testDbId1);
        long txnId6 = lableToTxnId.get(GlobalStateMgrTestUtil.testTxnLable6);
        TransactionState transactionState6 = masterDbTransMgr.getTransactionState(txnId6);
        long txnId7 = lableToTxnId.get(GlobalStateMgrTestUtil.testTxnLable7);
        TransactionState transactionState7 = masterDbTransMgr.getTransactionState(txnId7);
        long txnId8 = lableToTxnId.get(GlobalStateMgrTestUtil.testTxnLable8);
        TransactionState transactionState8 = masterDbTransMgr.getTransactionState(txnId8);
        List<TransactionState> states = new ArrayList<>();
        states.add(transactionState6);
        states.add(transactionState7);
        states.add(transactionState8);

        new MockUp<Table>() {
            @Mock
            public boolean isCloudNativeTableOrMaterializedView() {
                return true;
            }
        };

        TransactionStateBatch stateBatch = new TransactionStateBatch(states);
        masterTransMgr.finishTransactionBatch(GlobalStateMgrTestUtil.testDbId1, stateBatch, null);

        assertEquals(3, masterDbTransMgr.getRunningTxnNums());
        assertEquals(4, masterDbTransMgr.getFinishedTxnNums());
        assertEquals(TransactionStatus.VISIBLE, transactionState6.getTransactionStatus());
        assertEquals(TransactionStatus.VISIBLE, transactionState7.getTransactionStatus());
        assertEquals(TransactionStatus.VISIBLE, transactionState8.getTransactionStatus());

        FakeGlobalStateMgr.setGlobalStateMgr(slaveGlobalStateMgr);
        slaveTransMgr.replayUpsertTransactionStateBatch(stateBatch);
        assertEquals(4, masterDbTransMgr.getFinishedTxnNums());
    }