filebeat无法把消息输入到kafka中？

环境：
centos7.4
kafka2.11
filebeat5.6.8
-----------------------
filbeat配置：
- input_type: log
  paths:
    - /var/log/nginx/*.log
  encoding: utf-8
  tags: ["nginx-accesslog"]
output.kafka:
  enabled: true
  hosts: ["127.0.0.1:9092"]
  topic: logstash-nginx

创建主题： logstash-nginx
启动filebeat，zookeeper及kafka（单节点）

执行消费消息命令：bin/kafka-console-consumer.sh --zookeeper 192.168.31.101:2181 --topic logstash-nginx --from-beginning

访问nginx页面，生成新消息到nginx的log文件中。
此时，消费消息终端，显示错误信息：
WARN Fetching topic metadata with correlation id 0 for topics [Set(logstash-nginx)] from broker [BrokerEndPoint(0,ec2-52-60-148-196.ca-central-1
.compute.amazonaws.com,9092)] failed (kafka.client.ClientUtils$)java.nio.channels.ClosedChannelException
    at kafka.network.BlockingChannel.send(BlockingChannel.scala:110)
    at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:80)
    at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:79)
    at kafka.producer.SyncProducer.send(SyncProducer.scala:124)
    at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:60)
    at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:95)
    at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:68)
    at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)

请问是什么原因导致？如何解决？谢谢。