private ElasticsearchSinkFunction<RowData, DocWriteRequest<?>> createSinkFunction(
            SerializationSchema<RowData> format,
            Context context) {
        if (multipleSink) {
            return new MultipleRowElasticsearchSinkFunction(
                    config.getDocumentType(), // this is deprecated in es 7+
                    format,
                    XContentType.JSON,
                    REQUEST_FACTORY,
                    KeyExtractor.createKeyExtractor(schema, config.getKeyDelimiter()),
                    RoutingExtractor.createRoutingExtractor(
                            schema, config.getRoutingField().orElse(null)),
                    dirtySinkHelper,
                    new TableSchemaFactory(schema.getFieldNames(), schema.getFieldTypes()),
                    multipleFormat,
                    indexPattern,
                    schemaUpdateExceptionPolicy);
        }

        return new RowElasticsearchSinkFunction(
                IndexGeneratorFactory.createIndexGenerator(config.getIndex(), schema),
                config.getDocumentType(), // this is deprecated in es 7+
                format,
                XContentType.JSON,
                REQUEST_FACTORY,
                KeyExtractor.createKeyExtractor(schema, config.getKeyDelimiter()),
                RoutingExtractor.createRoutingExtractor(
                        schema, config.getRoutingField().orElse(null)),
                dirtySinkHelper);
    }
A {@link DynamicTableSink} that describes how to create a {@link ElasticsearchSink} from a
logical description.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-6/src/main/java/org/apache/inlong/sort/elasticsearch6/table/Elasticsearch6DynamicSink.java