使用弹性RestHighLevelClient批量请求时的ConnectionClosedException

Question

我正在为弹性搜索中的RestHighLevelClient使用以下代码。

val credentialsProvider = new BasicCredentialsProvider
credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(conf.value.getString("elkUserName"), conf.value.getString("elkPassword")))
val builder = RestClient.builder(new HttpHost(conf.value.getString("elkIp"), Integer.valueOf(conf.value.getString("elkPort"))))
    .setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
        //set timeout
        override def customizeRequestConfig(requestConfigBuilder: RequestConfig.Builder): RequestConfig.Builder = requestConfigBuilder.setConnectTimeout(Integer.valueOf(conf.value.getString("elkWriteTimeOut"))).setSocketTimeout(Integer.valueOf(conf.value.getString("elkWriteTimeOut")))
    }).setHttpClientConfigCallback(new HttpClientConfigCallback() {
        override def customizeHttpClient(httpClientBuilder: HttpAsyncClientBuilder): HttpAsyncClientBuilder = {
            httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider)
        }
    })
client = new RestHighLevelClient(builder)
val requestBuilder = RequestOptions.DEFAULT.toBuilder
requestBuilder.setHttpAsyncResponseConsumerFactory(
    new HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory(1024 * 1024 * 1024))


var request = new BulkRequest()
request.setRefreshPolicy("wait_for")
var sizeOfRequest = 1 L
newListOfMap.foreach {
    vals =>
        val newMap = vals.asJava
    request.add(new IndexRequest(indexName).source(newMap))

}
client.bulk(request, requestBuilder.build)

但我遇到以下异常

java.lang.NoSuchMethodError:
org.apache.http.ConnectionClosedException: method <init>()V not found
  at org.apache.http.nio.protocol.HttpAsyncRequestExecutor.endOfInput(HttpAsyncRequestExecutor.java:356)
  at org.apache.http.impl.nio.DefaultNHttpClientConnection.consumeInput(DefaultNHttpClientConnection.java:261)
  at org.apache.http.impl.nio.client.InternalIODispatch.onInputReady(InternalIODispatch.java:81)
  at org.apache.http.impl.nio.client.InternalIODispatch.onInputReady(InternalIODispatch.java:39)
  at org.apache.http.impl.nio.reactor.AbstractIODispatch.inputReady(AbstractIODispatch.java:114)
  at org.apache.http.impl.nio.reactor.BaseIOReactor.readable(BaseIOReactor.java:162)
  at org.apache.http.impl.nio.reactor.AbstractIOReactor.processEvent(AbstractIOReactor.java:337)
  at org.apache.http.impl.nio.reactor.AbstractIOReactor.processEvents(AbstractIOReactor.java:315)
  at org.apache.http.impl.nio.reactor.AbstractIOReactor.execute(AbstractIOReactor.java:276)
  at org.apache.http.impl.nio.reactor.BaseIOReactor.execute(BaseIOReactor.java:104)
  at org.apache.http.impl.nio.reactor.AbstractMultiworkerIOReactor$Worker.run(AbstractMultiworkerIOReactor.java:591)
  at java.lang.Thread.run(Thread.java:748)
org.apache.http.ConnectionClosedException: Connection closed
unexpectedly at org.elasticsearch.client.RestClient.extractAndWrapCause(RestClient.java:778)
  at org.elasticsearch.client.RestClient.performRequest(RestClient.java:218)
  at org.elasticsearch.client.RestClient.performRequest(RestClient.java:205)
  at org.elasticsearch.client.RestHighLevelClient.internalPerformRequest(RestHighLevelClient.java:1454)
  at org.elasticsearch.client.RestHighLevelClient.performRequest(RestHighLevelClient.java:1424)
  at org.elasticsearch.client.RestHighLevelClient.performRequestAndParseEntity(RestHighLevelClient.java:1394)
  at org.elasticsearch.client.RestHighLevelClient.bulk(RestHighLevelClient.java:492)
  at Utils.ELKUtil$.postDataToELK(ELKUtil.scala:59)

注意：上面的代码适用于较小大小的请求，但在发布较大大小的请求时会出现上述错误。请建议。

Answer 1

如果您的项目同时使用httpcore和httpcore-nio依赖项，请确保它们的两个版本同时为<=4.4.10或>4.4.10。

Harshit的建议是造成我这个问题的部分原因。仔细检查日志，可以明显看出HTTPASyncRequestExecutor正在调用一个名为ConnectionClosedException的默认构造函数，但是，目标中不存在此默认构造函数

HttpAsyncRequestExecutor是HttpCore-NIO包的类。ConnectionClosedException是HttpCore包的类。此问题在V4.4.10之后开始出现，超过这一点，ConnectionClosedException类包含了一个默认构造函数，HttpAsyncRequestExecutor会调用该构造函数。对于版本<=4.4.10，默认构造函数不存在，但是，它是用HTTPASyncRequestExecutor中的一个参数调用的。因此，当两个库一起使用时，它们的版本应该高于或低于V4.4.10。