ES High Level Rest Client 超时问题排查及解决

(1048条消息) ES异常：Connection reset by peer_浊酒入清梦的博客-CSDN博客https://blog.csdn.net/m0_37862405/article/details/108324096

ES High Level Rest Client 超时问题排查及解决 - 墨天轮 (modb.pro)https://www.modb.pro/db/388569记一次elasticsearch client的conncection reset 异常－小专栏 (xiaozhuanlan.com)https://xiaozhuanlan.com/topic/7350912846(1049条消息) Es 超时设置 high-level-client_小檗的博客-CSDN博客_es 设置超时时间https://blog.csdn.net/xiaobozhi1993/article/details/114260115

ES的high level查询超时设置失效问题 - 简书 (jianshu.com)https://www.jianshu.com/p/f781e38b3bb8

问题描述
es7.4.1客户端查询时报异常：

Caused by: java.io.IOException: Connection reset by peer
at org.elasticsearch.client.RestClient.extractAndWrapCause(RestClient.java:793)
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.search(RestHighLevelClient.java:930)
...
Caused by: java.io.IOException: Connection reset by peer
at sun.nio.ch.FileDispatcherImpl.read0(Native Method)
at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:39)
at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223)
at sun.nio.ch.IOUtil.read(IOUtil.java:197)
at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:380)
at org.apache.http.impl.nio.reactor.SessionInputBufferImpl.fill(SessionInputBufferImpl.java:231)
...

原因排查

es7.4.1客户端（RestHighLevelClient）使用的apache httpclient 版本为4.1.3，keepAlive默认为-1（即无限）。在某些特殊情况下，ES服务端的keepAlive短于ES客户端的keepAlive，进而导致：ES服务端已经关闭了连接，但是客户端还继续复用该连接，从而抛出上述异常。

解决方法

手动设置RestHighLevelClient的keepAlive，通过KeepAliveStrategy手动配置keepAlive代码如下：

public static RestHighLevelClient createRestHighLevelClient(String esUrl, Long keepAlive) {RestClientBuilder clientBuilder = RestClient.builder(createHttpHost(URI.create(esUrl))).setHttpClientConfigCallback(requestConfig -> requestConfig.setKeepAliveStrategy((response, context) -> keepAlive));return new RestHighLevelClient(clientBuilder);
}private static HttpHost createHttpHost(URI uri) {if (StringUtils.isEmpty(uri.getUserInfo())) {return HttpHost.create(uri.toString());}try {return HttpHost.create(new URI(uri.getScheme(), null, uri.getHost(), uri.getPort(), uri.getPath(),uri.getQuery(), uri.getFragment()).toString());} catch (URISyntaxException ex) {throw new IllegalStateException(ex);}}

源码可参考

原理

RestHighLevelClient设置http配置的源码：

public RestClientBuilder setHttpClientConfigCallback(HttpClientConfigCallback httpClientConfigCallback) {Objects.requireNonNull(httpClientConfigCallback, "httpClientConfigCallback must not be null");this.httpClientConfigCallback = httpClientConfigCallback;return this;
}public interface HttpClientConfigCallback {HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder);}

根据HttpAsyncClientBuilder中的部分源码可知，RestHighLevelClient默认keepAliveStrategy为DefaultConnectionKeepAliveStrategy：

ConnectionKeepAliveStrategy keepAliveStrategy = this.keepAliveStrategy;
if (keepAliveStrategy == null) {keepAliveStrategy = DefaultConnectionKeepAliveStrategy.INSTANCE;
}

DefaultConnectionKeepAliveStrategy源码，如果没有在http表头中设置，默认返回-1：

@Contract(threading = ThreadingBehavior.IMMUTABLE)
public class DefaultConnectionKeepAliveStrategy implements ConnectionKeepAliveStrategy {public static final DefaultConnectionKeepAliveStrategy INSTANCE = new DefaultConnectionKeepAliveStrategy();@Overridepublic long getKeepAliveDuration(final HttpResponse response, final HttpContext context) {Args.notNull(response, "HTTP response");final HeaderElementIterator it = new BasicHeaderElementIterator(response.headerIterator(HTTP.CONN_KEEP_ALIVE));while (it.hasNext()) {final HeaderElement he = it.nextElement();final String param = he.getName();final String value = he.getValue();if (value != null && param.equalsIgnoreCase("timeout")) {try {return Long.parseLong(value) * 1000;} catch(final NumberFormatException ignore) {}}}return -1;}}

参考文档：

https://juejin.im/post/6844904069442568205

https://stackoverflow.com/questions/52997697/how-to-get-around-connection-reset-by-peer-when-using-elasticsearchs-restclie