@Override
    @Transactional(rollbackFor = Throwable.class, isolation = Isolation.REPEATABLE_READ)
    public void updateOpt(SourceRequest request, Integer streamStatus, Integer groupMode, String operator) {
        StreamSourceEntity entity = sourceMapper.selectByIdForUpdate(request.getId());
        if (entity == null) {
            throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_NOT_FOUND,
                    String.format("not found source record by id=%d", request.getId()));
        }
        if (SourceType.AUTO_PUSH.equals(entity.getSourceType())) {
            updateFieldOpt(entity, request.getFieldList());
            return;
        }
        boolean allowUpdate = InlongConstants.DATASYNC_REALTIME_MODE.equals(groupMode)
                || InlongConstants.DATASYNC_OFFLINE_MODE.equals(groupMode)
                || SourceStatus.ALLOWED_UPDATE.contains(entity.getStatus());
        if (!allowUpdate) {
            throw new BusinessException(ErrorCodeEnum.SOURCE_OPT_NOT_ALLOWED,
                    String.format(
                            "source=%s is not allowed to update, please wait until its changed to final status or stop / frozen / delete it firstly",
                            entity));
        }
        String errMsg = String.format("source has already updated with groupId=%s, streamId=%s, name=%s, curVersion=%s",
                request.getInlongGroupId(), request.getInlongStreamId(), request.getSourceName(), request.getVersion());
        if (!Objects.equals(entity.getVersion(), request.getVersion())) {
            throw new BusinessException(ErrorCodeEnum.CONFIG_EXPIRED, errMsg);
        }

        // source type cannot be changed
        if (!Objects.equals(entity.getSourceType(), request.getSourceType())) {
            throw new BusinessException(ErrorCodeEnum.INVALID_PARAMETER,
                    String.format("source type=%s cannot change to %s", entity.getSourceType(),
                            request.getSourceType()));
        }

        String groupId = request.getInlongGroupId();
        String streamId = request.getInlongStreamId();
        String sourceName = request.getSourceName();
        List<StreamSourceEntity> sourceList = sourceMapper.selectByRelatedId(groupId, streamId, sourceName);
        for (StreamSourceEntity sourceEntity : sourceList) {
            Integer sourceId = sourceEntity.getId();
            if (!Objects.equals(sourceId, request.getId())) {
                throw new BusinessException(ErrorCodeEnum.SOURCE_ALREADY_EXISTS,
                        String.format("source name=%s already exists with the groupId=%s streamId=%s", sourceName,
                                groupId, streamId));
            }
        }

        // setting updated parameters of stream source entity.
        setTargetEntity(request, entity);
        entity.setModifier(operator);
        entity.setPreviousStatus(entity.getStatus());

        // re-issue task if necessary
        if (InlongConstants.STANDARD_MODE.equals(groupMode)) {
            SourceStatus sourceStatus = SourceStatus.forCode(entity.getStatus());
            Integer nextStatus = entity.getStatus();
            if (StreamStatus.forCode(streamStatus).equals(StreamStatus.CONFIG_SUCCESSFUL)) {
                nextStatus = SourceStatus.TO_BE_ISSUED_RETRY.getCode();
            } else {
                switch (SourceStatus.forCode(entity.getStatus())) {
                    case SOURCE_NORMAL:
                    case HEARTBEAT_TIMEOUT:
                        nextStatus = SourceStatus.TO_BE_ISSUED_RETRY.getCode();
                        break;
                    case SOURCE_FAILED:
                        nextStatus = SourceStatus.SOURCE_NEW.getCode();
                        break;
                    default:
                        // others leave it be
                        break;
                }
            }
            entity.setStatus(nextStatus);
        }

        int rowCount = sourceMapper.updateByPrimaryKeySelective(entity);
        if (rowCount != InlongConstants.AFFECTED_ONE_ROW) {
            throw new BusinessException(ErrorCodeEnum.CONFIG_EXPIRED, errMsg);
        }
        updateFieldOpt(entity, request.getFieldList());
        LOGGER.debug("success to update source of type={}", request.getSourceType());
        if (StreamStatus.forCode(streamStatus).equals(StreamStatus.CONFIG_SUCCESSFUL)) {
            updateAgentTaskConfig(request, operator);
        }
    }
Setting the parameters of the latest entity.

@param request source request
@param targetEntity entity object which will set the new parameters.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractSourceOperator.java