I wouldn't say that Kafka's making it difficult. The cloud environment is making it difficult. The VM that the Kafka broker is running on can only see it's private IP (at the OS level) so you have to add the advertised.host.name config so that it knows what public IP is assigned to it.
On Fri, Jun 13, 2014 at 8:32 AM, <jamesw...@yahoo.com.tw> wrote: > Hi, > I found this caused by broker's config. I miss typed advertise.host.name. > Actually it should be advertised.host.name. > > However, I still don't know what's different between host.name and > advertised.host.name. Why Kafka makes it so difficult to connect from > public ip. Thanks your reply. > > Best regards, > James > > > > > > Jun Rao <jun...@gmail.com> 於 2014/6/13 下午10:22 寫道： > > > > Could you telnet to the public ip from your local machine? > > > > Thanks, > > > > Jun > > > > > >> On Thu, Jun 12, 2014 at 4:04 AM, 傅駿浩 <jamesw...@yahoo.com.tw> wrote: > >> > >> Hi all, > >> > >> I want to use the KafkaSpout(where is > >> > https://github.com/apache/incubator-storm/tree/master/external/storm-kafka > ) > >> to read data from kafka. But my kafka and zookeeper cluster are builted > on > >> GCE(google cloud engine) and my storm testing is on local machine. The > >> problem is I cannot set public IP by kafka config - server.properties to > >> reach from my local(out of kafka's domain) storm machines. > >> ***My server.properties is as follows: > >> host.name=[Brokers's Private IP] > >> advertise.port=9092 > >> advertise.host.name=[Brokers's Public IP] > >> > >> ***My KafkaSpout's config in Java is as follows: > >> BrokerHosts a = new ZkHosts("[IP1],[IP2],[IP3]","/brokers"); > >> SpoutConfig kafkaConf = new SpoutConfig(a, topic, "/stormkafka", > >> topic);//Using [topic] as id > >> kafkaConf.scheme = new SchemeAsMultiScheme(new StringScheme()); > >> kafkaConf.zkServers = new ArrayList<String>() { > >> { > >> add("[IP1]"); > >> add("[IP1]"); > >> add("[IP1]"); > >> } > >> }; > >> kafkaConf.zkPort = 2181; > >> kafkaConf.forceFromStart = true; > >> ps. I have 3-node kafka cluster and 3-node zookeeper cluster on 3 > >> machines. That's to say, Each machine with a broker and zookeeper node > >> > >> ====================================== > >> So the problem is it seens not to connect from the domain different from > >> kafka's domain, e.g.my local storm machine. This is a big problem if I > >> need to communcate with kafka out of its domain. The error exception is > as > >> follows, and can anyone help me with this issue. How can I config it in > a > >> right way so that I can subscribe topic of kafka from my local > >> machine(storm). > >> Thanks a lot :) > >> > >> > >> > >> java.lang.RuntimeException: java.net.NoRouteToHostException: > 找不到通往該主機的路由路徑 > >> at storm.kafka.ZkCoordinator.refresh(ZkCoordinator.java:83) > >> ~[storm-kafka-0.8-plus-0.4.0.jar:na] > >> at > storm.kafka.ZkCoordinator.getMyManagedPartitions(ZkCoordinator.java:45) > >> ~[storm-kafka-0.8-plus-0.4.0.jar:na] > >> at storm.kafka.KafkaSpout.nextTuple(KafkaSpout.java:118) > >> ~[storm-kafka-0.8-plus-0.4.0.jar:na] > >> at > >> > backtype.storm.daemon.executor$eval5100$fn__5101$fn__5116$fn__5145.invoke(executor.clj:562) > >> ~[na:na] > >> at backtype.storm.util$async_loop$fn__390.invoke(util.clj:433) ~[na:na] > >> at clojure.lang.AFn.run(AFn.java:24) [clojure-1.4.0.jar:na] > >> at java.lang.Thread.run(Thread.java:745) [na:1.7.0_60] > >> Caused by: java.net.NoRouteToHostException: 找不到通往該主機的路由路徑 > >> at sun.nio.ch.Net.connect0(Native Method) ~[na:1.7.0_60] > >> at sun.nio.ch.Net.connect(Net.java:465) ~[na:1.7.0_60] > >> at sun.nio.ch.Net.connect(Net.java:457) ~[na:1.7.0_60] > >> at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:670) > >> ~[na:1.7.0_60] > >> at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57) > >> ~[kafka_2.10-0.8.1.1.jar:na] > >> at kafka.consumer.SimpleConsumer.connect(SimpleConsumer.scala:44) > >> ~[kafka_2.10-0.8.1.1.jar:na] > >> at > >> > kafka.consumer.SimpleConsumer.getOrMakeConnection(SimpleConsumer.scala:142) > >> ~[kafka_2.10-0.8.1.1.jar:na] > >> at > >> > kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:69) > >> ~[kafka_2.10-0.8.1.1.jar:na] > >> at > >> kafka.consumer.SimpleConsumer.getOffsetsBefore(SimpleConsumer.scala:124) > >> ~[kafka_2.10-0.8.1.1.jar:na] > >> at > >> > kafka.javaapi.consumer.SimpleConsumer.getOffsetsBefore(SimpleConsumer.scala:79) > >> ~[kafka_2.10-0.8.1.1.jar:na] > >> at storm.kafka.KafkaUtils.getOffset(KafkaUtils.java:55) > >> ~[storm-kafka-0.8-plus-0.4.0.jar:na] > >> at storm.kafka.KafkaUtils.getOffset(KafkaUtils.java:45) > >> ~[storm-kafka-0.8-plus-0.4.0.jar:na] > >> at storm.kafka.PartitionManager.<init>(PartitionManager.java:77) > >> ~[storm-kafka-0.8-plus-0.4.0.jar:na] > >> at storm.kafka.ZkCoordinator.refresh(ZkCoordinator.java:78) > >> ~[storm-kafka-0.8-plus-0.4.0.jar:na] > >> ... 6 common frames omitted > >> > >> Best regards, > >> James Fu >
