@Override
    public StreamSink getFromEntity(StreamSinkEntity entity) {
        IcebergSink sink = new IcebergSink();
        if (entity == null) {
            return sink;
        }

        IcebergSinkDTO dto = IcebergSinkDTO.getFromJson(entity.getExtParams());
        if (StringUtils.isBlank(dto.getCatalogUri()) && CATALOG_TYPE_HIVE.equals(dto.getCatalogType())) {
            if (StringUtils.isBlank(entity.getDataNodeName())) {
                throw new BusinessException(ErrorCodeEnum.SINK_INFO_INCORRECT,
                        "iceberg catalog uri unspecified and data node is blank");
            }
            IcebergDataNodeInfo dataNodeInfo = (IcebergDataNodeInfo) dataNodeHelper.getDataNodeInfo(
                    entity.getDataNodeName(), entity.getSinkType());
            CommonBeanUtils.copyProperties(dataNodeInfo, dto, true);
            dto.setCatalogUri(dataNodeInfo.getUrl());
        }

        CommonBeanUtils.copyProperties(entity, sink, true);
        CommonBeanUtils.copyProperties(dto, sink, true);
        List<SinkField> sinkFields = getSinkFields(entity.getId());
        sink.setSinkFieldList(sinkFields);
        return sink;
    }
Iceberg sink operator, such as save or update iceberg field, etc.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/iceberg/IcebergSinkOperator.java