@Override
    public void updateOpt(SinkRequest request, SinkStatus nextStatus, String operator) {
        StreamSinkEntity entity = sinkMapper.selectByPrimaryKey(request.getId());
        if (entity == null) {
            throw new BusinessException(ErrorCodeEnum.SINK_INFO_NOT_FOUND);
        }
        if (!Objects.equals(entity.getVersion(), request.getVersion())) {
            throw new BusinessException(ErrorCodeEnum.CONFIG_EXPIRED,
                    String.format("sink has already updated with groupId=%s, streamId=%s, name=%s, curVersion=%s",
                            request.getInlongGroupId(), request.getInlongStreamId(), request.getSinkName(),
                            request.getVersion()));
        }
        CommonBeanUtils.copyProperties(request, entity, true);
        setTargetEntity(request, entity);
        entity.setPreviousStatus(entity.getStatus());
        if (nextStatus != null) {
            entity.setStatus(nextStatus.getCode());
        }
        entity.setModifier(operator);
        int rowCount = sinkMapper.updateByIdSelective(entity);
        if (rowCount != InlongConstants.AFFECTED_ONE_ROW) {
            throw new BusinessException(ErrorCodeEnum.CONFIG_EXPIRED,
                    String.format("sink has already updated with groupId=%s, streamId=%s, name=%s, curVersion=%s",
                            request.getInlongGroupId(), request.getInlongStreamId(), request.getSinkName(),
                            request.getVersion()));
        }

        boolean onlyAdd = SinkStatus.CONFIG_SUCCESSFUL.getCode().equals(entity.getPreviousStatus());
        this.updateFieldOpt(onlyAdd, request);

        LOGGER.info("success to update sink of type={}", request.getSinkType());
    }
Getting the sink type.

@return sink type string.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/AbstractSinkOperator.java