elasticsearch - 使用 Elastic RestHighLevelClient 批量请求时的 ConnectionClosedException

Question

我在 Elasticsearch 中为 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 一起使用时。