public long beginTransaction(long dbId, List<Long> tableIdList, String label, TUniqueId requestId,
                                 TxnCoordinator coordinator, LoadJobSourceType sourceType, long listenerId,
                                 long timeoutSecond,
                                 long warehouseId)
            throws LabelAlreadyUsedException, RunningTxnExceedException, DuplicatedRequestException, AnalysisException {

        if (Config.disable_load_job) {
            throw ErrorReportException.report(ErrorCode.ERR_BEGIN_TXN_FAILED,
                    "disable_load_job is set to true, all load jobs are rejected");
        }

        if (Config.metadata_enable_recovery_mode) {
            throw ErrorReportException.report(ErrorCode.ERR_BEGIN_TXN_FAILED,
                    "The cluster is under recovery mode, all load jobs are rejected");
        }

        if (GlobalStateMgr.getCurrentState().isSafeMode()) {
            throw ErrorReportException.report(ErrorCode.ERR_BEGIN_TXN_FAILED,
                    "The cluster is under safe mode state, all load jobs are rejected.");
        }

        switch (sourceType) {
            case BACKEND_STREAMING:
                checkValidTimeoutSecond(timeoutSecond, Config.max_stream_load_timeout_second, Config.min_load_timeout_second);
                break;
            case LAKE_COMPACTION:
            case REPLICATION:
                // skip transaction timeout range check for lake compaction and replication
                break;
            default:
                checkValidTimeoutSecond(timeoutSecond, Config.max_load_timeout_second, Config.min_load_timeout_second);
        }

        DatabaseTransactionMgr dbTransactionMgr = getDatabaseTransactionMgr(dbId);
        return dbTransactionMgr.beginTransaction(tableIdList, label, requestId, coordinator, sourceType, listenerId,
                timeoutSecond, warehouseId);
    }
@Test
    public void testBeginTransaction() throws LabelAlreadyUsedException, AnalysisException,
            RunningTxnExceedException, DuplicatedRequestException {
        FakeGlobalStateMgr.setGlobalStateMgr(masterGlobalStateMgr);
        long transactionId = masterTransMgr
                .beginTransaction(GlobalStateMgrTestUtil.testDbId1, Lists.newArrayList(GlobalStateMgrTestUtil.testTableId1),
                        GlobalStateMgrTestUtil.testTxnLable1,
                        transactionSource,
                        LoadJobSourceType.FRONTEND, Config.stream_load_default_timeout_second);
        TransactionState transactionState =
                masterTransMgr.getTransactionState(GlobalStateMgrTestUtil.testDbId1, transactionId);
        assertNotNull(transactionState);
        assertEquals(transactionId, transactionState.getTransactionId());
        assertEquals(TransactionStatus.PREPARE, transactionState.getTransactionStatus());
        assertEquals(GlobalStateMgrTestUtil.testDbId1, transactionState.getDbId());
        assertEquals(transactionSource.toString(), transactionState.getCoordinator().toString());
    }