logstash 读取不了kafka 报错。集群使用ambari搭建 报错如下，配置如下

[2018-08-21T20:14:04,331][WARN ][org.apache.kafka.clients.NetworkClient] [Consumer clientId=test-1, groupId=test] Connection to node -2 could not be established. Broker may not be available.
[2018-08-21T20:14:04,381][WARN ][org.apache.kafka.clients.NetworkClient] [Consumer clientId=test-0, groupId=test] Connection to node -3 could not be established. Broker may not be available.
[2018-08-21T20:14:04,433][WARN ][org.apache.kafka.clients.NetworkClient] [Consumer clientId=test-4, groupId=test] Connection to node -1 could not be established. Broker may not be available.
[2018-08-21T20:14:04,582][WARN ][org.apache.kafka.clients.NetworkClient] [Consumer clientId=test-3, groupId=test] Connection to node -1 could not be established. Broker may not be available.
[2018-08-21T20:14:04,633][WARN ][org.apache.kafka.clients.NetworkClient] [Consumer clientId=test-0, groupId=test] Connection to node -2 could not be established. Broker may not be available.
[2018-08-21T20:14:04,633][WARN ][org.apache.kafka.clients.NetworkClient] [Consumer clientId=test-2, groupId=test] Connection to node -3 could not be established. Broker may not be available.
[2018-08-21T20:14:04,683][WARN ][org.apache.kafka.clients.NetworkClient] [Consumer clientId=test-3, groupId=test] Connection to node -2 could not be established. Broker may not be available.
[2018-08-21T20:14:04,834][WARN ][org.apache.kafka.clients.NetworkClient] [Consumer clientId=test-2, groupId=test] Connection to node -1 could not be established. Broker may not be available.
[2018-08-21T20:14:04,885][WARN ][org.apache.kafka.clients.NetworkClient] [Consumer clientId=test-2, groupId=test] Connection to node -2 could not be established. Broker may not be available.
[2018-08-21T20:14:05,036][WARN ][org.apache.kafka.clients.NetworkClient] [Consumer clientId=test-4, groupId=test] Connection to node -3 could not be established. Broker may not be available.
[2018-08-21T20:14:05,084][WARN ][org.apache.kafka.clients.NetworkClient] [Consumer clientId=test-1, groupId=test] Connection to node -1 could not be established. Broker may not be available.
[2018-08-21T20:14:05,085][WARN ][org.apache.kafka.clients.NetworkClient] [Consumer clientId=test-1, groupId=test] Connection to node -3 could not be established. Broker may not be available.
[2018-08-21T20:14:05,285][WARN ][org.apache.kafka.clients.NetworkClient] [Consumer clientId=test-0, groupId=test] Connection to node -1 could not be established. Broker may not be available.
[2018-08-21T20:14:05,287][WARN ][org.apache.kafka.clients.NetworkClient] [Consumer clientId=test-4, groupId=test] Connection to node -2 could not be established. Broker may not be available.
[2018-08-21T20:14:05,288][WARN ][org.apache.kafka.clients.NetworkClient] [Consumer clientId=test-4, groupId=test] Connection to node -1 could not be established. Broker may not be available.
[2018-08-21T20:14:05,336][WARN ][org.apache.kafka.clients.NetworkClient] [Consumer clientId=test-0, groupId=test] Connection to node -3 could not be established. Broker may not be available.
[2018-08-21T20:14:05,386][WARN ][org.apache.kafka.clients.NetworkClient] [Consumer clientId=test-3, groupId=test] Connection to node -3 could not be established. Broker may not be available.
 
 
 
 
===============================================================================

input {
    kafka {
        bootstrap_servers => ["192.168.103.3:9092,192.168.103.4:9092,192.168.103.5:9092"]
client_id => "test"
group_id => "test"
        auto_offset_reset => "latest"
consumer_threads => 5
topics => ["test"]
codec => json {
            charset => "UTF-8"
        }
    }
}
filter {
        json {
            source => "message"
        }
}
output {
    elasticsearch {
        hosts => [ "dev-bg-m01:9200","dev-bg-w01:9200","dev-bg-w02:9200" ]
        index => "logstash-%{+YYYY.MM.dd}"
  }
}