@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());
        }
        ElasticsearchSinkRequest sinkRequest = (ElasticsearchSinkRequest) request;
        try {
            ElasticsearchSinkDTO dto = ElasticsearchSinkDTO.getFromRequest(sinkRequest, targetEntity.getExtParams());

            InlongStreamEntity stream = inlongStreamEntityMapper
                    .selectByIdentifier(request.getInlongGroupId(), request.getInlongStreamId());
            dto.setSeparator(String.valueOf((char) (Integer.parseInt(stream.getDataSeparator()))));

            InlongStreamExtParam streamExt =
                    JsonUtils.parseObject(stream.getExtParams(), InlongStreamExtParam.class);
            dto.setFieldOffset(streamExt.getExtendedFieldSize());

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