kafka集群发送消息报错怎么办

蜗牛互联网技术资讯 2021-12-08 253 0

这篇文章将为大家详细讲解有关kafka集群发送消息报错怎么办，小编觉得挺实用的，因此分享给大家做个参考，希望大家阅读完这篇文章后可以有所收获。

因为logstash采集的日志要发往kafka做一个队列机制，搭建完kafka集群后发送消息出现问题

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

强制关闭 selinux

setenforce 0

发送消息还是报一样的错

查看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

在发送消息成功，接收消息成功

关于“kafka集群发送消息报错怎么办”这篇文章就分享到这里了，希望以上内容可以对大家有一定的帮助，使各位可以学到更多知识，如果觉得文章不错，请把它分享出去让更多的人看到。

免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：niceseo99@gmail.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。