Hi,
I have a remote server (EC2) setup with Kafka cluster setup. There are 3
brokers each running in the port 9092,9093,9094. The zookeeper is running in
the port 2181.
When I send message to the brokers from my PC, I get an exception which is
given below. I did a dump in the remote server, the request is received in the
remote server.
I am able to locally test the consumer/producer script present in the bin
folder. What am I missing? Can you kindly help me in this error? Any help will
be highly grateful.
[ INFO] [main 2014-01-27 16:06:50,083] Verifying properties
[ INFO] [main 2014-01-27 16:06:50,108] Property metadata.broker.list is
overridden to 54.241.44.129:9092,54.241.44.129:9093,54.241.44.129:9094
[ INFO] [main 2014-01-27 16:06:50,108] Property request.required.acks is
overridden to 1
[ INFO] [main 2014-01-27 16:06:50,108] Property key.serializer.class is
overridden to kafka.serializer.StringEncoder
[ INFO] [main 2014-01-27 16:06:50,108] Property serializer.class is overridden
to kafka.utils.EncryptEncoder
[ INFO] [main 2014-01-27 16:06:50,154] send: encrypted - Message_1
[DEBUG] [main 2014-01-27 16:06:50,298] Handling 1 events [ INFO] [main
2014-01-27 15:59:43,540] Fetching metadata from broker
id:0,host:54.241.44.129,port:9093 with correlation id 0 for 1 topic(s)
Set(mytopic)
[DEBUG] [main 2014-01-27 15:59:43,737] Created socket with SO_TIMEOUT = 10000
(requested 10000), SO_RCVBUF = 8192 (requested -1), SO_SNDBUF = 102400
(requested 102400).
[ INFO] [main 2014-01-27 15:59:43,738] Connected to 54.241.44.129:9093 for
producing
[ INFO] [main 2014-01-27 15:59:44,018] Disconnecting from 54.241.44.129:9093
[DEBUG] [main 2014-01-27 15:59:44,025] Successfully fetched metadata for 1
topic(s) Set(mytopic)
[DEBUG] [main 2014-01-27 15:59:44,058] Getting broker partition info for topic
mytopic
[DEBUG] [main 2014-01-27 15:59:44,060] Partition [mytopic,0] has leader 2
[DEBUG] [main 2014-01-27 15:59:44,072] Broker partitions registered for topic:
mytopic are 0
[DEBUG] [main 2014-01-27 15:59:44,091] Sending 1 messages with no compression
to [mytopic,0]
[DEBUG] [main 2014-01-27 15:59:44,109] Producer sending messages with
correlation id 2 for topics [mytopic,0] to broker 2 on
ip-10-199-31-87.us-west-1.compute.internal:9093
[ERROR] [main 2014-01-27 15:59:44,129] Producer connection to
ip-10-199-31-87.us-west-1.compute.internal:9093 unsuccessful
java.nio.channels.UnresolvedAddressException
at sun.nio.ch.Net.checkAddress(Net.java:127)
at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:640)
at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57)
at kafka.producer.SyncProducer.connect(SyncProducer.scala:146)
at
kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:161)
at
kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:68)
at
kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SyncProducer.scala:102)
at
kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply(SyncProducer.scala:102)
at
kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply(SyncProducer.scala:102)
at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
at
kafka.producer.SyncProducer$$anonfun$send$1.apply$mcV$sp(SyncProducer.scala:101)
at
kafka.producer.SyncProducer$$anonfun$send$1.apply(SyncProducer.scala:101)
at
kafka.producer.SyncProducer$$anonfun$send$1.apply(SyncProducer.scala:101)
at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
at kafka.producer.SyncProducer.send(SyncProducer.scala:100)
at
kafka.producer.async.DefaultEventHandler.kafka$producer$async$DefaultEventHandler$$send(DefaultEventHandler.scala:254)
at
kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$1.apply(DefaultEventHandler.scala:106)
at
kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$1.apply(DefaultEventHandler.scala:100)
at
scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:80)
at
scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:80)
at scala.collection.Iterator$class.foreach(Iterator.scala:631)
at
scala.collection.mutable.HashTable$$anon$1.foreach(HashTable.scala:161)
at
scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:194)
at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:39)
at scala.collection.mutable.HashMap.foreach(HashMap.scala:80)
at
kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:100)
at
kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:72)
at kafka.producer.Producer.send(Producer.scala:76)
at kafka.javaapi.producer.Producer.send(Producer.scala:33)
at kafka.application.KafkaProducer.sendMessage(KafkaProducer.java:39)
at kafka.test.KafkaProducerTest.main(KafkaProducerTest.java:21)
[ WARN] [main 2014-01-27 15:59:44,139] Failed to send producer request with
correlation id 2 to broker 2 with data for partitions [mytopic,0]
Thanks
Bala
