溫馨提示×

溫馨提示×

您好,登錄后才能下訂單哦!

密碼登錄×
登錄注冊×
其他方式登錄
點擊 登錄注冊 即表示同意《億速云用戶服務(wù)條款》

kafka集群發(fā)送消息報錯怎么辦

發(fā)布時間:2021-12-08 17:10:28 來源:億速云 閱讀:165 作者:小新 欄目:云計算

這篇文章將為大家詳細(xì)講解有關(guān)kafka集群發(fā)送消息報錯怎么辦,小編覺得挺實用的,因此分享給大家做個參考,希望大家閱讀完這篇文章后可以有所收獲。

因為logstash采集的日志要發(fā)往kafka做一個隊列機(jī)制,搭建完kafka集群后發(fā)送消息出現(xiàn)問題

ERROR fetching topic metadata for topics [Set(order)] from broker [ArrayBuffer(id:0,host:slave4,port:9092)] failed (kafka.utils.Utils$)
kafka.common.KafkaException: fetching topic metadata for topics [Set(order)] from broker [ArrayBuffer(id:0,host:slave4,port:9092)] failed
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:67)
        at kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82)
        at kafka.producer.async.DefaultEventHandler$$anonfun$handle$2.apply$mcV$sp(DefaultEventHandler.scala:78)
        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:78)
        at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:104)
        at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:87)
        at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:67)
        at scala.collection.immutable.Stream.foreach(Stream.scala:547)
        at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:66)
        at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:44)
Caused by: java.net.ConnectException: Connection refused
        at sun.nio.ch.Net.connect0(Native Method)
        at sun.nio.ch.Net.connect(Net.java:484)
        at sun.nio.ch.Net.connect(Net.java:476)
        at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:675)
        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)
        ... 12 more

ERROR Error in handling batch of 1 events (kafka.producer.async.ProducerSendThread)
kafka.common.FailedToSendMessageException: Failed to send messages after 3 tries.
        at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:90)
        at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:104)
        at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:87)
        at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:67)
        at scala.collection.immutable.Stream.foreach(Stream.scala:547)
        at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:66)
        at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:44)

因為看到 Caused by: java.net.ConnectException: Connection refused ,所以查看 selinux

vim /etc/selinux/config

# This file controls the state of SELinux on the system.
# SELINUX= can take one of these three values:
#     enforcing - SELinux security policy is enforced.
#     permissive - SELinux prints warnings instead of enforcing.
#     disabled - No SELinux policy is loaded.
SELINUX=disabled
# SELINUXTYPE= can take one of these two values:
#     targeted - Targeted processes are protected,
#     mls - Multi Level Security protection.
SELINUXTYPE=targeted

強(qiáng)制關(guān)閉 selinux

setenforce 0

發(fā)送消息還是報一樣的錯

查看zookeeper

[zk: master:2181(CONNECTED) 18] get /brokers/ids/0
{"jmx_port":-1,"timestamp":"1465262832441","host":"localhost","version":1,"port":9092}
cZxid = 0x70000005e
ctime = Tue Jun 07 09:27:12 CST 2016
mZxid = 0x70000005e
mtime = Tue Jun 07 09:27:12 CST 2016
pZxid = 0x70000005e
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x255213e787e0002
dataLength = 86
numChildren = 0

host是localhost,而我的kafka集群是slave4、5上,zk在master,slave2、3上,所以分別修改 slave4、5  上的server.properties的 host.name=slave4, host.name=slave5

重新啟動kafka,查看zk

[zk: master:2181(CONNECTED) 27] get /brokers/ids/0
{"jmx_port":-1,"timestamp":"1465266328389","host":"slave4","version":1,"port":9092}
cZxid = 0x70000008a
ctime = Tue Jun 07 10:25:27 CST 2016
mZxid = 0x70000008a
mtime = Tue Jun 07 10:25:27 CST 2016
pZxid = 0x70000008a
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x355213e7be80003
dataLength = 83
numChildren = 0

0上的host變成了slave4

在發(fā)送消息成功,接收消息成功

關(guān)于“kafka集群發(fā)送消息報錯怎么辦”這篇文章就分享到這里了,希望以上內(nèi)容可以對大家有一定的幫助,使各位可以學(xué)到更多知識,如果覺得文章不錯,請把它分享出去讓更多的人看到。

向AI問一下細(xì)節(jié)

免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進(jìn)行舉報,并提供相關(guān)證據(jù),一經(jīng)查實,將立刻刪除涉嫌侵權(quán)內(nèi)容。

AI