@Override
    protected void setTargetEntity(SinkRequest request, StreamSinkEntity targetEntity) {
        KuduSinkRequest sinkRequest = (KuduSinkRequest) request;
        try {
            KuduSinkDTO dto = KuduSinkDTO.getFromRequest(sinkRequest, targetEntity.getExtParams());
            targetEntity.setExtParams(objectMapper.writeValueAsString(dto));
        } catch (Exception e) {
            throw new BusinessException(ErrorCodeEnum.SINK_SAVE_FAILED,
                    String.format("serialize extParams of Kudu SinkDTO failure: %s", e.getMessage()));
        }

    }
Kudu sink operator, such as save or update Kudu field, etc.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/kudu/KuduSinkOperator.java