0

我正在尝试在 HDP 上运行简单的 kafka 生产者消费者示例,但面临以下异常。

[2016-03-03 18:26:38,683] WARN Fetching topic metadata with correlation id 0 for topics [Set(page_visits)] from broker [BrokerEndPoint(0,sandbox.hortonworks.com,9092)] failed (kafka.client.ClientUtils$)
java.nio.channels.ClosedChannelException
        at kafka.network.BlockingChannel.send(BlockingChannel.scala:120)
        at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:75)
        at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:74)
        at kafka.producer.SyncProducer.send(SyncProducer.scala:115)
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
        at kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82)
        at kafka.producer.async.DefaultEventHandler$$anonfun$handle$1.apply$mcV$sp(DefaultEventHandler.scala:68)
        at kafka.utils.CoreUtils$.swallow(CoreUtils.scala:89)
        at kafka.utils.Logging$class.swallowError(Logging.scala:106)
        at kafka.utils.CoreUtils$.swallowError(CoreUtils.scala:51)
        at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:68)
        at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:105)
        at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:88)
        at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:68)
        at scala.collection.immutable.Stream.foreach(Stream.scala:547)
        at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67)
        at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
[2016-03-03 18:26:38,688] ERROR fetching topic metadata for topics [Set(page_visits)] from broker [ArrayBuffer(BrokerEndPoint(0,sandbox.hortonworks.com,9092))] failed (kafka.utils.CoreUtils$)
kafka.common.KafkaException: fetching topic metadata for topics [Set(page_visits)] from broker [ArrayBuffer(BrokerEndPoint(0,sandbox.hortonworks.com,9092))] failed
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:73)
        at kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82)
        at kafka.producer.async.DefaultEventHandler$$anonfun$handle$1.apply$mcV$sp(DefaultEventHandler.scala:68)
        at kafka.utils.CoreUtils$.swallow(CoreUtils.scala:89)
        at kafka.utils.Logging$class.swallowError(Logging.scala:106)
        at kafka.utils.CoreUtils$.swallowError(CoreUtils.scala:51)
        at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:68)
        at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:105)
        at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:88)
        at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:68)
        at scala.collection.immutable.Stream.foreach(Stream.scala:547)
        at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67)
        at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
Caused by: java.nio.channels.ClosedChannelException
        at kafka.network.BlockingChannel.send(BlockingChannel.scala:120)
        at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:75)
        at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:74)
        at kafka.producer.SyncProducer.send(SyncProducer.scala:115)
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
        ... 12 more
[2016-03-03 18:26:38,693] WARN Fetching topic metadata with correlation id 1 for topics [Set(page_visits)] from broker [BrokerEndPoint(0,sandbox.hortonworks.com,9092)] failed (kafka.client.ClientUtils$)
java.nio.channels.ClosedChannelException

这是我用于生产者的命令。

./kafka-console-producer.sh --broker-list sandbox.hortonworks.com:9092  --topic page_visits

做了一些谷歌搜索后,我发现我需要在 server.properties 文件中添加adverted.host.name 属性。这是我的 server.properties 文件。

# Generated by Apache Ambari. Thu Mar  3 18:12:50 2016

advertised.host.name=sandbox.hortonworks.com
auto.create.topics.enable=true
auto.leader.rebalance.enable=true
broker.id=0
compression.type=producer
controlled.shutdown.enable=true
controlled.shutdown.max.retries=3
controlled.shutdown.retry.backoff.ms=5000
controller.message.queue.size=10
controller.socket.timeout.ms=30000
default.replication.factor=1
delete.topic.enable=false
fetch.purgatory.purge.interval.requests=10000
host.name=sandbox.hortonworks.com
kafka.ganglia.metrics.group=kafka
kafka.ganglia.metrics.host=localhost
kafka.ganglia.metrics.port=8671
kafka.ganglia.metrics.reporter.enabled=true
kafka.metrics.reporters=org.apache.hadoop.metrics2.sink.kafka.KafkaTimelineMetricsReporter
kafka.timeline.metrics.host=sandbox.hortonworks.com
kafka.timeline.metrics.maxRowCacheSize=10000
kafka.timeline.metrics.port=6188
kafka.timeline.metrics.reporter.enabled=true
kafka.timeline.metrics.reporter.sendInterval=5900
leader.imbalance.check.interval.seconds=300
leader.imbalance.per.broker.percentage=10
listeners=PLAINTEXT://sandbox.hortonworks.com:6667
log.cleanup.interval.mins=10
log.dirs=/kafka-logs
log.index.interval.bytes=4096
log.index.size.max.bytes=10485760
log.retention.bytes=-1
log.retention.hours=168
log.roll.hours=168
log.segment.bytes=1073741824
message.max.bytes=1000000
min.insync.replicas=1
num.io.threads=8
num.network.threads=3
num.partitions=1
num.recovery.threads.per.data.dir=1
num.replica.fetchers=1
offset.metadata.max.bytes=4096
offsets.commit.required.acks=-1
offsets.commit.timeout.ms=5000
offsets.load.buffer.size=5242880
offsets.retention.check.interval.ms=600000
offsets.retention.minutes=86400000
offsets.topic.compression.codec=0
offsets.topic.num.partitions=50
offsets.topic.replication.factor=3
offsets.topic.segment.bytes=104857600
producer.purgatory.purge.interval.requests=10000
queued.max.requests=500
replica.fetch.max.bytes=1048576
replica.fetch.min.bytes=1
replica.fetch.wait.max.ms=500
replica.high.watermark.checkpoint.interval.ms=5000
replica.lag.max.messages=4000
replica.lag.time.max.ms=10000
replica.socket.receive.buffer.bytes=65536
replica.socket.timeout.ms=30000
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
socket.send.buffer.bytes=102400
zookeeper.connect=sandbox.hortonworks.com:2181
zookeeper.connection.timeout.ms=15000
zookeeper.session.timeout.ms=30000
zookeeper.sync.time.ms=2000

添加属性后,我得到了同样的异常。

任何建议。

4

3 回答 3

0

/usr/hdp/current/kafka-broker/config/server.properties用以下 2 行修改了文件

advertised.host.name=sandbox.hortonworks.com
listeners=PLAINTEXT://sandbox.hortonworks.com:6667,PLAINTEXT://0.0.0.0:6667 

运行以下执行命令

 ./kafka-console-producer.sh --broker-list sandbox.hortonworks.com:6667 --topic tst2
 ./kafka-console-consumer.sh --zookeeper localhost:2181 --topic tst2 --from-beginning

有了这个它的工作正常

于 2016-04-22T06:26:33.113 回答
0

根据日志,似乎 kafka 服务器(代理)没有运行。代理服务器应该首先运行。

生产者和消费者是客户端程序,它们也将与代理服务器和动物园管理员进行交互。

在运行生产者或消费者之前,请检查 broker 和 zookeeper 是否运行成功。

运行服务器 ./kafka-server-start.sh ../config/server.properties 检查日志是否有任何错误,如果没有错误,则开始向服务器生成消息。

还要检查 zookeeper 服务。

于 2016-03-04T02:17:35.247 回答
0

我有类似的问题。首先,我检查了 Ambari 中 Kafka 代理的 listeners属性在此处输入图像描述

也可以通过以下方式检查:

[root@sandbox bin]# cat /usr/hdp/current/kafka-broker/conf/server.properties  | grep listeners
listeners=PLAINTEXT://sandbox.hortonworks.com:6667

如您所见,Ambari 将 localhost 替换为主机名,并且端口相同 - 6667。

然后我检查了代理是否真的在那个端口上监听:

[root@sandbox bin]# netstat -tulpn | grep 6667
tcp        0      0 10.0.2.15:6667              0.0.0.0:*                   LISTEN      11137/java

下一步是启动生产者:

./kafka-console-producer.sh --broker-list 10.0.2.15:6667 --topic test

最后我推出了消费者:

 ./kafka-console-consumer.sh --zookeeper 10.0.2.15:2181 --topic test --from-beginning

在生产者端按 Enter 输入几个单词后,消费者收到消息。

于 2016-04-23T23:35:36.493 回答