Elasticsearch RestHighLevelClient批量操作运行一段时间后失效求助

1. 客户端与服务端版本兼容性不匹配 #

你的Elasticsearch服务端版本为7.10.2，但使用的elasticsearch-rest-high-level-client版本是7.5.1，两者minor版本差距较大。ES官方明确建议客户端与服务端版本尽量保持一致，版本差异过大可能引发：

• 通信协议不兼容，批量请求在传输过程中被静默丢弃或无法正常解析
• API行为差异，高版本服务端的响应逻辑变化后，低版本客户端无法正确处理，导致请求挂起无响应

2. 无限超时配置导致请求阻塞 #

从客户端初始化代码的注释可以看到，三个超时参数支持设置为-1（无限超时）：

//Sets a specified timeout value, in milliseconds, to be used when opening a communications link to the resource referenced by this URLConnection.
// A timeout of -1 is interpreted as an infinite timeout.
requestConfigBuilder.setConnectTimeout(configuration.getConnectionTimeout())
        //Sets the socket timeout value. A timeout of -1 is interpreted as an infinite timeout.
        .setSocketTimeout(configuration.getSocketTimeout())
        //Set the time to wait for a connection from the connection manager/pool. A timeout of -1 is interpreted as an infinite timeout.
        .setConnectionRequestTimeout(configuration.getConnectionRequestTimeout());

如果你的配置中这三个超时为-1，当ES集群出现压力突增、网络波动、节点临时不可用等情况时，批量请求会一直阻塞等待响应，不会抛出异常。随着阻塞请求增多，处理线程池会被耗尽，后续批量操作无法获得执行线程，最终表现为"停止工作"但无错误日志。

3. 未处理批量响应中的部分失败 #

当前批量操作代码仅捕获了请求抛出的异常，但BulkResponse中可能存在部分请求失败的情况（比如文档字段类型错误、版本冲突、磁盘空间不足等）：

public BulkResponse bulkOperation(BulkRequest bulkRequest) {
        BulkResponse bulkResponse;
        try {
            bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
        } catch (Exception e) {
            LOGGER.error("Exception occurred while indexing bulk request data to elastic search index : " + bulkRequest.requests().get(0).index(), e);
            throw new RuntimeException("Error while bulk request index : " + bulkRequest.requests().get(0).index(),e);
        }
        return bulkResponse;
    }

代码中没有对bulkResponse.hasFailures()进行判断和处理，如果持续出现部分失败且Kafka偏移量被自动提交，应用会一直重复处理有问题的数据，甚至导致客户端进入异常静默状态。

4. 连接池资源耗尽风险 #

RestHighLevelClient基于Apache HttpClient实现，默认连接池大小有限。如果批量请求因无限超时长期占用连接不释放，会导致连接池耗尽，新的批量请求无法获取连接，只能阻塞等待，没有异常抛出，最终表现为批量操作停滞。

内容的提问来源于stack exchange，提问作者Hari Prasad N