public void replayCreateCatalog(Catalog catalog) throws DdlException {
        String type = catalog.getType();
        String catalogName = catalog.getName();
        Map<String, String> config = catalog.getConfig();
        CatalogConnector catalogConnector = null;

        try {
            if (Strings.isNullOrEmpty(type)) {
                throw new DdlException("Missing properties 'type'");
            }

            // skip unsupported connector type
            if (!ConnectorType.isSupport(type)) {
                LOG.error("Replay catalog [{}] encounter unknown catalog type [{}], ignore it", catalogName, type);
                return;
            }

            readLock();
            try {
                Preconditions.checkState(!catalogs.containsKey(catalogName), "Catalog '%s' already exists", catalogName);
            } finally {
                readUnlock();
            }

            Map<String, String> properties = catalog.getConfig();
            String serviceName = properties.get("ranger.plugin.hive.service.name");
            if (serviceName == null || serviceName.isEmpty()) {
                if (Config.access_control.equals("ranger")) {
                    Authorizer.getInstance().setAccessControl(catalogName, new RangerStarRocksAccessController());
                } else {
                    Authorizer.getInstance().setAccessControl(catalogName, new NativeAccessController());
                }
            } else {
                Authorizer.getInstance().setAccessControl(catalogName, new RangerHiveAccessController(serviceName));
            }

            try {
                catalogConnector = connectorMgr.createConnector(
                        new ConnectorContext(catalogName, type, config), true);
                if (catalogConnector == null) {
                    LOG.error("{} connector [{}] create failed.", type, catalogName);
                    throw new DdlException("connector create failed");
                }
            } catch (StarRocksConnectorException e) {
                LOG.error("connector create failed [{}], reason {}", catalogName, e.getMessage());
                throw new DdlException(String.format("connector create failed: %s", e.getMessage()));
            }

            writeLock();
            try {
                catalogs.put(catalogName, catalog);
            } finally {
                writeUnLock();
            }
        } catch (Exception e) {
            if (catalogConnector != null && connectorMgr.connectorExists(catalogName)) {
                connectorMgr.removeConnector(catalogName);
            }

            catalogs.remove(catalogName);
            throw e;
        }
    }
@Test
    public void testCreate() throws DdlException {
        CatalogMgr catalogMgr = GlobalStateMgr.getCurrentState().getCatalogMgr();
        Map<String, String> config = new HashMap<>();

        config.put("type", "paimon");
        final ExternalCatalog catalog = new ExternalCatalog(10000, "catalog_0", "", config);
        catalogMgr.replayCreateCatalog(catalog);
    }