2015-03-17 4 views
6

Я установил kafka на одном узле и запустил zookeeper, а также kafka server.I проверил его для внутреннего производителя и потребителя на консоли, и он работает хорошо. Но когда я Я запускаю внутреннего пользователя kafka на консоли, и мой пользовательский производитель он не работает.Невозможно связаться с сервером kafka с использованием kafka API-интерфейса производителя

Ниже мой продюсер класс

Properties props = new Properties(); 

    props.put("metadata.broker.list", "xx.xx.xx.xx:9092"); 
    props.put("serializer.class", "kafka.serializer.StringEncoder"); 
    props.put("partitioner.class", "com.example.producer.SimplePartitioner"); 
    props.put("request.required.acks", "1"); 

    ProducerConfig config = new ProducerConfig(props); 

    Producer<String, String> producer = new Producer<String, String>(config); 
    KeyedMessage<String, String> data = new KeyedMessage<String, String>(
      "mails", "xxxx"); 
    producer.send(data); 

Когда управление достигает producer.send(), он останавливается после 3-х попыток со следующим исключением

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) 
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: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:78) 
at kafka.producer.Producer.send(Producer.scala:77) 
at kafka.javaapi.producer.Producer.send(Producer.scala:33) 
+0

Вы можете попробовать комментировать «props.put (« partitioner.class »,« com.example.producer.SimplePartitioner »)' part – user2720864

ответ

3

Я пытался подключиться к Кафка сервер из класса Producer в eclipse из внешней VM. Мне пришлось заменить localhost на ip-адрес в файле manufacturer.properties в config/kafka.

+0

также вставить IP-адрес в файл/etc/hosts – Shams

 Смежные вопросы

  • Нет связанных вопросов^_^