Scala, Elasticsearch the new RestHighLevelClient and BulkPro

Question

在我的 Scala 项目中，我正在尝试用新的 RestHighLevelClient 更改旧的 transportClient，以连接到 Elasticsearch (6.1(。

但是我在尝试创建批量处理器时遇到了问题，我不知道如何将此示例从 Java 转换为 Scala

`BulkProcessor.Builder builder = BulkProcessor.builder(client::bulkAsync, listener);`

我的代码是这样的：

val client=new RestHighLevelClient(
  RestClient.builder(
    new HttpHost("localhost", 9200, "http") ) )
val builder: BulkProcessor.Builder =BulkProcessor.builder(client, listener)

但是我得到这个错误：

 overloaded method value builder with alternatives:
[error]   (x$1: java.util.function.BiConsumer[org.elasticsearch.action.bulk.BulkRequest,org.elasticsearch.action.ActionListener[org.elasticsearch.action.bulk.BulkResponse]],x$2: org.elasticsearch.action.bulk.BulkProcessor.Listener)org.elasticsearch.action.bulk.BulkProcessor.Builder <and>
[error]   (x$1: org.elasticsearch.client.Client,x$2: org.elasticsearch.action.bulk.BulkProcessor.Listener)org.elasticsearch.action.bulk.BulkProcessor.Builder
[error]  cannot be applied to (org.elasticsearch.client.RestHighLevelClient, org.elasticsearch.action.bulk.BulkProcessor.Listener)
[error]   val builder: BulkProcessor.Builder =BulkProcessor.builder(client, listener)

如果我尝试将" client::bulkAsync"与" client.bulkAsync _"或" client::bulkAsync(_._._)" 进行转换，则会出现其他错误：

type mismatch;
[error] found   : org.apache.http.Header*
[error] required: org.apache.http.Header
[error] var bulkProcessor =BulkProcessor.builder(client.bulkAsync(_,_,_),listener)

这是Java中的bulkAsync方法：

 public final void bulkAsync(BulkRequest bulkRequest, ActionListener<BulkResponse> listener, Header... headers) {

有没有人在Scala中使用过RestHighLevelClient和BulkProcessor？ 或者知道如何解决此错误？

Answer 1

Scala 2.11 与 Java 8 函数接口没有很好的互操作性。创建BiConsumerId(接受两个输入参数并且不返回任何结果(应该解决问题：

val bulkAsyncAsJava :  BiConsumer[BulkRequest, ActionListener[BulkResponse]] = new BiConsumer[BulkRequest, ActionListener[BulkResponse]] {
  override def accept(bulkRequest : BulkRequest , actionListener:ActionListener[BulkResponse]): Unit = {
    restHighLevelClient.bulkAsync(bulkRequest, actionListener)
  }
}
val bulkProcessor = BulkProcessor.builder(bulkAsyncAsJava, new BulkProcessor.Listener() {
override def beforeBulk(executionId: Long, request: BulkRequest): Unit = {
}
override
def afterBulk(executionId: Long, request: BulkRequest, response: BulkResponse): Unit = {
}
override
def afterBulk(executionId: Long, request: BulkRequest, failure: Throwable): Unit = {
}}).build