@Override
    public StreamSink getFromEntity(StreamSinkEntity entity) {
        ElasticsearchSink sink = new ElasticsearchSink();
        if (entity == null) {
            return sink;
        }

        ElasticsearchSinkDTO dto = ElasticsearchSinkDTO.getFromJson(entity.getExtParams());
        CommonBeanUtils.copyProperties(entity, sink, true);
        CommonBeanUtils.copyProperties(dto, sink, true);
        List<SinkField> sinkFields = getSinkFields(entity.getId());
        sink.setSinkFieldList(sinkFields);
        return sink;
    }
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