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

        StarRocksSinkDTO dto = StarRocksSinkDTO.getFromJson(entity.getExtParams());
        if (StringUtils.isBlank(dto.getJdbcUrl())) {
            if (StringUtils.isBlank(entity.getDataNodeName())) {
                throw new BusinessException(ErrorCodeEnum.SINK_INFO_INCORRECT,
                        "starRocks jdbc url unspecified and data node is blank");
            }
            StarRocksDataNodeInfo dataNodeInfo = (StarRocksDataNodeInfo) dataNodeHelper.getDataNodeInfo(
                    entity.getDataNodeName(), entity.getSinkType());
            CommonBeanUtils.copyProperties(dataNodeInfo, dto, true);
            dto.setJdbcUrl(dataNodeInfo.getUrl());
            dto.setPassword(dataNodeInfo.getToken());
        }
        if (StringUtils.isBlank(dto.getLoadUrl())) {
            throw new BusinessException(ErrorCodeEnum.SINK_INFO_INCORRECT,
                    "StarRocks load url is blank");
        }
        if (StringUtils.isBlank(dto.getJdbcUrl())) {
            throw new BusinessException(ErrorCodeEnum.SINK_INFO_INCORRECT,
                    "StarRocks jdbc url is blank");
        }
        CommonBeanUtils.copyProperties(entity, sink, true);
        CommonBeanUtils.copyProperties(dto, sink, true);
        List<SinkField> sinkFields = super.getSinkFields(entity.getId());
        sink.setSinkFieldList(sinkFields);
        return sink;
    }
StarRocks sink operator, such as save or update StarRocks field, etc.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/starrocks/StarRocksSinkOperator.java