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

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

Flume连接Kafka的broker出错的更多相关文章

Flume连接oracle实时推送数据到kafka
版本号: RedHat6.5 JDK1.8 flume-1.6.0 kafka_2.11-0.8.2.1 flume安装 RedHat6.5安装单机flume1.6:RedHat6.5安 ...
物联网架构成长之路(8)-EMQ-Hook了解、连接Kafka发送消息
1. 前言按照我自己设计的物联网框架,对于MQTT集群中的所有消息,是要持久化到磁盘的,这里采用一个消息队列中间件Kafka作为数据缓冲,缓冲结果存到数据仓库中,以供后续作为数据分析.由于MQTT集 ...
Kafka实战－Flume到Kafka
1.概述前面给大家介绍了整个Kafka项目的开发流程,今天给大家分享Kafka如何获取数据源,即Kafka生产数据.下面是今天要分享的目录: 数据来源 Flume到Kafka 数据源加载预览下面 ...
flume从kafka中读取数据
a1.sources = r1 a1.sinks = k1 a1.channels = c1 #使用内置kafka source a1.sources.r1.type = org.apache.flu ...
【转】Kafka实战－Flume到Kafka
Kafka实战-Flume到Kafka Kafka 2015-07-03 08:46:24 发布您的评价: 0.0 收藏 2收藏 1.概述前面给大家介绍了整个Kafka ...
Flume与Kafka集成
一.Flume介绍 Flume是一个分布式.可靠.和高可用的海量日志聚合的系统,支持在系统中定制各类数据发送方,用于收集数据:同时,Flume提供对数据进行简单处理,并写到各种数据接受方(可定制)的能 ...
消息系统Flume与Kafka的区别
首先Flume和Kafka都是消息系统,但是它俩也有着很多不同的地方,Flume更趋向于消息采集系统,而Kafka更趋向于消息缓存系统. [一]设计上的不同 Flume是消息采集系统,它主要解决问题是 ...
使用Flume消费Kafka数据到HDFS
1.概述对于数据的转发,Kafka是一个不错的选择.Kafka能够装载数据到消息队列,然后等待其他业务场景去消费这些数据,Kafka的应用接口API非常的丰富,支持各种存储介质,例如HDFS.HBa ...
ambari下的flume和kafka整合
1.配置flume #扫描指定文件配置 agent.sources = s1 agent.channels = c1 agent.sinks = k1 agent.sources.s1.type=ex ...

随机推荐

xshell5激活
Xshell是一个用于MS Windows平台的强大的SSH,TELNET,和RLOGIN终端仿真软件.它使得用户能轻松和安全地从Windows PC*问Unix/Linux主机.以上内容全部为广告 ...
pythonchallenge之C++学习篇-03
提示说一个小写字母两面精确地被大写字母包围,应该指的是周围没有四个而仅仅这两个像这样的:xXXXxXXXx的中间的那个应该是符合条件的好了标题是re,提示该是使用正则表达式,网页源码里有待处理的字符 ...
c&num;获取枚举
在实际开发项目中,我们定义了一个枚举,往往我们需要在下拉框或其它地方展示枚举.为了加深印象,也为了帮到有需要的人,我写了一个DEMO. 第一步,我们定义一个枚举: /// <summary&gt ...
详解eNSP下的PPP之MP、PAP/CHAP认证实验配置
一.PPP MP实验(用虚拟模板配置) 1.拓扑图
查询一个ID出现2种结果的情况
项目中书籍分个人和机构,分属不同的表所以有的时候ID是一样的,那么只根据ID查询书籍就会存在ID=xxx的既有个人又有机构,而通常我们可能只需要一个,多的没做区分就出问题了! 所以数据统一做查询的时 ...
CSS hack——不同浏览器的CSS应对法
1.IE条件注释法: lte表示“小于等于”,“lt”表示“小于”,“gte”表示“大于等于”,“gt”表示“大于”,“!”表示“不等于”. <!--[if IE 6]> <link ...
Python学习之 OS模块
1.目录操作 import os os.mkdir('abc') #创建abc文件 tree a #查看目录结构 2.目录遍历方式一:递归 import os def dirList(path ...
提取图像(tif)中水体的矢量数据（shp）研究
方法一：1、利用envi打开tif数据，原投影信息为beijing54.envi中没有这个投影。这里选择投影信息（WGS-84）选取水体roi——进行监督分类。这里可以对分类后进行处理（消除文字等干扰 ...
silverlight visifire控件图表制作——silverlight 后台方法ControlChart&period;xaml&period;cs
一.构造方法ControlChart 1.前台页面控件赋值 //时间下拉框赋值,下拉框赋选定值 for (int ii = DateTime.Today.Year; ii ...
clion idea jetbrain windows下搞c/c++
安装 clion 把MinGW(搜MinGW.zip)放到c盘根目录下 ok