背景：使用了自定义插件之后，出现了数据无法写入es的问题。

查看日志后发现如下信息：

[2019-xxx][INFO ][logstash.outputs.elasticsearch] retrying failed action with response code: 429 ({"type"=>"es_rejected_execution_exception", "reason"=>"rejected execution of processing of [1406096379][indices:data/write/bulk[s][p]]: request: BulkShardRequest [[logstash-xxx][0]] containing [3000] requests, target allocation id: 5nHO5QaBQ36pxxxx, primary term: 1 on EsThreadPoolExecutor[name = data-xxx/write, queue capacity = 200, org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor@25eb117b[Running, pool size = 24, active threads = 24, queued tasks = 200, completed tasks = 621218675]]"})

查阅相关资料，了解到出现"es_rejected_execution_exception"是因为logstash写入es速度小于logstash读取数据速度，导致logstash频繁给es发送bulk reuqest，从而导致es集群的网络io过载，进而产生以上的问题提示，表明elasticsearch无法继续接收数据。
可以通过调整logstash.yml的pipeline.workers,pipeline.output.workers ,pipeline.batch.size,pipeline.batch.delay，或者output plugin的flush_size, idle_flush_time参数来进行优化，经过试验后发现7.3.1版本的output plugin中flush_size,idle_flush_time已经不可用。
调整pipeline.batch.size=3000,pipeline.batch.delay=100后，正常运行了几小时后又出现了上述问题。
查阅官方文档后，找到一个output plugin配置项sniffing_delay与上述idle_flush_time类似。
怀疑是用了自定义插件导致处理速度过慢，但是经过验证发现处理速度与使用自定义插件之前并没有差距。
问题解决：将pipeline.batch.size调整为2000后，logstash能流畅处理，并未产生拥堵。所以，产生这个问题的原因还是进入logstash的速度大于logstash输出的速度导致堵塞。在进行logstash性能优化时，不同的处理流程可能需要设置不同的控制参数才能达到较优的性能。
说明：这个错误的问题的产生与logstash的bulk request息息相关，es_rejected_execution_exception正是logstash像es发出请求后，若bulk queue已满，es就会向logstash返回该应答。

bulk request详解

Why am I seeing bulk rejections in my Elasticsearch cluster?

当我们发送一个bulk indexing request给ES集群时，ES集群做了什么？
当一个bulk request（该请求中包含待处理的文档）到达es集群时，该请求会被放入bulk queue中，并由bulk thread pool中的bulk thread处理，收到请求的节点为协调节点，因为该节点将会管理该请求的生命周期，并且负责响应请求。
请求处理的第一步是将bulk request按照文档将要被分配到的分片进行分割。每一个sub-request就会被发送到分片所在的节点，并加入该节点的bulk queue中。如果该节点的bulk queue已满，那么协调节点将会收到bulk sub-request被拒绝的消息。bulk queue中的请求将被bulk thread pool处理，并将文档当做处理的一部分发送到副本分片。当bulk sub-request处理结束后，该节点将向协调节点返回响应。
当所有bulk sub-request被处理或被拒绝，协调节点将向客户端返回响应，这其中有可能有些文档是被拒绝了的。
bulk queue被设置为有最大限制是因为，如果没有限制的话，将会很容易通过恶意的行为或者仅仅是无心之失从而让es集群变得不稳定或不可靠。这个限制通过大量应用实例的经验总结而来，所以通常来说是不需要修改的。
使用HTTP或其他协议的接口时，当收到429响应，就表示bulk request被拒绝了。这时一般的做法是重试这个请求，logstash和filebeat的做法就是这样。

那么请求被拒绝的频率与什么有关呢？
该文通过试验得出的结论是：

• 当分片数与索引数越多，请求被拒绝的频率就越大
• 当集群节点越多，请求被拒绝的频率就越小

能通过提高bulk queue size来解决这个问题吗？

• 提高bulk queue size能做到的只是延迟问题到来的时间，并不能解决问题
• 有可能增大jvm heap占用，从而导致集群不稳定

能通过设置专用的协调节点解决问题吗？

• 增加专用协调节点只是让数据节点能专心处理被分配到的子请求，对于有些用例可能有效，但是大多数时候是没有什么大的效果的。

我们能做什么？
并没有一个一劳永逸的方法，我们能做的是在问题发生时，尝试了解原因，看看是单节点还是整个集群出现拒绝的情况。如果集群无法处理负载，请确认集群负载是否均匀。如果负载均衡不能解决问题，能做的可能就是增加集群节点数量了，这将会增加队列的容量，减少队列被填满的可能性。

关于分片

How many shards should I have in my Elasticsearch cluster？

Elasticsearch 中的数据会整理为索引。每个索引又由一个或多个分片组成。每个分片都是一个 Lucene 索引实例，您可以将其视作一个独立的搜索引擎，它能够对 Elasticsearch 集群中的数据子集进行索引并处理相关查询。
每个分片都有一部分数据需要保存在内存中，这部分数据也会占用堆内存空间。这包括存储分片级别以及段级别信息的数据结构，因为只有这样才能确定数据在磁盘上的存储位置。这些数据结构的大小并不固定，不同用例之间会有很大的差别。段相关开销有一个重要特征，那就是其并不与段的大小呈严格正比关系。这意味着，与较小的段相比，对于较大的段而言，其单位数据量所需的开销要小一些。二者之间的差异可能会十分巨大。
在 Elasticsearch 中，每个查询都是在单个分片上以单线程方式执行的。然而，可以同时对多个分片进行处理，正如可以针对同一分片进行多次查询和聚合一样。这意味着，最低查询延时（假设没有缓存）将取决于数据、查询类型，以及分片大小。尽管查询很多个小分片会加快单个分片的处理速度，但是由于有很多任务需要进入队列并按顺序加以处理，所以与查询较少的大分片相比，这种方法并不一定会加快查询速度。如果有多个并发查询，拥有很多小分片还会降低查询吞吐量。

ELK系统之logstash问题：retrying failed action with response code: 429
logstash output plugin 官方文档
Why am I seeing bulk rejections in my Elasticsearch cluster?
How many shards should I have in my Elasticsearch cluster？