Flume连接Kafka的broker出错

Posted WOTGL

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Flume连接Kafka的broker出错相关的知识,希望对你有一定的参考价值。

在启动Flume的时候,出现下面的异常,但是程序照样能运行,Kafka也能够收到数据,只是偶尔会断点。

2016-08-25 15:32:54,561 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - kafka.utils.Logging$class.info(Logging.scala:68)] Fetching metadata from broker id:2,host:10.208.129.5,port:9092 with correlation id 192126244 for 1 topic(s) Set(TRAFFICxxx_LOG)
2016-08-25 15:32:54,562 (SinkRunner-PollingRunner-DefaultSinkProcessor) [ERROR - kafka.utils.Logging$class.error(Logging.scala:103)] Producer connection to 10.208.129.5:9092 unsuccessful
java.net.ConnectException: Connection refused
        at sun.nio.ch.Net.connect0(Native Method)
        at sun.nio.ch.Net.connect(Net.java:454)
        at sun.nio.ch.Net.connect(Net.java:446)
        at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:648)
        at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57)
        at kafka.producer.SyncProducer.connect(SyncProducer.scala:141)
        at kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:156)
        at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:68)
        at kafka.producer.SyncProducer.send(SyncProducer.scala:112)
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53)
        at kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82)
        at kafka.producer.async.DefaultEventHandler$$anonfun$handle$1.apply$mcV$sp(DefaultEventHandler.scala:67)
        at kafka.utils.Utils$.swallow(Utils.scala:167)
        at kafka.utils.Logging$class.swallowError(Logging.scala:106)
        at kafka.utils.Utils$.swallowError(Utils.scala:46)
        at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:67)
        at kafka.producer.Producer.send(Producer.scala:76)
        at kafka.javaapi.producer.Producer.send(Producer.scala:42)
        at org.apache.flume.sink.kafka.KafkaSink.process(KafkaSink.java:129)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:745)
2016-08-25 15:32:54,563 (SinkRunner-PollingRunner-DefaultSinkProcessor) [WARN - kafka.utils.Logging$class.warn(Logging.scala:89)] Fetching topic metadata with correlation id 192126244 for topics [Set(TRAFFICxxx_LOG)] from broker [id:2,host:10.208.129.5,port:9092] failed
java.net.ConnectException: Connection refused
        at sun.nio.ch.Net.connect0(Native Method)
        at sun.nio.ch.Net.connect(Net.java:454)
        at sun.nio.ch.Net.connect(Net.java:446)
        at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:648)
        at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57)
        at kafka.producer.SyncProducer.connect(SyncProducer.scala:141)
        at kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:156)
        at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:68)
        at kafka.producer.SyncProducer.send(SyncProducer.scala:112)
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53)
        at kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82)
        at kafka.producer.async.DefaultEventHandler$$anonfun$handle$1.apply$mcV$sp(DefaultEventHandler.scala:67)
        at kafka.utils.Utils$.swallow(Utils.scala:167)
        at kafka.utils.Logging$class.swallowError(Logging.scala:106)
        at kafka.utils.Utils$.swallowError(Utils.scala:46)
        at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:67)
        at kafka.producer.Producer.send(Producer.scala:76)
        at kafka.javaapi.producer.Producer.send(Producer.scala:42)
        at org.apache.flume.sink.kafka.KafkaSink.process(KafkaSink.java:129)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:745)
2016-08-25 15:32:54,564 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - kafka.utils.Logging$class.info(Logging.scala:68)] Fetching metadata from broker id:1,host:10.208.129.4,port:9092 with correlation id 192126244 for 1 topic(s) Set(TRAFFICxxx_LOG)
2016-08-25 15:32:54,565 (SinkRunner-PollingRunner-DefaultSinkProcessor) [ERROR - kafka.utils.Logging$class.error(Logging.scala:103)] Producer connection to 10.208.129.4:9092 unsuccessful
java.net.ConnectException: Connection refused
        at sun.nio.ch.Net.connect0(Native Method)
        at sun.nio.ch.Net.connect(Net.java:454)
        at sun.nio.ch.Net.connect(Net.java:446)
        at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:648)
        at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57)
        at kafka.producer.SyncProducer.connect(SyncProducer.scala:141)
        at kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:156)
        at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:68)
        at kafka.producer.SyncProducer.send(SyncProducer.scala:112)
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53)
        at kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82)
        at kafka.producer.async.DefaultEventHandler$$anonfun$handle$1.apply$mcV$sp(DefaultEventHandler.scala:67)
        at kafka.utils.Utils$.swallow(Utils.scala:167)
        at kafka.utils.Logging$class.swallowError(Logging.scala:106)
        at kafka.utils.Utils$.swallowError(Utils.scala:46)
        at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:67)
        at kafka.producer.Producer.send(Producer.scala:76)
        at kafka.javaapi.producer.Producer.send(Producer.scala:42)
        at org.apache.flume.sink.kafka.KafkaSink.process(KafkaSink.java:129)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:745)
2016-08-25 15:32:54,566 (SinkRunner-PollingRunner-DefaultSinkProcessor) [WARN - kafka.utils.Logging$class.warn(Logging.scala:89)] Fetching topic metadata with correlation id 192126244 for topics [Set(TRAFFICxxx_LOG)] from broker [id:1,host:10.208.129.4,port:9092] failed
java.net.ConnectException: Connection refused
        at sun.nio.ch.Net.connect0(Native Method)
        at sun.nio.ch.Net.connect(Net.java:454)
        at sun.nio.ch.Net.connect(Net.java:446)
        at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:648)
        at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57)
        at kafka.producer.SyncProducer.connect(SyncProducer.scala:141)
        at kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:156)
        at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:68)
        at kafka.producer.SyncProducer.send(SyncProducer.scala:112)
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53)
        at kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82)
        at kafka.producer.async.DefaultEventHandler$$anonfun$handle$1.apply$mcV$sp(DefaultEventHandler.scala:67)
        at kafka.utils.Utils$.swallow(Utils.scala:167)
        at kafka.utils.Logging$class.swallowError(Logging.scala:106)
        at kafka.utils.Utils$.swallowError(Utils.scala:46)
        at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:67)
        at kafka.producer.Producer.send(Producer.scala:76)
        at kafka.javaapi.producer.Producer.send(Producer.scala:42)
        at org.apache.flume.sink.kafka.KafkaSink.process(KafkaSink.java:129)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:745)
2016-08-25 15:32:54,566 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - kafka.utils.Logging$class.info(Logging.scala:68)] Fetching metadata from broker id:0,host:10.208.129.3,port:9092 with correlation id 192126244 for 1 topic(s) Set(TRAFFICxxx_LOG)
2016-08-25 15:32:54,567 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - kafka.utils.Logging$class.info(Logging.scala:68)] Connected to 10.208.129.3:9092 for producing
2016-08-25 15:32:54,569 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - kafka.utils.Logging$class.info(Logging.scala:68)] Disconnecting from 10.208.129.3:9092
2016-08-25 15:32:54,570 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - kafka.utils.Logging$class.info(Logging.scala:68)] Disconnecting from kafka2:9092
2016-08-25 15:32:54,572 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - kafka.utils.Logging$class.info(Logging.scala:68)] Connected to kafka5:9092 for producing

一开始怀疑.4和.5号机器ping不通或者端口没开,然后查了下端口,发现.4和.5上的Kafka没启动。为了验证这个还专门去Zookeeper上查看了下Kafka的节点信息。

10.208.129.3:2182上有Zookeeper,但上面没有跑Kafka,只跑了4个Storm拓扑
[zk: 10.208.129.3:2182(CONNECTED) 0] ls /
[consumers, storm, brokers, zookeeper]
[zk: 10.208.129.3:2182(CONNECTED) 1] ls /storm
[workerbeats, storms, supervisors, errors, assignments]
[zk: 10.208.129.3:2182(CONNECTED) 2] ls /storm/storms
[rule-hot-config-stat-5-1471598199, traffic-info-stat-8-1471599337, ruleStatTopology-9-1471599960, rule-isp-stat-4-1471598183]
[zk: 10.208.129.3:2182(CONNECTED) 3] ls /consumers   
[]
[zk: 10.208.129.3:2182(CONNECTED) 4] ls /brokers  
[topics]
[zk: 10.208.129.3:2182(CONNECTED) 5] ls /brokers/topics
[]
[zk: 10.208.129.3:2182(CONNECTED) 6] ls /zookeeper     
[quota]
[zk: 10.208.129.3:2182(CONNECTED) 7] ls /zookeeper/quota
[]

10.208.129.4:2181上有Zookeeper,上面跑了Kafka,并且显示启动的Kafka broker id为144、139、141分别对应10.208.129.3、10.208.129.6、10.208.129.11。
[zk: 10.208.129.4:2181(CONNECTED) 0] ls /
[traffic-infoxxx-stat-offset, rule-hot-config-offset, rule-info-stat-offset, traffic-info-stat-offset, zookeeper, rule-isp-stat-offset, consumers, kafka, hive_zookeeper_namespace_hive, rule-config-stat-offset, brokers]
[zk: 10.208.129.4:2181(CONNECTED) 1] ls /kafka
[consumers, config, controller, brokers, admin, controller_epoch]
[zk: 10.208.129.4:2181(CONNECTED) 2] ls /kafka/brokers
[topics, ids]
[zk: 10.208.129.4:2181(CONNECTED) 3] ls /kafka/brokers/ids
[144, 139, 141]

因为Kafka是别的团队搭建的,不好改动。所以只能改Flume的配置文件,修改如下:

agent1.sinks.kafkaSink.type = org.apache.flume.sink.kafka.KafkaSink
agent1.sinks.kafkaSink.topic = TRAFFICxxx_LOG
#agent1.sinks.kafkaSink.brokerList = 10.208.129.3:9092,10.208.129.4:9092,10.208.129.5:9092
#agent1.sinks.kafkaSink.metadata.broker.list = 10.208.129.3:9092,10.208.129.4:9092,10.208.129.5:9092
agent1.sinks.kafkaSink.brokerList = 10.208.129.3:9092,10.208.129.6:9092,10.208.129.11:9092
agent1.sinks.kafkaSink.metadata.broker.list = 10.208.129.3:9092,10.208.129.6:9092,10.208.129.11:9092
agent1.sinks.kafkaSink.producer.type=sync
agent1.sinks.kafkaSink.serializer.class=kafka.serializer.DefaultEncoder
agent1.sinks.kafkaSink.channel = memoryChannel

将broke.list和metadata.broker.list修改为启动了Kafka的ip,一切运行正常

以上是关于Flume连接Kafka的broker出错的主要内容,如果未能解决你的问题,请参考以下文章

Flume连接KAFKA

为啥Kafka消费者连接zookeeper,生产者从broker获取元数据?

kafka连接kafka报错 partitions have leader brokers without a matching listener

当 Spring Boot 容器尝试连接 kafka 容器时出现“Broker 可能不可用”错误

kafkaStream

flume与kafka集成配置