4

I've been playing with Apache Kafka for a few days, and here is my problem, If I set up the local test described in the "quick start" section on the website, everything is fine, the kafka producer/ consumer, zookeeper server and kafka broker work perfectly.

Now if I run on a remote server (let's call it node2) : - Zookeeper - port 2181 - Kafka Broker - port 9092 - kafka consumer

And then if I run from my local computer : - kafka producer

Assuming that there is no firewall on node2. The connection end up with a timeout.

Here is the error log : 

/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

And here is my Producer's code : 

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));
  }
}

**So I also tested to switch 

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

by 

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

And in that case I can connect successfully.**

4

1 回答 1

3

请参阅http://kafka.apache.org/faq.html中的第 3 项

解决方法是在 Kafka 的 server.properties 中显式设置主机名属性

您可以使用 Zookeeper 来验证这一点。如果您使用的是 kafka 0.7*，请打开 ZkCli 控制台并获取 /brokers/ids/0，您应该会获取所有代理元数据。确保此处的 IP 地址/主机名与您在生产者代码中使用的 Zk 连接字符串匹配 -

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

就我而言，我正在使用在本地机器上运行的生产者连接到 ubuntu VM（相同的机器，不同的 IP），这种解决方法很有帮助。

于 2013-04-17T22:09:55.573 回答