public void loadJournal(GlobalStateMgr globalStateMgr, JournalEntity journal)
            throws JournalInconsistentException {
        short opCode = journal.getOpCode();
        if (opCode != OperationType.OP_SAVE_NEXTID
                && opCode != OperationType.OP_TIMESTAMP
                && opCode != OperationType.OP_TIMESTAMP_V2) {
            LOG.debug("replay journal op code: {}", opCode);
        }
        try {
            switch (opCode) {
                case OperationType.OP_SAVE_NEXTID: {
                    String idString = journal.getData().toString();
                    long id = Long.parseLong(idString);
                    globalStateMgr.setNextId(id + 1);
                    break;
                }
                case OperationType.OP_SAVE_TRANSACTION_ID_V2: {
                    TransactionIdInfo idInfo = (TransactionIdInfo) journal.getData();
                    GlobalStateMgr.getCurrentState().getGlobalTransactionMgr().getTransactionIDGenerator()
                            .initTransactionId(idInfo.getTxnId() + 1);
                    break;
                }
                case OperationType.OP_SAVE_AUTO_INCREMENT_ID:
                case OperationType.OP_DELETE_AUTO_INCREMENT_ID: {
                    AutoIncrementInfo info = (AutoIncrementInfo) journal.getData();
                    LocalMetastore metastore = globalStateMgr.getLocalMetastore();
                    if (opCode == OperationType.OP_SAVE_AUTO_INCREMENT_ID) {
                        metastore.replayAutoIncrementId(info);
                    } else if (opCode == OperationType.OP_DELETE_AUTO_INCREMENT_ID) {
                        metastore.replayDeleteAutoIncrementId(info);
                    }
                    break;
                }
                case OperationType.OP_CREATE_DB_V2: {
                    CreateDbInfo db = (CreateDbInfo) journal.getData();
                    LocalMetastore metastore = (LocalMetastore) globalStateMgr.getMetadata();
                    metastore.replayCreateDb(db);
                    break;
                }
                case OperationType.OP_DROP_DB: {
                    DropDbInfo dropDbInfo = (DropDbInfo) journal.getData();
                    LocalMetastore metastore = (LocalMetastore) globalStateMgr.getMetadata();
                    metastore.replayDropDb(dropDbInfo.getDbName(), dropDbInfo.isForceDrop());
                    break;
                }
                case OperationType.OP_ALTER_DB:
                case OperationType.OP_ALTER_DB_V2: {

                    DatabaseInfo dbInfo = (DatabaseInfo) journal.getData();
                    globalStateMgr.getLocalMetastore().replayAlterDatabaseQuota(dbInfo);
                    break;
                }
                case OperationType.OP_ERASE_DB: {
                    Text dbId = (Text) journal.getData();
                    globalStateMgr.getLocalMetastore().replayEraseDatabase(Long.parseLong(dbId.toString()));
                    break;
                }
                case OperationType.OP_RECOVER_DB:
                case OperationType.OP_RECOVER_DB_V2: {
                    RecoverInfo info = (RecoverInfo) journal.getData();
                    globalStateMgr.getLocalMetastore().replayRecoverDatabase(info);
                    break;
                }
                case OperationType.OP_RENAME_DB:
                case OperationType.OP_RENAME_DB_V2: {
                    DatabaseInfo dbInfo = (DatabaseInfo) journal.getData();
                    String dbName = dbInfo.getDbName();
                    LOG.info("Begin to unprotect rename db {}", dbName);
                    globalStateMgr.getLocalMetastore().replayRenameDatabase(dbName, dbInfo.getNewDbName());
                    break;
                }
                case OperationType.OP_CREATE_TABLE_V2: {
                    CreateTableInfo info = (CreateTableInfo) journal.getData();

                    if (info.getTable().isMaterializedView()) {
                        LOG.info("Begin to unprotect create materialized view. db = " + info.getDbName()
                                + " create materialized view = " + info.getTable().getId()
                                + " tableName = " + info.getTable().getName());
                    } else {
                        LOG.info("Begin to unprotect create table. db = "
                                + info.getDbName() + " table = " + info.getTable().getId());
                    }
                    globalStateMgr.getLocalMetastore().replayCreateTable(info);
                    break;
                }
                case OperationType.OP_DROP_TABLE:
                case OperationType.OP_DROP_TABLE_V2: {
                    DropInfo info = (DropInfo) journal.getData();
                    Database db = globalStateMgr.getDb(info.getDbId());
                    if (db == null) {
                        LOG.warn("failed to get db[{}]", info.getDbId());
                        break;
                    }
                    LOG.info("Begin to unprotect drop table. db = "
                            + db.getOriginName() + " table = " + info.getTableId());
                    globalStateMgr.getLocalMetastore().replayDropTable(db, info.getTableId(), info.isForceDrop());
                    break;
                }
                case OperationType.OP_ADD_PARTITION_V2: {
                    PartitionPersistInfoV2 info = (PartitionPersistInfoV2) journal.getData();
                    LOG.info("Begin to unprotect add partition. db = " + info.getDbId()
                            + " table = " + info.getTableId()
                            + " partitionName = " + info.getPartition().getName());
                    globalStateMgr.getLocalMetastore().replayAddPartition(info);
                    break;
                }
                case OperationType.OP_ADD_PARTITIONS_V2: {
                    AddPartitionsInfoV2 infos = (AddPartitionsInfoV2) journal.getData();
                    for (PartitionPersistInfoV2 info : infos.getAddPartitionInfos()) {
                        globalStateMgr.getLocalMetastore().replayAddPartition(info);
                    }
                    break;
                }
                case OperationType.OP_ADD_SUB_PARTITIONS_V2: {
                    AddSubPartitionsInfoV2 infos = (AddSubPartitionsInfoV2) journal.getData();
                    for (PhysicalPartitionPersistInfoV2 info : infos.getAddSubPartitionInfos()) {
                        globalStateMgr.getLocalMetastore().replayAddSubPartition(info);
                    }
                    break;
                }
                case OperationType.OP_DROP_PARTITION: {
                    DropPartitionInfo info = (DropPartitionInfo) journal.getData();
                    LOG.info("Begin to unprotect drop partition. db = " + info.getDbId()
                            + " table = " + info.getTableId()
                            + " partitionName = " + info.getPartitionName());
                    globalStateMgr.getLocalMetastore().replayDropPartition(info);
                    break;
                }
                case OperationType.OP_DROP_PARTITIONS: {
                    DropPartitionsInfo info = (DropPartitionsInfo) journal.getData();
                    globalStateMgr.getLocalMetastore().replayDropPartitions(info);
                    break;
                }
                case OperationType.OP_MODIFY_PARTITION:
                case OperationType.OP_MODIFY_PARTITION_V2: {
                    ModifyPartitionInfo info = (ModifyPartitionInfo) journal.getData();
                    LOG.info("Begin to unprotect modify partition. db = " + info.getDbId()
                            + " table = " + info.getTableId() + " partitionId = " + info.getPartitionId());
                    globalStateMgr.getAlterJobMgr().replayModifyPartition(info);
                    break;
                }
                case OperationType.OP_BATCH_MODIFY_PARTITION: {
                    BatchModifyPartitionsInfo info = (BatchModifyPartitionsInfo) journal.getData();
                    for (ModifyPartitionInfo modifyPartitionInfo : info.getModifyPartitionInfos()) {
                        globalStateMgr.getAlterJobMgr().replayModifyPartition(modifyPartitionInfo);
                    }
                    break;
                }
                case OperationType.OP_ERASE_TABLE: {
                    Text tableId = (Text) journal.getData();
                    globalStateMgr.getLocalMetastore().replayEraseTable(Long.parseLong(tableId.toString()));
                    break;
                }
                case OperationType.OP_ERASE_MULTI_TABLES: {
                    MultiEraseTableInfo multiEraseTableInfo = (MultiEraseTableInfo) journal.getData();
                    globalStateMgr.getLocalMetastore().replayEraseMultiTables(multiEraseTableInfo);
                    break;
                }
                case OperationType.OP_DISABLE_TABLE_RECOVERY: {
                    DisableTableRecoveryInfo disableTableRecoveryInfo = (DisableTableRecoveryInfo) journal.getData();
                    globalStateMgr.getLocalMetastore().replayDisableTableRecovery(disableTableRecoveryInfo);
                    break;
                }
                case OperationType.OP_DISABLE_PARTITION_RECOVERY: {
                    DisablePartitionRecoveryInfo disableRecoveryInfo = (DisablePartitionRecoveryInfo) journal.getData();
                    globalStateMgr.getLocalMetastore().replayDisablePartitionRecovery(disableRecoveryInfo);
                    break;
                }
                case OperationType.OP_ERASE_PARTITION: {
                    Text partitionId = (Text) journal.getData();
                    globalStateMgr.getLocalMetastore().replayErasePartition(Long.parseLong(partitionId.toString()));
                    break;
                }
                case OperationType.OP_RECOVER_TABLE:
                case OperationType.OP_RECOVER_TABLE_V2: {
                    RecoverInfo info = (RecoverInfo) journal.getData();
                    globalStateMgr.getLocalMetastore().replayRecoverTable(info);
                    break;
                }
                case OperationType.OP_RECOVER_PARTITION:
                case OperationType.OP_RECOVER_PARTITION_V2: {
                    RecoverInfo info = (RecoverInfo) journal.getData();
                    globalStateMgr.getLocalMetastore().replayRecoverPartition(info);
                    break;
                }
                case OperationType.OP_RENAME_TABLE:
                case OperationType.OP_RENAME_TABLE_V2: {
                    TableInfo info = (TableInfo) journal.getData();
                    globalStateMgr.getLocalMetastore().replayRenameTable(info);
                    break;
                }
                case OperationType.OP_CHANGE_MATERIALIZED_VIEW_REFRESH_SCHEME: {
                    ChangeMaterializedViewRefreshSchemeLog log =
                            (ChangeMaterializedViewRefreshSchemeLog) journal.getData();
                    globalStateMgr.getAlterJobMgr().replayChangeMaterializedViewRefreshScheme(log);
                    break;
                }
                case OperationType.OP_ALTER_MATERIALIZED_VIEW_PROPERTIES: {
                    ModifyTablePropertyOperationLog log =
                            (ModifyTablePropertyOperationLog) journal.getData();
                    globalStateMgr.getAlterJobMgr().replayAlterMaterializedViewProperties(opCode, log);
                    break;
                }
                case OperationType.OP_ALTER_MATERIALIZED_VIEW_STATUS: {
                    AlterMaterializedViewStatusLog log =
                            (AlterMaterializedViewStatusLog) journal.getData();
                    globalStateMgr.getAlterJobMgr().replayAlterMaterializedViewStatus(log);
                    break;
                }
                case OperationType.OP_ALTER_MATERIALIZED_VIEW_BASE_TABLE_INFOS: {
                    AlterMaterializedViewBaseTableInfosLog log =
                            (AlterMaterializedViewBaseTableInfosLog) journal.getData();
                    globalStateMgr.getAlterJobMgr().replayAlterMaterializedViewBaseTableInfos(log);
                    break;
                }
                case OperationType.OP_RENAME_MATERIALIZED_VIEW: {
                    RenameMaterializedViewLog log = (RenameMaterializedViewLog) journal.getData();
                    globalStateMgr.getAlterJobMgr().replayRenameMaterializedView(log);
                    break;
                }
                case OperationType.OP_MODIFY_VIEW_DEF: {
                    AlterViewInfo info = (AlterViewInfo) journal.getData();
                    globalStateMgr.getAlterJobMgr().alterView(info);
                    break;
                }
                case OperationType.OP_RENAME_PARTITION:
                case OperationType.OP_RENAME_PARTITION_V2: {
                    TableInfo info = (TableInfo) journal.getData();
                    globalStateMgr.getLocalMetastore().replayRenamePartition(info);
                    break;
                }
                case OperationType.OP_RENAME_COLUMN_V2: {
                    ColumnRenameInfo info = (ColumnRenameInfo) journal.getData();
                    globalStateMgr.getLocalMetastore().replayRenameColumn(info);
                    break;
                }
                case OperationType.OP_BACKUP_JOB:
                case OperationType.OP_BACKUP_JOB_V2: {
                    BackupJob job = (BackupJob) journal.getData();
                    globalStateMgr.getBackupHandler().replayAddJob(job);
                    break;
                }
                case OperationType.OP_RESTORE_JOB:
                case OperationType.OP_RESTORE_JOB_V2: {
                    RestoreJob job = (RestoreJob) journal.getData();
                    job.setGlobalStateMgr(globalStateMgr);
                    globalStateMgr.getBackupHandler().replayAddJob(job);
                    break;
                }
                case OperationType.OP_DROP_ROLLUP:
                case OperationType.OP_DROP_ROLLUP_V2: {
                    DropInfo info = (DropInfo) journal.getData();
                    globalStateMgr.getRollupHandler().replayDropRollup(info, globalStateMgr);
                    break;
                }
                case OperationType.OP_BATCH_DROP_ROLLUP: {
                    BatchDropInfo batchDropInfo = (BatchDropInfo) journal.getData();
                    for (long indexId : batchDropInfo.getIndexIdSet()) {
                        globalStateMgr.getRollupHandler().replayDropRollup(
                                new DropInfo(batchDropInfo.getDbId(), batchDropInfo.getTableId(), indexId, false),
                                globalStateMgr);
                    }
                    break;
                }
                case OperationType.OP_FINISH_CONSISTENCY_CHECK:
                case OperationType.OP_FINISH_CONSISTENCY_CHECK_V2: {
                    ConsistencyCheckInfo info = (ConsistencyCheckInfo) journal.getData();
                    globalStateMgr.getConsistencyChecker().replayFinishConsistencyCheck(info, globalStateMgr);
                    break;
                }
                case OperationType.OP_CLEAR_ROLLUP_INFO: {
                    // Nothing to do
                    break;
                }
                case OperationType.OP_RENAME_ROLLUP:
                case OperationType.OP_RENAME_ROLLUP_V2: {
                    TableInfo info = (TableInfo) journal.getData();
                    globalStateMgr.getLocalMetastore().replayRenameRollup(info);
                    break;
                }
                case OperationType.OP_EXPORT_CREATE:
                case OperationType.OP_EXPORT_CREATE_V2: {
                    ExportJob job = (ExportJob) journal.getData();
                    ExportMgr exportMgr = globalStateMgr.getExportMgr();
                    exportMgr.replayCreateExportJob(job);
                    break;
                }
                case OperationType.OP_EXPORT_UPDATE_STATE:
                    ExportJob.StateTransfer op = (ExportJob.StateTransfer) journal.getData();
                    ExportMgr exportMgr = globalStateMgr.getExportMgr();
                    exportMgr.replayUpdateJobState(op.getJobId(), op.getState());
                    break;
                case OperationType.OP_EXPORT_UPDATE_INFO_V2:
                case OperationType.OP_EXPORT_UPDATE_INFO:
                    ExportJob.ExportUpdateInfo exportUpdateInfo = (ExportJob.ExportUpdateInfo) journal.getData();
                    globalStateMgr.getExportMgr().replayUpdateJobInfo(exportUpdateInfo);
                    break;
                case OperationType.OP_FINISH_DELETE: {
                    DeleteInfo info = (DeleteInfo) journal.getData();
                    DeleteMgr deleteHandler = globalStateMgr.getDeleteMgr();
                    deleteHandler.replayDelete(info, globalStateMgr);
                    break;
                }
                case OperationType.OP_FINISH_MULTI_DELETE: {
                    MultiDeleteInfo info = (MultiDeleteInfo) journal.getData();
                    DeleteMgr deleteHandler = globalStateMgr.getDeleteMgr();
                    deleteHandler.replayMultiDelete(info, globalStateMgr);
                    break;
                }
                case OperationType.OP_ADD_REPLICA:
                case OperationType.OP_ADD_REPLICA_V2: {
                    ReplicaPersistInfo info = (ReplicaPersistInfo) journal.getData();
                    globalStateMgr.getLocalMetastore().replayAddReplica(info);
                    break;
                }
                case OperationType.OP_UPDATE_REPLICA:
                case OperationType.OP_UPDATE_REPLICA_V2: {
                    ReplicaPersistInfo info = (ReplicaPersistInfo) journal.getData();
                    globalStateMgr.getLocalMetastore().replayUpdateReplica(info);
                    break;
                }
                case OperationType.OP_DELETE_REPLICA:
                case OperationType.OP_DELETE_REPLICA_V2: {
                    ReplicaPersistInfo info = (ReplicaPersistInfo) journal.getData();
                    globalStateMgr.getLocalMetastore().replayDeleteReplica(info);
                    break;
                }
                case OperationType.OP_BATCH_DELETE_REPLICA: {
                    BatchDeleteReplicaInfo info = (BatchDeleteReplicaInfo) journal.getData();
                    globalStateMgr.getLocalMetastore().replayBatchDeleteReplica(info);
                    break;
                }
                case OperationType.OP_ADD_COMPUTE_NODE: {
                    ComputeNode computeNode = (ComputeNode) journal.getData();
                    GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo().replayAddComputeNode(computeNode);
                    break;
                }
                case OperationType.OP_DROP_COMPUTE_NODE: {
                    DropComputeNodeLog dropComputeNodeLog = (DropComputeNodeLog) journal.getData();
                    GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo()
                            .replayDropComputeNode(dropComputeNodeLog.getComputeNodeId());
                    break;
                }
                case OperationType.OP_ADD_BACKEND:
                case OperationType.OP_ADD_BACKEND_V2: {
                    Backend be = (Backend) journal.getData();
                    GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo().replayAddBackend(be);
                    break;
                }
                case OperationType.OP_DROP_BACKEND:
                case OperationType.OP_DROP_BACKEND_V2: {
                    Backend be = (Backend) journal.getData();
                    GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo().replayDropBackend(be);
                    break;
                }
                case OperationType.OP_BACKEND_STATE_CHANGE:
                case OperationType.OP_BACKEND_STATE_CHANGE_V2: {
                    Backend be = (Backend) journal.getData();
                    GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo().updateInMemoryStateBackend(be);
                    break;
                }
                case OperationType.OP_ADD_FIRST_FRONTEND:
                case OperationType.OP_ADD_FIRST_FRONTEND_V2:
                case OperationType.OP_ADD_FRONTEND:
                case OperationType.OP_ADD_FRONTEND_V2: {
                    Frontend fe = (Frontend) journal.getData();
                    globalStateMgr.getNodeMgr().replayAddFrontend(fe);
                    break;
                }
                case OperationType.OP_REMOVE_FRONTEND:
                case OperationType.OP_REMOVE_FRONTEND_V2: {
                    Frontend fe = (Frontend) journal.getData();
                    globalStateMgr.getNodeMgr().replayDropFrontend(fe);
                    if (fe.getNodeName().equals(GlobalStateMgr.getCurrentState().getNodeMgr().getNodeName())) {
                        throw new JournalInconsistentException("current fe " + fe + " is removed. will exit");
                    }
                    break;
                }
                case OperationType.OP_UPDATE_FRONTEND:
                case OperationType.OP_UPDATE_FRONTEND_V2: {
                    Frontend fe = (Frontend) journal.getData();
                    globalStateMgr.getNodeMgr().replayUpdateFrontend(fe);
                    break;
                }
                case OperationType.OP_TIMESTAMP:
                case OperationType.OP_TIMESTAMP_V2: {
                    Timestamp stamp = (Timestamp) journal.getData();
                    globalStateMgr.setSynchronizedTime(stamp.getTimestamp());
                    break;
                }
                case OperationType.OP_LEADER_INFO_CHANGE:
                case OperationType.OP_LEADER_INFO_CHANGE_V2: {
                    LeaderInfo info = (LeaderInfo) journal.getData();
                    globalStateMgr.setLeader(info);
                    break;
                }
                //compatible with old community meta, newly added log using OP_META_VERSION_V2
                case OperationType.OP_META_VERSION: {
                    break;
                }
                case OperationType.OP_META_VERSION_V2: {
                    MetaVersion metaVersion = (MetaVersion) journal.getData();
                    if (!MetaVersion.isCompatible(metaVersion.getStarRocksVersion(), FeConstants.STARROCKS_META_VERSION)) {
                        throw new JournalInconsistentException("Not compatible with meta version "
                                + metaVersion.getStarRocksVersion()
                                + ", current version is " + FeConstants.STARROCKS_META_VERSION);
                    }
                    MetaContext.get().setStarRocksMetaVersion(metaVersion.getStarRocksVersion());
                    break;
                }
                case OperationType.OP_CREATE_CLUSTER: {
                    // ignore
                    break;
                }
                case OperationType.OP_ADD_BROKER:
                case OperationType.OP_ADD_BROKER_V2: {
                    final BrokerMgr.ModifyBrokerInfo param = (BrokerMgr.ModifyBrokerInfo) journal.getData();
                    globalStateMgr.getBrokerMgr().replayAddBrokers(param.brokerName, param.brokerAddresses);
                    break;
                }
                case OperationType.OP_DROP_BROKER:
                case OperationType.OP_DROP_BROKER_V2: {
                    final BrokerMgr.ModifyBrokerInfo param = (BrokerMgr.ModifyBrokerInfo) journal.getData();
                    globalStateMgr.getBrokerMgr().replayDropBrokers(param.brokerName, param.brokerAddresses);
                    break;
                }
                case OperationType.OP_DROP_ALL_BROKER: {
                    final String param = journal.getData().toString();
                    globalStateMgr.getBrokerMgr().replayDropAllBroker(param);
                    break;
                }
                case OperationType.OP_SET_LOAD_ERROR_HUB: {
                    final LoadErrorHub.Param param = (LoadErrorHub.Param) journal.getData();
                    globalStateMgr.getLoadInstance().setLoadErrorHubInfo(param);
                    break;
                }
                case OperationType.OP_UPDATE_CLUSTER_AND_BACKENDS: {
                    final BackendIdsUpdateInfo info = (BackendIdsUpdateInfo) journal.getData();
                    globalStateMgr.replayUpdateClusterAndBackends(info);
                    break;
                }
                case OperationType.OP_UPSERT_TRANSACTION_STATE_V2: {
                    final TransactionState state = (TransactionState) journal.getData();
                    GlobalStateMgr.getCurrentState().getGlobalTransactionMgr().replayUpsertTransactionState(state);
                    LOG.debug("opcode: {}, tid: {}", opCode, state.getTransactionId());
                    break;
                }
                case OperationType.OP_UPSERT_TRANSACTION_STATE_BATCH: {
                    final TransactionStateBatch stateBatch = (TransactionStateBatch) journal.getData();
                    GlobalStateMgr.getCurrentState().getGlobalTransactionMgr().replayUpsertTransactionStateBatch(stateBatch);
                    LOG.debug("opcode: {}, txn ids: {}", opCode, stateBatch.getTxnIds());
                    break;
                }
                case OperationType.OP_CREATE_REPOSITORY:
                case OperationType.OP_CREATE_REPOSITORY_V2: {
                    Repository repository = (Repository) journal.getData();
                    globalStateMgr.getBackupHandler().getRepoMgr().addAndInitRepoIfNotExist(repository, true);
                    break;
                }
                case OperationType.OP_DROP_REPOSITORY: {
                    String repoName = ((Text) journal.getData()).toString();
                    globalStateMgr.getBackupHandler().getRepoMgr().removeRepo(repoName, true);
                    break;
                }
                case OperationType.OP_TRUNCATE_TABLE: {
                    TruncateTableInfo info = (TruncateTableInfo) journal.getData();
                    globalStateMgr.getLocalMetastore().replayTruncateTable(info);
                    break;
                }
                case OperationType.OP_COLOCATE_ADD_TABLE:
                case OperationType.OP_COLOCATE_ADD_TABLE_V2: {
                    final ColocatePersistInfo info = (ColocatePersistInfo) journal.getData();
                    globalStateMgr.getColocateTableIndex().replayAddTableToGroup(info);
                    break;
                }
                case OperationType.OP_COLOCATE_REMOVE_TABLE: {
                    final ColocatePersistInfo info = (ColocatePersistInfo) journal.getData();
                    globalStateMgr.getColocateTableIndex().replayRemoveTable(info);
                    break;
                }
                case OperationType.OP_COLOCATE_BACKENDS_PER_BUCKETSEQ:
                case OperationType.OP_COLOCATE_BACKENDS_PER_BUCKETSEQ_V2: {
                    final ColocatePersistInfo info = (ColocatePersistInfo) journal.getData();
                    globalStateMgr.getColocateTableIndex().replayAddBackendsPerBucketSeq(info);
                    break;
                }
                case OperationType.OP_COLOCATE_MARK_UNSTABLE:
                case OperationType.OP_COLOCATE_MARK_UNSTABLE_V2: {
                    final ColocatePersistInfo info = (ColocatePersistInfo) journal.getData();
                    globalStateMgr.getColocateTableIndex().replayMarkGroupUnstable(info);
                    break;
                }
                case OperationType.OP_COLOCATE_MARK_STABLE:
                case OperationType.OP_COLOCATE_MARK_STABLE_V2: {
                    final ColocatePersistInfo info = (ColocatePersistInfo) journal.getData();
                    globalStateMgr.getColocateTableIndex().replayMarkGroupStable(info);
                    break;
                }
                case OperationType.OP_MODIFY_TABLE_COLOCATE:
                case OperationType.OP_MODIFY_TABLE_COLOCATE_V2: {
                    final TablePropertyInfo info = (TablePropertyInfo) journal.getData();
                    globalStateMgr.getColocateTableIndex().replayModifyTableColocate(info);
                    break;
                }
                case OperationType.OP_HEARTBEAT_V2:
                case OperationType.OP_HEARTBEAT: {
                    final HbPackage hbPackage = (HbPackage) journal.getData();
                    GlobalStateMgr.getCurrentState().getHeartbeatMgr().replayHearbeat(hbPackage);
                    break;
                }
                case OperationType.OP_ADD_FUNCTION:
                case OperationType.OP_ADD_FUNCTION_V2: {
                    final Function function = (Function) journal.getData();
                    if (function.getFunctionName().isGlobalFunction()) {
                        GlobalStateMgr.getCurrentState().getGlobalFunctionMgr().replayAddFunction(function);
                    } else {
                        Database.replayCreateFunctionLog(function);
                    }
                    break;
                }
                case OperationType.OP_DROP_FUNCTION:
                case OperationType.OP_DROP_FUNCTION_V2: {
                    FunctionSearchDesc function = (FunctionSearchDesc) journal.getData();
                    if (function.getName().isGlobalFunction()) {
                        GlobalStateMgr.getCurrentState().getGlobalFunctionMgr().replayDropFunction(function);
                    } else {
                        Database.replayDropFunctionLog(function);
                    }
                    break;
                }
                case OperationType.OP_BACKEND_TABLETS_INFO:
                case OperationType.OP_BACKEND_TABLETS_INFO_V2: {
                    BackendTabletsInfo backendTabletsInfo = (BackendTabletsInfo) journal.getData();
                    GlobalStateMgr.getCurrentState().getLocalMetastore().replayBackendTabletsInfo(backendTabletsInfo);
                    break;
                }
                case OperationType.OP_CREATE_ROUTINE_LOAD_JOB_V2:
                case OperationType.OP_CREATE_ROUTINE_LOAD_JOB: {
                    RoutineLoadJob routineLoadJob = (RoutineLoadJob) journal.getData();
                    GlobalStateMgr.getCurrentState().getRoutineLoadMgr().replayCreateRoutineLoadJob(routineLoadJob);
                    break;
                }
                case OperationType.OP_CHANGE_ROUTINE_LOAD_JOB_V2:
                case OperationType.OP_CHANGE_ROUTINE_LOAD_JOB: {
                    RoutineLoadOperation operation = (RoutineLoadOperation) journal.getData();
                    GlobalStateMgr.getCurrentState().getRoutineLoadMgr().replayChangeRoutineLoadJob(operation);
                    break;
                }
                case OperationType.OP_REMOVE_ROUTINE_LOAD_JOB: {
                    RoutineLoadOperation operation = (RoutineLoadOperation) journal.getData();
                    globalStateMgr.getRoutineLoadMgr().replayRemoveOldRoutineLoad(operation);
                    break;
                }
                case OperationType.OP_CREATE_STREAM_LOAD_TASK:
                case OperationType.OP_CREATE_STREAM_LOAD_TASK_V2: {
                    StreamLoadTask streamLoadTask = (StreamLoadTask) journal.getData();
                    globalStateMgr.getStreamLoadMgr().replayCreateLoadTask(streamLoadTask);
                    break;
                }
                case OperationType.OP_CREATE_LOAD_JOB_V2:
                case OperationType.OP_CREATE_LOAD_JOB: {
                    com.starrocks.load.loadv2.LoadJob loadJob =
                            (com.starrocks.load.loadv2.LoadJob) journal.getData();
                    globalStateMgr.getLoadMgr().replayCreateLoadJob(loadJob);
                    break;
                }
                case OperationType.OP_END_LOAD_JOB_V2:
                case OperationType.OP_END_LOAD_JOB: {
                    LoadJobFinalOperation operation = (LoadJobFinalOperation) journal.getData();
                    globalStateMgr.getLoadMgr().replayEndLoadJob(operation);
                    break;
                }
                case OperationType.OP_UPDATE_LOAD_JOB: {
                    LoadJobStateUpdateInfo info = (LoadJobStateUpdateInfo) journal.getData();
                    globalStateMgr.getLoadMgr().replayUpdateLoadJobStateInfo(info);
                    break;
                }
                case OperationType.OP_CREATE_RESOURCE: {
                    final Resource resource = (Resource) journal.getData();
                    globalStateMgr.getResourceMgr().replayCreateResource(resource);
                    break;
                }
                case OperationType.OP_DROP_RESOURCE: {
                    final DropResourceOperationLog operationLog = (DropResourceOperationLog) journal.getData();
                    globalStateMgr.getResourceMgr().replayDropResource(operationLog);
                    break;
                }
                case OperationType.OP_RESOURCE_GROUP: {
                    final ResourceGroupOpEntry entry = (ResourceGroupOpEntry) journal.getData();
                    globalStateMgr.getResourceGroupMgr().replayResourceGroupOp(entry);
                    break;
                }
                case OperationType.OP_CREATE_TASK: {
                    final Task task = (Task) journal.getData();
                    globalStateMgr.getTaskManager().replayCreateTask(task);
                    break;
                }
                case OperationType.OP_DROP_TASKS: {
                    DropTasksLog dropTasksLog = (DropTasksLog) journal.getData();
                    globalStateMgr.getTaskManager().replayDropTasks(dropTasksLog.getTaskIdList());
                    break;
                }
                case OperationType.OP_ALTER_TASK: {
                    final Task task = (Task) journal.getData();
                    globalStateMgr.getTaskManager().replayAlterTask(task);
                    break;
                }
                case OperationType.OP_CREATE_TASK_RUN: {
                    final TaskRunStatus status = (TaskRunStatus) journal.getData();
                    globalStateMgr.getTaskManager().replayCreateTaskRun(status);
                    break;
                }
                case OperationType.OP_UPDATE_TASK_RUN: {
                    final TaskRunStatusChange statusChange =
                            (TaskRunStatusChange) journal.getData();
                    globalStateMgr.getTaskManager().replayUpdateTaskRun(statusChange);
                    break;
                }
                case OperationType.OP_DROP_TASK_RUNS: {
                    DropTaskRunsLog dropTaskRunsLog = (DropTaskRunsLog) journal.getData();
                    globalStateMgr.getTaskManager().replayDropTaskRuns(dropTaskRunsLog.getQueryIdList());
                    break;
                }
                case OperationType.OP_UPDATE_TASK_RUN_STATE: {
                    TaskRunPeriodStatusChange taskRunPeriodStatusChange = (TaskRunPeriodStatusChange) journal.getData();
                    globalStateMgr.getTaskManager().replayAlterRunningTaskRunProgress(
                            taskRunPeriodStatusChange.getTaskRunProgressMap());
                    break;
                }
                case OperationType.OP_ARCHIVE_TASK_RUNS: {
                    ArchiveTaskRunsLog log = (ArchiveTaskRunsLog) journal.getData();
                    globalStateMgr.getTaskManager().replayArchiveTaskRuns(log);
                    break;
                }
                case OperationType.OP_CREATE_SMALL_FILE:
                case OperationType.OP_CREATE_SMALL_FILE_V2: {
                    SmallFile smallFile = (SmallFile) journal.getData();
                    globalStateMgr.getSmallFileMgr().replayCreateFile(smallFile);
                    break;
                }
                case OperationType.OP_DROP_SMALL_FILE:
                case OperationType.OP_DROP_SMALL_FILE_V2: {
                    SmallFile smallFile = (SmallFile) journal.getData();
                    globalStateMgr.getSmallFileMgr().replayRemoveFile(smallFile);
                    break;
                }
                case OperationType.OP_ALTER_JOB_V2: {
                    AlterJobV2 alterJob = (AlterJobV2) journal.getData();
                    switch (alterJob.getType()) {
                        case ROLLUP:
                            globalStateMgr.getRollupHandler().replayAlterJobV2(alterJob);
                            break;
                        case SCHEMA_CHANGE:
                        case OPTIMIZE:
                            globalStateMgr.getSchemaChangeHandler().replayAlterJobV2(alterJob);
                            break;
                        default:
                            break;
                    }
                    break;
                }
                case OperationType.OP_BATCH_ADD_ROLLUP:
                case OperationType.OP_BATCH_ADD_ROLLUP_V2: {
                    BatchAlterJobPersistInfo batchAlterJobV2 = (BatchAlterJobPersistInfo) journal.getData();
                    for (AlterJobV2 alterJobV2 : batchAlterJobV2.getAlterJobV2List()) {
                        globalStateMgr.getRollupHandler().replayAlterJobV2(alterJobV2);
                    }
                    break;
                }
                case OperationType.OP_MODIFY_DISTRIBUTION_TYPE:
                case OperationType.OP_MODIFY_DISTRIBUTION_TYPE_V2: {
                    TableInfo tableInfo = (TableInfo) journal.getData();
                    globalStateMgr.getLocalMetastore().replayConvertDistributionType(tableInfo);
                    break;
                }
                case OperationType.OP_DYNAMIC_PARTITION:
                case OperationType.OP_MODIFY_IN_MEMORY:
                case OperationType.OP_SET_FORBIDDEN_GLOBAL_DICT:
                case OperationType.OP_MODIFY_REPLICATION_NUM:
                case OperationType.OP_MODIFY_WRITE_QUORUM:
                case OperationType.OP_MODIFY_REPLICATED_STORAGE:
                case OperationType.OP_MODIFY_BUCKET_SIZE:
                case OperationType.OP_MODIFY_MUTABLE_BUCKET_NUM:
                case OperationType.OP_MODIFY_BINLOG_AVAILABLE_VERSION:
                case OperationType.OP_MODIFY_BINLOG_CONFIG:
                case OperationType.OP_MODIFY_ENABLE_PERSISTENT_INDEX:
                case OperationType.OP_MODIFY_PRIMARY_INDEX_CACHE_EXPIRE_SEC:
                case OperationType.OP_ALTER_TABLE_PROPERTIES:
                case OperationType.OP_MODIFY_TABLE_CONSTRAINT_PROPERTY: {
                    ModifyTablePropertyOperationLog modifyTablePropertyOperationLog =
                            (ModifyTablePropertyOperationLog) journal.getData();
                    globalStateMgr.getLocalMetastore().replayModifyTableProperty(opCode, modifyTablePropertyOperationLog);
                    break;
                }
                case OperationType.OP_REPLACE_TEMP_PARTITION: {
                    ReplacePartitionOperationLog replaceTempPartitionLog =
                            (ReplacePartitionOperationLog) journal.getData();
                    globalStateMgr.getLocalMetastore().replayReplaceTempPartition(replaceTempPartitionLog);
                    break;
                }
                case OperationType.OP_INSTALL_PLUGIN: {
                    PluginInfo pluginInfo = (PluginInfo) journal.getData();
                    try {
                        globalStateMgr.getPluginMgr().replayLoadDynamicPlugin(pluginInfo);
                    } catch (Exception e) {
                        LOG.warn("replay install plugin failed.", e);
                    }

                    break;
                }
                case OperationType.OP_UNINSTALL_PLUGIN: {
                    PluginInfo pluginInfo = (PluginInfo) journal.getData();
                    try {
                        globalStateMgr.getPluginMgr().uninstallPlugin(pluginInfo.getName());
                    } catch (Exception e) {
                        LOG.warn("replay uninstall plugin failed.", e);
                    }
                    break;
                }
                case OperationType.OP_SET_REPLICA_STATUS: {
                    SetReplicaStatusOperationLog log = (SetReplicaStatusOperationLog) journal.getData();
                    globalStateMgr.getLocalMetastore().replaySetReplicaStatus(log);
                    break;
                }
                case OperationType.OP_REMOVE_ALTER_JOB_V2: {
                    RemoveAlterJobV2OperationLog log = (RemoveAlterJobV2OperationLog) journal.getData();
                    switch (log.getType()) {
                        case ROLLUP:
                            globalStateMgr.getRollupHandler().replayRemoveAlterJobV2(log);
                            break;
                        case SCHEMA_CHANGE:
                            globalStateMgr.getSchemaChangeHandler().replayRemoveAlterJobV2(log);
                            break;
                        case OPTIMIZE:
                            globalStateMgr.getSchemaChangeHandler().replayRemoveAlterJobV2(log);
                            break;
                        default:
                            break;
                    }
                    break;
                }
                case OperationType.OP_ALTER_ROUTINE_LOAD_JOB: {
                    AlterRoutineLoadJobOperationLog log = (AlterRoutineLoadJobOperationLog) journal.getData();
                    globalStateMgr.getRoutineLoadMgr().replayAlterRoutineLoadJob(log);
                    break;
                }
                case OperationType.OP_ALTER_LOAD_JOB: {
                    AlterLoadJobOperationLog log = (AlterLoadJobOperationLog) journal.getData();
                    globalStateMgr.getLoadMgr().replayAlterLoadJob(log);
                    break;
                }
                case OperationType.OP_GLOBAL_VARIABLE_V2: {
                    GlobalVarPersistInfo info = (GlobalVarPersistInfo) journal.getData();
                    VariableMgr.replayGlobalVariableV2(info);
                    break;
                }
                case OperationType.OP_SWAP_TABLE: {
                    SwapTableOperationLog log = (SwapTableOperationLog) journal.getData();
                    globalStateMgr.getAlterJobMgr().replaySwapTable(log);
                    break;
                }
                case OperationType.OP_ADD_ANALYZER_JOB: {
                    NativeAnalyzeJob nativeAnalyzeJob = (NativeAnalyzeJob) journal.getData();
                    globalStateMgr.getAnalyzeMgr().replayAddAnalyzeJob(nativeAnalyzeJob);
                    break;
                }
                case OperationType.OP_REMOVE_ANALYZER_JOB: {
                    NativeAnalyzeJob nativeAnalyzeJob = (NativeAnalyzeJob) journal.getData();
                    globalStateMgr.getAnalyzeMgr().replayRemoveAnalyzeJob(nativeAnalyzeJob);
                    break;
                }
                case OperationType.OP_ADD_ANALYZE_STATUS: {
                    NativeAnalyzeStatus analyzeStatus = (NativeAnalyzeStatus) journal.getData();
                    globalStateMgr.getAnalyzeMgr().replayAddAnalyzeStatus(analyzeStatus);
                    break;
                }
                case OperationType.OP_REMOVE_ANALYZE_STATUS: {
                    NativeAnalyzeStatus analyzeStatus = (NativeAnalyzeStatus) journal.getData();
                    globalStateMgr.getAnalyzeMgr().replayRemoveAnalyzeStatus(analyzeStatus);
                    break;
                }
                case OperationType.OP_ADD_EXTERNAL_ANALYZE_STATUS: {
                    ExternalAnalyzeStatus analyzeStatus = (ExternalAnalyzeStatus) journal.getData();
                    globalStateMgr.getAnalyzeMgr().replayAddAnalyzeStatus(analyzeStatus);
                    break;
                }
                case OperationType.OP_REMOVE_EXTERNAL_ANALYZE_STATUS: {
                    ExternalAnalyzeStatus analyzeStatus = (ExternalAnalyzeStatus) journal.getData();
                    globalStateMgr.getAnalyzeMgr().replayRemoveAnalyzeStatus(analyzeStatus);
                    break;
                }
                case OperationType.OP_ADD_EXTERNAL_ANALYZER_JOB: {
                    ExternalAnalyzeJob externalAnalyzeJob = (ExternalAnalyzeJob) journal.getData();
                    globalStateMgr.getAnalyzeMgr().replayAddAnalyzeJob(externalAnalyzeJob);
                    break;
                }
                case OperationType.OP_REMOVE_EXTERNAL_ANALYZER_JOB: {
                    ExternalAnalyzeJob externalAnalyzeJob = (ExternalAnalyzeJob) journal.getData();
                    globalStateMgr.getAnalyzeMgr().replayRemoveAnalyzeJob(externalAnalyzeJob);
                    break;
                }
                case OperationType.OP_ADD_BASIC_STATS_META: {
                    BasicStatsMeta basicStatsMeta = (BasicStatsMeta) journal.getData();
                    globalStateMgr.getAnalyzeMgr().replayAddBasicStatsMeta(basicStatsMeta);
                    // The follower replays the stats meta log, indicating that the master has re-completed
                    // statistic, and the follower's should refresh cache here.
                    // We don't need to refresh statistics when checkpointing
                    if (!GlobalStateMgr.isCheckpointThread()) {
                        globalStateMgr.getAnalyzeMgr().refreshBasicStatisticsCache(basicStatsMeta.getDbId(),
                                basicStatsMeta.getTableId(), basicStatsMeta.getColumns(), true);
                    }
                    break;
                }
                case OperationType.OP_REMOVE_BASIC_STATS_META: {
                    BasicStatsMeta basicStatsMeta = (BasicStatsMeta) journal.getData();
                    globalStateMgr.getAnalyzeMgr().replayRemoveBasicStatsMeta(basicStatsMeta);
                    break;
                }
                case OperationType.OP_ADD_HISTOGRAM_STATS_META: {
                    HistogramStatsMeta histogramStatsMeta = (HistogramStatsMeta) journal.getData();
                    globalStateMgr.getAnalyzeMgr().replayAddHistogramStatsMeta(histogramStatsMeta);
                    // The follower replays the stats meta log, indicating that the master has re-completed
                    // statistic, and the follower's should expire cache here.
                    // We don't need to refresh statistics when checkpointing
                    if (!GlobalStateMgr.isCheckpointThread()) {
                        globalStateMgr.getAnalyzeMgr().refreshHistogramStatisticsCache(
                                histogramStatsMeta.getDbId(), histogramStatsMeta.getTableId(),
                                Lists.newArrayList(histogramStatsMeta.getColumn()), true);
                    }
                    break;
                }
                case OperationType.OP_REMOVE_HISTOGRAM_STATS_META: {
                    HistogramStatsMeta histogramStatsMeta = (HistogramStatsMeta) journal.getData();
                    globalStateMgr.getAnalyzeMgr().replayRemoveHistogramStatsMeta(histogramStatsMeta);
                    break;
                }
                case OperationType.OP_ADD_EXTERNAL_BASIC_STATS_META: {
                    ExternalBasicStatsMeta basicStatsMeta = (ExternalBasicStatsMeta) journal.getData();
                    globalStateMgr.getAnalyzeMgr().replayAddExternalBasicStatsMeta(basicStatsMeta);
                    // The follower replays the stats meta log, indicating that the master has re-completed
                    // statistic, and the follower's should refresh cache here.
                    // We don't need to refresh statistics when checkpointing
                    if (!GlobalStateMgr.isCheckpointThread()) {
                        globalStateMgr.getAnalyzeMgr().refreshConnectorTableBasicStatisticsCache(
                                basicStatsMeta.getCatalogName(),
                                basicStatsMeta.getDbName(), basicStatsMeta.getTableName(),
                                basicStatsMeta.getColumns(), true);
                    }
                    break;
                }
                case OperationType.OP_REMOVE_EXTERNAL_BASIC_STATS_META: {
                    ExternalBasicStatsMeta basicStatsMeta = (ExternalBasicStatsMeta) journal.getData();
                    globalStateMgr.getAnalyzeMgr().replayRemoveExternalBasicStatsMeta(basicStatsMeta);
                    break;
                }
                case OperationType.OP_ADD_EXTERNAL_HISTOGRAM_STATS_META: {
                    ExternalHistogramStatsMeta histogramStatsMeta = (ExternalHistogramStatsMeta) journal.getData();
                    globalStateMgr.getAnalyzeMgr().replayAddExternalHistogramStatsMeta(histogramStatsMeta);
                    // The follower replays the stats meta log, indicating that the master has re-completed
                    // statistic, and the follower's should expire cache here.
                    // We don't need to refresh statistics when checkpointing
                    if (!GlobalStateMgr.isCheckpointThread()) {
                        globalStateMgr.getAnalyzeMgr().refreshConnectorTableHistogramStatisticsCache(
                                histogramStatsMeta.getCatalogName(), histogramStatsMeta.getDbName(),
                                histogramStatsMeta.getTableName(),
                                Lists.newArrayList(histogramStatsMeta.getColumn()), true);
                    }
                    break;
                }
                case OperationType.OP_REMOVE_EXTERNAL_HISTOGRAM_STATS_META: {
                    ExternalHistogramStatsMeta histogramStatsMeta = (ExternalHistogramStatsMeta) journal.getData();
                    globalStateMgr.getAnalyzeMgr().replayRemoveExternalHistogramStatsMeta(histogramStatsMeta);
                    break;
                }
                case OperationType.OP_MODIFY_HIVE_TABLE_COLUMN: {
                    ModifyTableColumnOperationLog modifyTableColumnOperationLog =
                            (ModifyTableColumnOperationLog) journal.getData();
                    globalStateMgr.getLocalMetastore().replayModifyHiveTableColumn(opCode, modifyTableColumnOperationLog);
                    break;
                }
                case OperationType.OP_CREATE_CATALOG: {
                    Catalog catalog = (Catalog) journal.getData();
                    globalStateMgr.getCatalogMgr().replayCreateCatalog(catalog);
                    break;
                }
                case OperationType.OP_DROP_CATALOG: {
                    DropCatalogLog dropCatalogLog = (DropCatalogLog) journal.getData();
                    globalStateMgr.getCatalogMgr().replayDropCatalog(dropCatalogLog);
                    break;
                }
                case OperationType.OP_ALTER_CATALOG:
                    AlterCatalogLog alterCatalogLog = (AlterCatalogLog) journal.getData();
                    globalStateMgr.getCatalogMgr().replayAlterCatalog(alterCatalogLog);
                    break;
                case OperationType.OP_CREATE_INSERT_OVERWRITE: {
                    CreateInsertOverwriteJobLog jobInfo = (CreateInsertOverwriteJobLog) journal.getData();
                    globalStateMgr.getInsertOverwriteJobMgr().replayCreateInsertOverwrite(jobInfo);
                    break;
                }
                case OperationType.OP_INSERT_OVERWRITE_STATE_CHANGE: {
                    InsertOverwriteStateChangeInfo stateChangeInfo = (InsertOverwriteStateChangeInfo) journal.getData();
                    globalStateMgr.getInsertOverwriteJobMgr().replayInsertOverwriteStateChange(stateChangeInfo);
                    break;
                }
                case OperationType.OP_ADD_UNUSED_SHARD:
                case OperationType.OP_DELETE_UNUSED_SHARD:
                    // Deprecated: Nothing to do
                    break;
                case OperationType.OP_STARMGR: {
                    StarMgrJournal j = (StarMgrJournal) journal.getData();
                    StarMgrServer.getCurrentState().getStarMgr().replay(j.getJournal());
                    break;
                }
                case OperationType.OP_CREATE_USER_V2: {
                    CreateUserInfo info = (CreateUserInfo) journal.getData();
                    globalStateMgr.getAuthenticationMgr().replayCreateUser(
                            info.getUserIdentity(),
                            info.getAuthenticationInfo(),
                            info.getUserProperty(),
                            info.getUserPrivilegeCollection(),
                            info.getPluginId(),
                            info.getPluginVersion());
                    break;
                }
                case OperationType.OP_UPDATE_USER_PRIVILEGE_V2: {
                    UserPrivilegeCollectionInfo info = (UserPrivilegeCollectionInfo) journal.getData();
                    globalStateMgr.getAuthorizationMgr().replayUpdateUserPrivilegeCollection(
                            info.getUserIdentity(),
                            info.getPrivilegeCollection(),
                            info.getPluginId(),
                            info.getPluginVersion());
                    break;
                }
                case OperationType.OP_ALTER_USER_V2: {
                    AlterUserInfo info = (AlterUserInfo) journal.getData();
                    globalStateMgr.getAuthenticationMgr().replayAlterUser(
                            info.getUserIdentity(), info.getAuthenticationInfo(), info.getProperties());
                    break;
                }
                case OperationType.OP_UPDATE_USER_PROP_V2:
                case OperationType.OP_UPDATE_USER_PROP_V3: {
                    UserPropertyInfo info = (UserPropertyInfo) journal.getData();
                    globalStateMgr.getAuthenticationMgr().replayUpdateUserProperty(info);
                    break;
                }
                case OperationType.OP_DROP_USER_V2:
                case OperationType.OP_DROP_USER_V3: {
                    UserIdentity userIdentity = (UserIdentity) journal.getData();
                    globalStateMgr.getAuthenticationMgr().replayDropUser(userIdentity);
                    break;
                }
                case OperationType.OP_UPDATE_ROLE_PRIVILEGE_V2: {
                    RolePrivilegeCollectionInfo info = (RolePrivilegeCollectionInfo) journal.getData();
                    globalStateMgr.getAuthorizationMgr().replayUpdateRolePrivilegeCollection(info);
                    break;
                }
                case OperationType.OP_DROP_ROLE_V2: {
                    RolePrivilegeCollectionInfo info = (RolePrivilegeCollectionInfo) journal.getData();
                    globalStateMgr.getAuthorizationMgr().replayDropRole(info);
                    break;
                }
                case OperationType.OP_AUTH_UPGRADE_V2: {
                    // for compatibility reason, just ignore the auth upgrade log
                    break;
                }
                case OperationType.OP_MV_JOB_STATE: {
                    MVMaintenanceJob job = (MVMaintenanceJob) journal.getData();
                    GlobalStateMgr.getCurrentState().getMaterializedViewMgr().replay(job);
                    break;
                }
                case OperationType.OP_MV_EPOCH_UPDATE: {
                    MVEpoch epoch = (MVEpoch) journal.getData();
                    GlobalStateMgr.getCurrentState().getMaterializedViewMgr().replayEpoch(epoch);
                    break;
                }
                case OperationType.OP_MODIFY_TABLE_ADD_OR_DROP_COLUMNS: {
                    final TableAddOrDropColumnsInfo info = (TableAddOrDropColumnsInfo) journal.getData();
                    globalStateMgr.getSchemaChangeHandler().replayModifyTableAddOrDrop(info);
                    break;
                }
                case OperationType.OP_SET_DEFAULT_STORAGE_VOLUME: {
                    SetDefaultStorageVolumeLog log = (SetDefaultStorageVolumeLog) journal.getData();
                    globalStateMgr.getStorageVolumeMgr().replaySetDefaultStorageVolume(log);
                    break;
                }
                case OperationType.OP_CREATE_STORAGE_VOLUME: {
                    StorageVolume sv = (StorageVolume) journal.getData();
                    globalStateMgr.getStorageVolumeMgr().replayCreateStorageVolume(sv);
                    break;
                }
                case OperationType.OP_UPDATE_STORAGE_VOLUME: {
                    StorageVolume sv = (StorageVolume) journal.getData();
                    globalStateMgr.getStorageVolumeMgr().replayUpdateStorageVolume(sv);
                    break;
                }
                case OperationType.OP_DROP_STORAGE_VOLUME: {
                    DropStorageVolumeLog log = (DropStorageVolumeLog) journal.getData();
                    globalStateMgr.getStorageVolumeMgr().replayDropStorageVolume(log);
                    break;
                }
                case OperationType.OP_PIPE: {
                    PipeOpEntry opEntry = (PipeOpEntry) journal.getData();
                    globalStateMgr.getPipeManager().getRepo().replay(opEntry);
                    break;
                }
                case OperationType.OP_CREATE_DICTIONARY: {
                    Dictionary dictionary = (Dictionary) journal.getData();
                    globalStateMgr.getDictionaryMgr().replayCreateDictionary(dictionary);
                    break;
                }
                case OperationType.OP_DROP_DICTIONARY: {
                    DropDictionaryInfo dropInfo = (DropDictionaryInfo) journal.getData();
                    globalStateMgr.getDictionaryMgr().replayDropDictionary(dropInfo.getDictionaryName());
                    break;
                }
                case OperationType.OP_MODIFY_DICTIONARY_MGR: {
                    DictionaryMgrInfo modifyInfo = (DictionaryMgrInfo) journal.getData();
                    globalStateMgr.getDictionaryMgr().replayModifyDictionaryMgr(modifyInfo);
                    break;
                }
                case OperationType.OP_DECOMMISSION_DISK: {
                    DecommissionDiskInfo info = (DecommissionDiskInfo) journal.getData();
                    globalStateMgr.getNodeMgr().getClusterInfo().replayDecommissionDisks(info);
                    break;
                }
                case OperationType.OP_CANCEL_DECOMMISSION_DISK: {
                    CancelDecommissionDiskInfo info = (CancelDecommissionDiskInfo) journal.getData();
                    globalStateMgr.getNodeMgr().getClusterInfo().replayCancelDecommissionDisks(info);
                    break;
                }
                case OperationType.OP_DISABLE_DISK: {
                    DisableDiskInfo info = (DisableDiskInfo) journal.getData();
                    globalStateMgr.getNodeMgr().getClusterInfo().replayDisableDisks(info);
                    break;
                }
                case OperationType.OP_CANCEL_DISABLE_DISK: {
                    CancelDisableDiskInfo info = (CancelDisableDiskInfo) journal.getData();
                    globalStateMgr.getNodeMgr().getClusterInfo().replayCancelDisableDisks(info);
                    break;
                }
                case OperationType.OP_REPLICATION_JOB: {
                    ReplicationJobLog replicationJobLog = (ReplicationJobLog) journal.getData();
                    globalStateMgr.getReplicationMgr().replayReplicationJob(replicationJobLog.getReplicationJob());
                    break;
                }
                case OperationType.OP_RECOVER_PARTITION_VERSION: {
                    PartitionVersionRecoveryInfo info = (PartitionVersionRecoveryInfo) journal.getData();
                    GlobalStateMgr.getCurrentState().getMetaRecoveryDaemon().recoverPartitionVersion(info);
                    break;
                }
                default: {
                    if (Config.ignore_unknown_log_id) {
                        LOG.warn("UNKNOWN Operation Type {}", opCode);
                    } else {
                        throw new IOException("UNKNOWN Operation Type " + opCode);
                    }
                }
            }
        } catch (Exception e) {
            JournalInconsistentException exception =
                    new JournalInconsistentException(opCode, "failed to load journal type " + opCode);
            exception.initCause(e);
            throw exception;
        }
    }
@Test
    public void testLoadJournalException(@Mocked GlobalStateMgr globalStateMgr) {
        JournalEntity journal = new JournalEntity();
        journal.setOpCode(OperationType.OP_SAVE_NEXTID);
        // set data to null, and it will throw NPE in loadJournal()
        journal.setData(null);

        EditLog editLog = new EditLog(null);
        new Expectations() {
            {
                globalStateMgr.getEditLog();
                result = editLog;
            }
        };

        try {
            GlobalStateMgr.getCurrentState().getEditLog().loadJournal(GlobalStateMgr.getCurrentState(), journal);
        } catch (JournalInconsistentException e) {
            Assert.assertEquals(OperationType.OP_SAVE_NEXTID, e.getOpCode());
        }
    }