ELK日志系统，logstash与kafka交互出错，望大大们解答

在部署日志系统时，使用logstash进入kafka时报错，logstash配置文件如下：
报错是第一个图，大概流程是：(nginx主机的accesslog)-->fielbeat-->kafka<--logstash（消费kafka数据）-output->es-->kibana。
希望大佬解答下，万分感谢，如果那个地方没说清楚，可以问我。
如果有好点的可用的elk整套的相关文档，可以分享的话也万分感谢。