背景
给普惠顺风车和打车使用的匹配引擎，他数据同步使用的是搜索dump引擎，dump引擎用flink程序消费往es引擎写数据。临近930大促却频繁的发生数据丢失的情况，经过分析发现flink里的部分worker发生了错误，而且错误之后就一直报错，再也无法恢复。这个问题之前就小概率发生，但是一般发生之后也没有什么问题，自己就恢复了，所以一直没有在意。但是最近发生的越来越频繁，又是临近930，又涉及到了普惠的底层最核心的数据服务层，一旦在930当天发生，就会导致顺风车和打车交易跌0的故障，直接影响目标达成，因此不得不重点关注起来。
我们首先做的一件事，是把这个情况监控起来，由于日志是在flink的TaskManager的日志里，只能通过手动查看是非常被动的，在询问了数据平台的同事之后，了解到这些日志是已经收集了(不过被某些规则过滤了)。然后针对这个错误做了一个监控和告警，一旦有发生，能第一时间知道，然后人肉的介入处理。
分析
光靠告警+人肉的处理，终归不是办法，人总不能24小时在电脑旁边，还是得分析问题的原因，然后解决他。
先看看报错如下，I/O reactor status: STOPPED，有了报错关键字，就比较容易寻找答案了。
2022-09-13 07:39:58,919 ERROR COMMON_ERROR [] - action [update {[hitch_economy_hitch_economy_car_driver_2022_09_10_11_18_24][_doc][1459871599], doc_as_upsert[true], doc[index {[null][_doc][null], source[{"engine_type":1,"driver_status":"121 131 141 152 161 112 221 232 211","user_new_id":1459871599,"open_status":20}]}], scripted_upsert[false], detect_noop[true]}] sink to es failed [Request cannot be executed; I/O reactor status: STOPPED]
java.lang.IllegalStateException: Request cannot be executed; I/O reactor status: STOPPED
at org.apache.http.util.Asserts.check(Asserts.java:46) ~[16590.jar:?]
at org.apache.http.impl.nio.client.CloseableHttpAsyncClientBase.ensureRunning(CloseableHttpAsyncClientBase.java:90) ~[16590.jar:?]
at org.apache.http.impl.nio.client.InternalHttpAsyncClient.execute(InternalHttpAsyncClient.java:123) ~[16590.jar:?]
at org.elasticsearch.client.RestClient.lambda$performRequestAsync$0(RestClient.java:327) ~[16590.jar:?]
at org.elasticsearch.client.Cancellable.runIfNotCancelled(Cancellable.java:81) ~[16590.jar:?]
at org.elasticsearch.client.RestClient.performRequestAsync(RestClient.java:325) ~[16590.jar:?]
at org.elasticsearch.client.RestClient.performRequestAsync(RestClient.java:314) ~[16590.jar:?]
at org.elasticsearch.client.RestHighLevelClient.internalPerformRequestAsync(RestHighLevelClient.java:1653) ~[16590.jar:?]
at org.elasticsearch.client.RestHighLevelClient.performRequestAsync(RestHighLevelClient.java:1614) ~[16590.jar:?]
at org.elasticsearch.client.RestHighLevelClient.performRequestAsyncAndParseEntity(RestHighLevelClient.java:1580) ~[16590.jar:?]
at org.elasticsearch.client.RestHighLevelClient.bulkAsync(RestHighLevelClient.java:509) ~[16590.jar:?]
at org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchUniversalApiCallBridge.lambda$createBulkProcessorBuilder$0(ElasticsearchUniversalApiCallBridge.java:86) ~[16590.jar:?]
at org.elasticsearch.action.bulk.Retry$RetryHandler.execute(Retry.java:205) [16590.jar:?]
at org.elasticsearch.action.bulk.Retry.withBackoff(Retry.java:59) [16590.jar:?]
at org.elasticsearch.action.bulk.BulkRequestHandler.execute(BulkRequestHandler.java:62) [16590.jar:?]
at org.elasticsearch.action.bulk.BulkProcessor.execute(BulkProcessor.java:455) [16590.jar:?]
at org.elasticsearch.action.bulk.BulkProcessor.execute(BulkProcessor.java:464) [16590.jar:?]
at org.elasticsearch.action.bulk.BulkProcessor.access$400(BulkProcessor.java:54) [16590.jar:?]
at org.elasticsearch.action.bulk.BulkProcessor$Flush.run(BulkProcessor.java:504) [16590.jar:?]
at org.elasticsearch.threadpool.Scheduler$ReschedulingRunnable.doRun(Scheduler.java:223) [16590.jar:?]
at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:37) [16590.jar:?]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_201]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_201]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) [?:1.8.0_201]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) [?:1.8.0_201]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_201]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_201]
at java.lang.Thread.run(Thread.java:748) [?:1.8.0_201]
初步方案
经过一通搜索，发现这个是ES Client的官方bug，并且有很多人遇到，而到目前为止，es官方并未能很好的解决。https://github.com/elastic/elasticsearch/issues/49124 这里有非常热烈的讨论，感兴趣的可以仔细爬下E文，有人说是httpclient的bug，要升级httpclient到5.0，也有人说是因为handler没处理好，并给出了示例代码，于是我们直接拷贝和使用了示例代码
RestClientBuilder builder = RestClient.builder(...);
CustomHttpClientConfigCallback configurationCallback = new CustomHttpClientConfigCallback();
builder.setHttpClientConfigCallback(configurationCallback);
public static class CustomHttpClientConfigCallback implements RestClientBuilder.HttpClientConfigCallback {
@Override
public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
// Add custom exception handler.
// - https://hc.apache.org/httpcomponents-core-ga/tutorial/html/nio.html#d5e601
// - This always handles the exception and just logs a warning.
try {
DefaultConnectingIOReactor ioReactor = new DefaultConnectingIOReactor();
ioReactor.setExceptionHandler(new IOReactorExceptionHandler() {
@Override
public boolean handle(IOException e) {
logger.warn("System may be unstable: IOReactor encountered a checked exception : " + e.getMessage(), e);
return true; // Return true to note this exception as handled, it will not be re-thrown
}
@Override
public boolean handle(RuntimeException e) {
logger.warn("System may be unstable: IOReactor encountered a runtime exception : " + e.getMessage(), e);
return true; // Return true to note this exception as handled, it will not be re-thrown
}
});
httpClientBuilder.setConnectionManager(new PoolingNHttpClientConnectionManager(ioReactor));
} catch (IOReactorException e) {
throw new RuntimeException(e);
}
return httpClientBuilder;
}
}本来以为解决了，很安心的睡觉去了，但是在0909中秋节大促的当天，又发生了同样问题，所幸0909并不是打车的高峰(0910为打车高峰)，并且我们有备用索引，通过切换索引，避免了故障的发生。
进一步的分析
看起来问题没有这么简单，于是再次仔细阅读官方的讨论，也去爬了一下httpclient的讨论https://lists.apache.org/thread/msdv250tl1kg6snxs89mjdng7qzj3r48 ，大概理解了来龙去脉。
问题原因是es client实际上使用的是httpclient发送http请求到es集群进行通信，因为各种原因(可能是网络抖动，可能是集群抖动)发生了IO异常，这个时候就会抛出异常，如果这个异常不被处理，就会抛给上层，造成上述不可恢复的问题。
那官方为什么不去处理这个异常呢？es官方也说了，他面对的情况太多了，无法顺利的拿到The IOReactor object，这个对象是由 connection manager 来管理的，用户构造他们的方式多种多样，是他们无法控制的，因此他也没办法修复这个，这就有了我们的第一个方案，就是我们自己设置这个异常的处理器。
为什么升级httpclient可以部分解决？
那么又关乎Apache Httpclient有什么关系？为什么有人说升级了httpclient的版本就解决问题了？ 那是因为httpcilent的5.0之前是需要用户自己来处理异常，5.0之后他自己设置了一个默认的异常处理器，所以即使你不用设置，也会有个默认的来兜底。当然，老外也吐槽了，Apache httpclient的文档混乱，连一个例子也没有
老外也很暴躁，也有键盘侠？ 上来就吐槽es的开发没有仔细阅读文档，导致产生了bug的
)
为什么我们的代码不生效?
经过我们寻找，发现了这段报错
回过头来，我们再看看我们的代码，可以发现，他处理了两种异常IOException, RunTimeException，而我们这个是由于类冲突，报出来的是NoSuchMethodError，自然也就没办法交给handler来处理了，他不是我们代码里所捕获的异常，因为也解决不了我们的问题了
ioReactor.setExceptionHandler(new IOReactorExceptionHandler() {
@Override
public boolean handle(IOException e) {
log.warn("System may be unstable: IOReactor encountered a checked exception : "
+ e.getMessage(), e);
return true; // Return true to note this exception as handled, it will not be re-thrown
}
@Override
public boolean handle(RuntimeException e) {
log.warn("System may be unstable: IOReactor encountered a runtime exception : "
+ e.getMessage(), e);
return true; // Return true to note this exception as handled, it will not be re-thrown
}
});为什么最近发生的特别频繁
在搜索之后，发现有人遇到了同样的问题，并且还稳定的复现了https://www.cnblogs.com/quwenli/p/15050215.html，顺着这个思路，我们再看看我们代码，发现在捕获异常的同时，还夹带了一些"私货"，不知道从哪拷来的，不过看看也还挺合理，连接数的设置比默认值要好。但是这里面的关键点在于keep alive的时间，720s是远远大于我们服务端(slb的nginx层配置)60s，这就导致了频繁的发生服务端主动断开了keep alive，而客户端还在傻傻的等待，以为自己还是好的这种情况。
httpClientBuilder.setMaxConnTotal(100);
httpClientBuilder.setMaxConnPerRoute(50);
List<Header> headers = new ArrayList<>(2);
headers.add(new BasicHeader("Connection", "keep-alive"));
headers.add(new BasicHeader("Keep-Alive", "720"));解决NoSuchMethodError
继续翻看es client部分的代码，在org.apache.http.nio.protocol.HttpAsyncRequestExecutor.endOfInput这个方法里，按照代码所写他会判断连接的状态，无论什么状态，他都能正确的处理，问题在那个无参的ConnectionClosedException()，IDEA里本地看到的没什么问题，但是把打出来的fat jar包解压之后发现，打在jar包里的Class是一个旧版本的只有一个有参构造函数的ConnectionClosedException，这个肯定就是类冲突了。
if (handler != null) {
if (state.isValid()) {
handler.inputTerminated();
} else {
handler.failed(new ConnectionClosedException());
}
}解决冲突的过程就不细说了，总之是非常奇怪，通过maven dependency:tree以及IDEA自带的Dependency Analyzer都没看到冲突，而且看到的版本都是正确的高版本ConnectionClosedException,可以从tree里看到依赖的版本明明是4.4.13，在也没有其他的地方有引入这个依赖，但是打到fat jar里却是4.4.6，都不知道这个4.4.6的版本哪来的。
由于这个类在httpcore包里，因此我把httpcore单独的申明出来，再到fat jar里解压出来确认打进去的是正确的版本之后，再继续测试
总结
到现在我们知道了问题所在了，由于keep alive的时间设置不合理，导致es的服务端断开了连接，而client由于类冲突，在解决Closed问题的途中，抛出了NoSuchMethodError，而这个错误又不是handler所能处理的，因此这个 I/O reactor的错误就一直报下去。
因为这个连接已经坏了，却不能正常的关闭，每次拿到这个连接去发送请求的时候都会发生错误。
我们遇到的情况，不是官方讨论里的IOException的情况，而是自己的类冲突带来的非正常关闭。正因为每个人遇到的情况都不同，所以官方讨论的代码有人能解决，有人却无法解决。
遇到问题，我们还是要深入去理解和分析，理解里面的原因，才能找到问题，别人的答案可能不一定适合你
解决和验证
我们尝试了两个方案，既然是keep alive设置不合理导致的连接断开，那么我们把keep alive设置的少于服务端的时长避免断开，不就可以解决问题了。
另外一个方面，由于类冲突导致的close情况无法正常处理，这个是本质，把类冲突解决之后，close就能正常处理，即使发生了各种原因的关闭(可能不止 keep alive这种情况)，程序也能正常处理
同时，我们的handler也需要添加，防止网络抖动抛出的IOException带来的问题。
这几天一直关注修改后的代码运行情况，直到看到这条日志，能证明我们代码上去之后，又发生了连接关闭事件，不同的是，这次程序能够正确的处理连接关闭的问题，并且继续运行下去，不会出错。
建议
由于我们用的是相同的依赖，都是用的数据平台释放给我们的父依赖关系，所以看到生产还有大量的报错，并且一定也发生了数据丢失，至于数据丢失是否重要，对业务有没有影响，这个不知道是谁的业务，快去检查下数据丢失吧
只要是flink往es写数据的，用的是这个依赖包的依赖关系的，应该都有同样的问题，去看看taskmanager.err日志里看看有没有NoSuchMethoError的报错，应该都能参照我们的方案解决
参考文档
https://github.com/elastic/elasticsearch/issues/49124
https://lists.apache.org/thread/msdv250tl1kg6snxs89mjdng7qzj3r48
https://www.cnblogs.com/quwenli/p/15050215.html