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

        DorisSinkDTO dto = DorisSinkDTO.getFromJson(entity.getExtParams());
        if (StringUtils.isBlank(dto.getFeNodes())) {
            if (StringUtils.isBlank(entity.getDataNodeName())) {
                throw new BusinessException(ErrorCodeEnum.SINK_INFO_INCORRECT,
                        "doris fe nodes unspecified and data node is blank");
            }
            DorisDataNodeInfo dataNodeInfo = (DorisDataNodeInfo) dataNodeHelper.getDataNodeInfo(
                    entity.getDataNodeName(), entity.getSinkType());
            dto.setFeNodes(dataNodeInfo.getUrl());
            dto.setUsername(dataNodeInfo.getUsername());
            dto.setPassword(dataNodeInfo.getToken());
            CommonBeanUtils.copyProperties(dataNodeInfo, dto, true);
        }

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