将DataSet数据发送到elasticsearch

正在尝试使用新的elasticsearch连接器将一些数据从DataSet发送到elasticsearch，但除了数据流结构之外我找不到任何资源：

https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/elasticsearch.html

我的数据集是行的数据集（来自sql查询），这是内容：

199947,6
199958,3
199964,2
199985,2
我创建了一个实现ElasticsearchSinkFunction的静态嵌套类：

public static class NumberOfTransactionsByBlocks implements ElasticsearchSinkFunction {

public void process(Row element, RuntimeContext ctx, RequestIndexer indexer) {
    indexer.add(createIndexRequest(element));

}

public IndexRequest createIndexRequest(Row element) {
    Map<String, String> json = new HashMap<>();
    json.put("block_number", element.getField(0).toString());
    json.put("numberOfTransactions", element.getField(1).toString());

    return Requests.indexRequest()
            .index("nbOfTransactionsByBlocks")
            .type("count-transactions")
            .source(json);
}

}
然后我的问题是我不知道如何发送我的内部类的实例...

DataSet data = tableEnv.toDataSet(sqlResult, Row.class);
List httpHosts = new ArrayList<>();
httpHosts.add(new HttpHost("127.0.0.1", 9200, "http"));
httpHosts.add(new HttpHost("10.2.3.1", 9200, "http"));

Map config = new HashMap<>();
config.put("bulk.flush.max.actions", "1"); // flush inserts after every event
config.put("cluster.name", "elasticsearch"); // default cluster name

data.output(new ElasticsearchSink<>(config, httpHosts, new NumberOfTransactionsByBlocks()));
我在实例化ElasticsearchSink时遇到错误，它说：

不能推断论点

但是，当我指定类型（行）时，它说：

ElasticsearchSink（java.util.Map，java.util.List，org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction，org.apache.flink.streaming.connectors.elasticsearch.ActionRequestFailureHandler，org.apache.flink.streaming。 connectors.elasticsearch6.RestClientFactory）'在'org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink'中拥有私有访问权限
到底出了什么问题？