2 u010081710 u010081710 于 2016.03.03 17:03 提问

kafka.common.KafkaException:

package com;

import java.util.Properties;

import java.util.concurrent.TimeUnit;
import kafka.javaapi.producer.Producer;

import kafka.producer.KeyedMessage;

import kafka.producer.ProducerConfig;

import kafka.serializer.StringEncoder;

public class kafkaProducer extends Thread{

private String topic;  

public kafkaProducer(String topic){  
    super();  
    this.topic = topic;  
}  


@Override  
public void run() {  
    Producer producer = createProducer();  
    int i=0;  
    while(true){  
        producer.send(new KeyedMessage<Integer, String>(topic, "message: " + i++));  
        try {  
            TimeUnit.SECONDS.sleep(1);  
        } catch (InterruptedException e) {  
            e.printStackTrace();  
        }  
    }  
}  

private Producer createProducer() {
    Properties properties = new Properties();  
    properties.put("zookeeper.connect", "localhost:2181");//声明zk  
    properties.put("serializer.class", StringEncoder.class.getName());  
    properties.put("metadata.broker.list", "localhost:9092");// 声明kafka broker  
    return new Producer<Integer, String>(new ProducerConfig(properties));  
 }  


public static void main(String[] args) {  
    new kafkaProducer("test").start();// 使用kafka集群中创建好的主题 test   

}  

}

kafka.common.KafkaException: fetching topic metadata for topics [Set(test)] from broker [ArrayBuffer(id:0,host:localhost,port:9092)] failed
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:72)
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:172)
at kafka.utils.Logging$class.swallowError(Logging.scala:106)
at kafka.utils.Utils$.swallowError(Utils.scala:45)
at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:67)
at kafka.producer.Producer.send(Producer.scala:77)
at kafka.javaapi.producer.Producer.send(Producer.scala:33)
at com.kafkaProducer.run(kafkaProducer.java:29)
Caused by: java.nio.channels.ClosedChannelException
at kafka.network.BlockingChannel.send(BlockingChannel.scala:100)
at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:73)
at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:72)
at kafka.producer.SyncProducer.send(SyncProducer.scala:113)
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:58)
... 9 more


2个回答

xiaoyu_BD
xiaoyu_BD   2016.08.25 23:30

你用的是本机kafka,确认kafka启动、如果是集群,建议多指定几台broker,这里 properties.put("metadata.broker.list", "192.168.1.1:9092,192.168.1.2:9092");// 声明kafka broker

a535395302
a535395302   2016.09.10 11:32

同样的问题,我是通过修改bin/server.properties解决的。

版本: kafka_2.11-0.9.0.1

解决方法:
listeners=PLAINTEXT://:9092
修改为
listeners=PLAINTEXT://[IP]:9092
即在双划线和冒号之间加上kafka ip地址。

Csdn user default icon
上传中...
上传图片
插入图片