private IcebergSinkDTO getIcebergInfo(SinkInfo sinkInfo) {
        IcebergSinkDTO icebergInfo = IcebergSinkDTO.getFromJson(sinkInfo.getExtParams());

        // read uri from data node if not supplied by user
        if (StringUtils.isBlank(icebergInfo.getCatalogUri())
                && CATALOG_TYPE_HIVE.equals(icebergInfo.getCatalogType())) {
            String dataNodeName = sinkInfo.getDataNodeName();
            Preconditions.expectNotBlank(dataNodeName, ErrorCodeEnum.INVALID_PARAMETER,
                    "iceberg catalog uri not specified and data node is empty");
            IcebergDataNodeInfo dataNodeInfo = (IcebergDataNodeInfo) dataNodeHelper.getDataNodeInfo(
                    dataNodeName, sinkInfo.getSinkType());
            CommonBeanUtils.copyProperties(dataNodeInfo, icebergInfo);
            icebergInfo.setCatalogUri(dataNodeInfo.getUrl());
        }
        return icebergInfo;
    }
Create iceberg table according to the sink config
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/iceberg/IcebergResourceOperator.java