logstash消费变慢

ELK集群版本都是5.4,已经正常运行一年多,星期一上班发现kafka几十个topic积压了几百上千万的日志.
只有某些数量量很少的topic在正常消费,数据量大的topic都停止消费了,我把logstash全部关闭后,单独指定消费某个积压量大的topic,发现es的数据在增长,但是kafka的消费偏移量没有变化,应该是es写入数据后没有应答给kafka.
 
然后把es集群重启了,重启后初始化分片和副本用了10多个小时,不知道什么原因.
然后用logstash单独消费某个topic是正常的,但是消费速度非常慢,速率大概 1500/s,积压的几百万数据用了几小时才消费完.
 
logstash少量的error日志



 es集群大量的error日志




 
es集群状态. 系统负载在7左右
 
问下社区大佬, 这突然的消费这么慢是怎么造成的, 从什么方面入手解决? 谢谢