private HudiSinkDTO getHudiInfo(SinkInfo sinkInfo) {
        HudiSinkDTO hudiInfo = HudiSinkDTO.getFromJson(sinkInfo.getExtParams());

        // read uri from data node if not supplied by user
        if (StringUtils.isBlank(hudiInfo.getCatalogUri())
                && CATALOG_TYPE_HIVE.equals(hudiInfo.getCatalogType())) {
            String dataNodeName = sinkInfo.getDataNodeName();
            Preconditions.expectNotBlank(dataNodeName, ErrorCodeEnum.INVALID_PARAMETER,
                    "Hudi catalog uri not specified and data node is empty");
            HudiDataNodeInfo dataNodeInfo = (HudiDataNodeInfo) dataNodeHelper.getDataNodeInfo(
                    dataNodeName, sinkInfo.getSinkType());
            CommonBeanUtils.copyProperties(dataNodeInfo, hudiInfo);
            hudiInfo.setCatalogUri(dataNodeInfo.getUrl());
        }

        hudiInfo.setDataPath(hudiInfo.getWarehouse() + "/" + hudiInfo.getDbName() + ".db/" + hudiInfo.getTableName());
        return hudiInfo;
    }
Create Hudi table according to the sink config
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/hudi/HudiResourceOperator.java