public void createResource(CreateResourceStmt stmt) throws DdlException {
        Resource resource = Resource.fromStmt(stmt);

        this.writeLock();
        try {
            String resourceName = stmt.getResourceName();
            String typeName = resource.getType().name().toLowerCase(Locale.ROOT);
            if (resource.needMappingCatalog()) {
                try {
                    String catalogName = getResourceMappingCatalogName(resourceName, typeName);
                    GlobalStateMgr.getCurrentState().getCatalogMgr().createCatalog(
                            typeName, catalogName, "mapping " + typeName + " catalog", stmt.getProperties());
                } catch (Exception e) {
                    LOG.error("Failed to create mapping {} catalog {} failed", typeName, resource, e);
                    throw new DdlException("Failed to create mapping catalog " + resourceName + " failed, msg: " +
                            e.getMessage());
                }
            }

            if (nameToResource.putIfAbsent(resourceName, resource) != null) {
                throw new DdlException("Resource(" + resourceName + ") already exist");
            }

            // log add
            GlobalStateMgr.getCurrentState().getEditLog().logCreateResource(resource);
            LOG.info("create resource success. resource: {}", resource);
        } finally {
            this.writeUnLock();
        }
    }
@Test(expected = DdlException.class)
    public void testAddResourceExist(@Injectable BrokerMgr brokerMgr, @Injectable EditLog editLog,
                                     @Mocked GlobalStateMgr globalStateMgr)
            throws UserException {
        ResourceMgr mgr = new ResourceMgr();

        // add
        CreateResourceStmt stmt = addSparkResource(mgr, brokerMgr, editLog, globalStateMgr);

        // add again
        mgr.createResource(stmt);
    }