@Override
    public void process(RowData element, RuntimeContext ctx, RequestIndexer<Request> indexer) {
        final byte[] document;
        try {
            document = serializationSchema.serialize(element);
        } catch (Exception e) {
            LOGGER.error(String.format("Serialize error, raw data: %s", element), e);
            dirtySinkHelper.invoke(element, DirtyType.SERIALIZE_ERROR, e);
            if (sinkMetricData != null) {
                sinkMetricData.invokeDirty(1, element.toString().getBytes(StandardCharsets.UTF_8).length);
            }
            return;
        }
        final String key;
        try {
            key = createKey.apply(element);
        } catch (Exception e) {
            LOGGER.error(String.format("Generate index id error, raw data: %s", element), e);
            dirtySinkHelper.invoke(element, DirtyType.INDEX_ID_GENERATE_ERROR, e);
            if (sinkMetricData != null) {
                sinkMetricData.invokeDirty(1, document.length);
            }
            return;
        }
        final String index;
        try {
            index = indexGenerator.generate(element);
        } catch (Exception e) {
            LOGGER.error(String.format("Generate index error, raw data: %s", element), e);
            dirtySinkHelper.invoke(element, DirtyType.INDEX_GENERATE_ERROR, e);
            if (sinkMetricData != null) {
                sinkMetricData.invokeDirty(1, document.length);
            }
            return;
        }
        addDocument(element, key, index, document, indexer);
    }
Sink function for converting upserts into Elasticsearch ActionRequests.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/table/ElasticsearchSinkFunctionBase.java