private RichSinkFunction createEsSinkFunction() {

        // check whether id fields is exists in columns
        List<String> filedNamesLists = Arrays.asList(fieldNames);
        if (idFiledNames != null && idFiledNames.size() != 0) {
            Preconditions.checkState(filedNamesLists.containsAll(idFiledNames), "elasticsearch7 type of id %s is should be exists in columns %s.", idFiledNames, filedNamesLists);
        }
        CustomerSinkFunc customerSinkFunc = new CustomerSinkFunc(index, index_definition, Arrays.asList(fieldNames), Arrays.asList(columnTypes), fieldTypes, idFiledNames);


        Map<String, String> userConfig = Maps.newHashMap();
        // This instructs the sink to emit after every element, otherwise they would be buffered
        userConfig.put(org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink.CONFIG_KEY_BULK_FLUSH_MAX_ACTIONS, "" + bulkFlushMaxActions);

        List<HttpHost> transports = esAddressList.stream()
                .map(address -> address.split(":"))
                .map(addressArray -> {
                    String host = addressArray[0].trim();
                    int port = addressArray.length > 1 ? Integer.valueOf(addressArray[1].trim()) : ES_DEFAULT_PORT;
                    return new HttpHost(host.trim(), port, ES_DEFAULT_SCHEMA);
                }).collect(Collectors.toList());
        ExtendEs7ApiCallBridge extendEs7ApiCallBridge = new ExtendEs7ApiCallBridge(transports, esTableInfo);

        return new MetricElasticsearch7Sink(userConfig, extendEs7ApiCallBridge, customerSinkFunc);
    }
@description:
@program: flink.sql
@author: lany
@create: 2021/01/04 17:20
elasticsearch7/elasticsearch7-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/ElasticsearchSink.java