elasticsearch - 带有 elasticsearch-rest-client-7.2.0 的间歇性 SocketTimeoutException

Question

我正在使用 RestHighLevelClient 版本 7.2 连接到 ElasticSearch 集群版本 7.2。我的集群有 3 个主节点和 2 个数据节点。数据节点内存配置:2 核和 8 GB。我习惯于在我的 spring boot 项目中使用下面的代码来创建 RestHighLevelClient 实例。

 @Bean(destroyMethod = "close")
    @Qualifier("readClient")
    public RestHighLevelClient readClient(){

        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY,
                new UsernamePasswordCredentials(elasticUser, elasticPass));

        RestClientBuilder builder = RestClient.builder(new HttpHost(elasticHost, elasticPort))
                .setHttpClientConfigCallback(httpClientBuilder ->httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider).setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(5).build()));

        builder.setRequestConfigCallback(requestConfigBuilder -> requestConfigBuilder.setConnectTimeout(30000).setSocketTimeout(60000)
                );

        RestHighLevelClient restClient = new RestHighLevelClient(builder);
        return restClient;
    }

RestHighLevelClient 是一个单例 bean。我间歇性地收到带有 GET 和 PUT 请求的 SocketTimeoutException。索引大小约为 50 MB。我尝试增加套接字超时值，但仍然收到相同的错误。我缺少一些配置吗？任何帮助将不胜感激。

Answer 1

我得到这个问题只是想分享一下，以便它可以帮助其他人。我正在使用负载均衡器连接到 ElasticSerach 集群。从我的 RestClientBuilder 代码中可以看出，我只使用了负载均衡器主机和端口。尽管我有多个主节点，但在连接超时的情况下，RestClient 仍然没有重试我的请求。

RestClientBuilder builder = RestClient.builder(new HttpHost(elasticHost, elasticPort))
                .setHttpClientConfigCallback(httpClientBuilder ->httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider).setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(5).build()));

根据 RestClient 代码，如果我们使用单个主机，那么在出现任何连接问题时它不会重试。所以我改变了我的代码如下，它开始工作了。

RestClientBuilder builder = RestClient.builder(new HttpHost(elasticHost, 9200),new HttpHost(elasticHost, 9201))).setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider));

完整的 RestClient 代码请引用 https://github.com/elastic/elasticsearch/blob/master/client/rest/src/main/java/org/elasticsearch/client/RestClient.java

RestClient 中的重试代码块

private Response performRequest(final NodeTuple<Iterator<Node>> nodeTuple,
                                    final InternalRequest request,
                                    Exception previousException) throws IOException {
        RequestContext context = request.createContextForNextAttempt(nodeTuple.nodes.next(), nodeTuple.authCache);
        HttpResponse httpResponse;
        try {
            httpResponse = client.execute(context.requestProducer, context.asyncResponseConsumer, context.context, null).get();
        } catch(Exception e) {
            RequestLogger.logFailedRequest(logger, request.httpRequest, context.node, e);
            onFailure(context.node);
            Exception cause = extractAndWrapCause(e);
            addSuppressedException(previousException, cause);
            if (nodeTuple.nodes.hasNext()) {
                return performRequest(nodeTuple, request, cause);
            }
            if (cause instanceof IOException) {
                throw (IOException) cause;
            }
            if (cause instanceof RuntimeException) {
                throw (RuntimeException) cause;
            }
            throw new IllegalStateException("unexpected exception type: must be either RuntimeException or IOException", cause);
        }
        ResponseOrResponseException responseOrResponseException = convertResponse(request, context.node, httpResponse);
        if (responseOrResponseException.responseException == null) {
            return responseOrResponseException.response;
        }
        addSuppressedException(previousException, responseOrResponseException.responseException);
        if (nodeTuple.nodes.hasNext()) {
            return performRequest(nodeTuple, request, responseOrResponseException.responseException);
        }
        throw responseOrResponseException.responseException;
    }