@Override
    public void process(RowData element, RuntimeContext ctx, RequestIndexer<Request> indexer) {
        JsonNode rootNode;
        // parse rootnode
        try {
            jsonDynamicSchemaFormat =
                    (JsonDynamicSchemaFormat) DynamicSchemaFormatFactory.getFormat(multipleFormat);
            rootNode = jsonDynamicSchemaFormat.deserialize(element.getBinary(0));
            // Ignore ddl change for now
            boolean isDDL = jsonDynamicSchemaFormat.extractDDLFlag(rootNode);
            if (isDDL) {
                LOGGER.error("ddl change unsupported");
                return;
            }
        } catch (Exception e) {
            LOGGER.error(String.format("deserialize error, raw data: %s", new String(element.getBinary(0))), e);
            throw new RuntimeException(e);
        }

        RowType rowType = jsonDynamicSchemaFormat.extractSchema(rootNode);
        RowData data = jsonDynamicSchemaFormat.extractRowData(rootNode, rowType).get(0);
        // generate the serialization schema
        serializationSchema = new JsonRowDataSerializationSchema(
                rowType, TimestampFormat.ISO_8601, MapNullKeyMode.LITERAL, "null", true);

        final byte[] document;
        try {
            // for multiple sink, need to update runtimeconverter to correct rowtype
            // for now create custom schema
            document = serializationSchema.serialize(data);
        } catch (Exception e) {
            LOGGER.error(String.format("Serialize error, raw data: %s", data), e);
            handleDirty(data, DirtyType.SERIALIZE_ERROR, e, null);
            return;
        }
        final String index;
        try {
            index = parseIndex(data, rootNode);
        } catch (Exception e) {
            LOGGER.error(String.format("Generate index error, raw data: %s", data), e);
            handleDirty(data, DirtyType.INDEX_GENERATE_ERROR, e, null);
            return;
        }
        final String key;
        try {
            // use uuid3 as key
            JsonNode physicalData = jsonDynamicSchemaFormat.getPhysicalData(rootNode);
            key = UUID.nameUUIDFromBytes(physicalData.toString().getBytes(StandardCharsets.UTF_8)).toString();
        } catch (Exception e) {
            LOGGER.error(String.format("Generate index id error, raw data: %s", data), e);
            handleDirty(data, DirtyType.INDEX_ID_GENERATE_ERROR, e, index);
            return;
        }
        // if the index is contained in errorset, then skip this record.
        if (errorSet.contains(index)) {
            return;
        }
        addDocument(data, 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/MultipleElasticsearchSinkFunctionBase.java