Не удается подключиться к удаленному зоопарку от производителя Kafka

Question

Я играю с Apache Kafka в течение нескольких дней, и вот моя проблема: Если я настроил локальный тест, описанный в разделе «Быстрый старт» на веб-сайте все в порядке, производитель/потребитель кафки, сервер zookeeper и брокер kafka отлично работают.

Теперь, если я бегу на удаленном сервере (назовем его node2): - Zookeeper - порт 2181 - Кафка Broker - порт 9092 - Кафка потребитель

И потом, если я бегу из моего локального компьютера: - производитель kafka

Предполагая, что на узле2 нет брандмауэра. Соединение заканчивается таймаутом.

Вот журнал ошибок:

/etc/java/jdk1.6.0_41/bin/java -Didea.launcher.port=7533 -Didea.launcher.bin.path=/home/kevin/Documents/idea-IU-123.169/bin -Dfile.encoding=UTF-8 -classpath /etc/java/jdk1.6.0_41/lib/dt.jar:/etc/java/jdk1.6.0_41/lib/tools.jar:/etc/java/jdk1.6.0_41/lib/jconsole.jar:/etc/java/jdk1.6.0_41/lib/htmlconverter.jar:/etc/java/jdk1.6.0_41/lib/sa-jdi.jar:/home/kevin/Desktop/kafka-0.7.2/examples/target/scala_2.8.0/classes:/home/kevin/Desktop/kafka-0.7.2/project/boot/scala-2.8.0/lib/scala-compiler.jar:/home/kevin/Desktop/kafka-0.7.2/project/boot/scala-2.8.0/lib/scala-library.jar:/home/kevin/Desktop/kafka-0.7.2/core/target/scala_2.8.0/classes:/home/kevin/Desktop/kafka-0.7.2/core/lib_managed/scala_2.8.0/compile/jopt-simple-3.2.jar:/home/kevin/Desktop/kafka-0.7.2/core/lib_managed/scala_2.8.0/compile/log4j-1.2.15.jar:/home/kevin/Desktop/kafka-0.7.2/core/lib_managed/scala_2.8.0/compile/zookeeper-3.3.4.jar:/home/kevin/Desktop/kafka-0.7.2/core/lib_managed/scala_2.8.0/compile/zkclient-0.1.jar:/home/kevin/Desktop/kafka-0.7.2/core/lib_managed/scala_2.8.0/compile/snappy-java-1.0.4.1.jar:/home/kevin/Desktop/kafka-0.7.2/examples/lib_managed/scala_2.8.0/compile/jopt-simple-3.2.jar:/home/kevin/Desktop/kafka-0.7.2/examples/lib_managed/scala_2.8.0/compile/log4j-1.2.15.jar:/home/kevin/Documents/idea-IU-123.169/lib/idea_rt.jar com.intellij.rt.execution.application.AppMain kafka.examples.KafkaConsumerProducerDemo 
log4j:WARN No appenders could be found for logger (org.I0Itec.zkclient.ZkConnection). 
log4j:WARN Please initialize the log4j system properly. 
Exception in thread "Thread-0" java.net.ConnectException: Connection timed out 
    at sun.nio.ch.Net.connect(Native Method) 
    at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:532) 
    at kafka.producer.SyncProducer.connect(SyncProducer.scala:173) 
    at kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:196) 
    at kafka.producer.SyncProducer.send(SyncProducer.scala:92) 
    at kafka.producer.SyncProducer.send(SyncProducer.scala:125) 
    at kafka.producer.ProducerPool$$anonfun$send$1.apply$mcVI$sp(ProducerPool.scala:114) 
    at kafka.producer.ProducerPool$$anonfun$send$1.apply(ProducerPool.scala:100) 
    at kafka.producer.ProducerPool$$anonfun$send$1.apply(ProducerPool.scala:100) 
    at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:57) 
    at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:43) 
    at kafka.producer.ProducerPool.send(ProducerPool.scala:100) 
    at kafka.producer.Producer.zkSend(Producer.scala:137) 
    at kafka.producer.Producer.send(Producer.scala:99) 
    at kafka.javaapi.producer.Producer.send(Producer.scala:103) 
    at kafka.examples.Producer.run(Producer.java:53) 

Process finished with exit code 0 

А вот код моего продюсера:

import java.util.Properties; 
import kafka.javaapi.producer.ProducerData; 
import kafka.producer.ProducerConfig; 


public class Producer extends Thread{ 

    private final kafka.javaapi.producer.Producer<String, String> producer; 
    private final String topic; 
    private final Properties props = new Properties(); 

    public Producer(String topic) 
    { 
    props.put("zk.connect", "node2:2181"); 
    props.put("connect.timeout.ms", "5000"); 
    props.put("socket.timeout.ms", "30000"); 
    props.put("serializer.class", "kafka.serializer.StringEncoder"); 
    props.put("producer.type", "sync"); 
    props.put("conpression.codec", "0"); 
    producer = new kafka.javaapi.producer.Producer<String, String>(new ProducerConfig(props)); 
    this.topic = topic; 
    } 

    public void run() { 
     String messageStr = new String("Message_test"); 
     producer.send(new ProducerData<String, String>(topic, messageStr)); 
    } 
} 

** Я также протестировали для переключения

props.put("zk.connect", "node2:2181"); 

по

props.put("broker.list", "0:node2:9082"); 

И в этом случае я могу подключить успешно. **

Answer 1

См пункт # 3 в http://kafka.apache.org/faq.html

Для устранения этой проблемы явно задать имя хоста свойство в server.properties Кафки

Вы можете проверить это используя Zookeeper. Если вы используете kafka 0.7 *, откройте консоль ZkCli и получите get/brokers/ids/0, и вы должны получить все метаданные брокеров. Убедитесь, что IP-адрес/имя хоста Он соответствует Zk строку подключения вы используете в коде производителя -

props.put("zk.connect", "node2:2181"); 

В моем случае, я использовал производитель, работающий на моей локальной машине, соединяющей к убунту VM (то же самое окно, другой IP), и это обходное решение помогло.