@Override
    protected void setTargetEntity(SinkRequest request, StreamSinkEntity targetEntity) {
        if (!this.getSinkType().equals(request.getSinkType())) {
            throw new BusinessException(ErrorCodeEnum.SINK_TYPE_NOT_SUPPORT,
                    ErrorCodeEnum.SINK_TYPE_NOT_SUPPORT.getMessage() + ": " + getSinkType());
        }
        HudiSinkRequest sinkRequest = (HudiSinkRequest) request;

        String partitionKey = sinkRequest.getPartitionKey();
        String primaryKey = sinkRequest.getPrimaryKey();
        boolean primaryKeyExist = StringUtils.isNotBlank(primaryKey);
        boolean partitionKeyExist = StringUtils.isNotBlank(partitionKey);
        if (primaryKeyExist || partitionKeyExist) {
            Set<String> fieldNames = sinkRequest.getSinkFieldList().stream().map(SinkField::getFieldName)
                    .collect(Collectors.toSet());
            if (partitionKeyExist) {
                List<String> partitionKeys = Arrays.asList(partitionKey.split(InlongConstants.COMMA));
                if (!CollectionUtils.isSubCollection(partitionKeys, fieldNames)) {
                    throw new BusinessException(ErrorCodeEnum.SINK_SAVE_FAILED,
                            String.format("The partitionKey(%s) must be included in the sinkFieldList(%s)",
                                    partitionKey, fieldNames));
                }
            }
            if (primaryKeyExist) {
                List<String> primaryKeys = Arrays.asList(primaryKey.split(InlongConstants.COMMA));
                if (!CollectionUtils.isSubCollection(primaryKeys, fieldNames)) {
                    throw new BusinessException(ErrorCodeEnum.SINK_SAVE_FAILED,
                            String.format("The primaryKey(%s) must be included in the sinkFieldList(%s)",
                                    primaryKey, fieldNames));
                }
            }
        }

        try {
            HudiSinkDTO dto = HudiSinkDTO.getFromRequest(sinkRequest, targetEntity.getExtParams());
            targetEntity.setExtParams(objectMapper.writeValueAsString(dto));
        } catch (Exception e) {
            throw new BusinessException(ErrorCodeEnum.SINK_SAVE_FAILED,
                    String.format("serialize extParams of Hudi SinkDTO failure: %s", e.getMessage()));
        }
    }
Hudi sink operator, such as save or update hudi field, etc.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/hudi/HudiSinkOperator.java