private String parseIndex(RowData rowData, JsonNode rootNode)
            throws Exception {
        String index;
        // parse the index dynamically, put index generators
        try {
            String rawIndex = jsonDynamicSchemaFormat.parse(rootNode, indexPattern);
            if (!indexGeneratorMap.containsKey(rawIndex)) {
                TableSchema schema = tableSchemaFactory.getSchema();
                IndexGenerator indexGenerator = IndexGeneratorFactory.createIndexGenerator(rawIndex, schema);
                indexGeneratorMap.put(rawIndex, indexGenerator);
                indexGenerator.open();
            }
            index = indexGeneratorMap.get(rawIndex).generate(rowData);
        } catch (Exception e) {
            LOGGER.error(String.format("json parse error, raw data: %s", new String(rowData.getBinary(0))), e);
            throw e;
        }
        return index;
    }
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