使用弹性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。

在我的案例中，哈利特的建议是导致这个问题的部分原因。仔细检查日志，可以明显看出HttpAsyncRequestExecutor正在调用一个名为ConnectionClosedException的默认构造函数，但是，这个默认构造函数在目标中并不存在

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

Answer 2

我们遇到了类似的问题，并修复了我们认为服务具有httpcore-4.4.3的问题，因为弹性客户端需要httpcore-4.4.12。因此，我们更新了所有的HTTP依赖项，以映射elastic-rest-high-level-client所需的版本

https://mvnrepository.com/artifact/org.elasticsearch.client/elasticsearch-rest-client/7.5.2