Hi Team, I am bit new to kafka. I was trying to set up kafka 0.8 and connect it to druid firehose. My console producer is working fine but when I try to connect to it using java code I am getting socket timeout exception even when increasing the timeout to 2 min.
We are getting socket timeout when fetching metadata from broker. Here is the stack trace [ERROR] 2014-08-10 20:52:02,671 AsyncAppender-Worker-Thread-5 [airpricingservice k.p.async.DefaultEventHandler] - [] Failed to collate messages by topic, partition due to: fetching topic metadata for topics [Set(wikipedia)] from broker [ArrayBuffer(id:0,host:localhost,port:9092)] failed [ERROR] 2014-08-10 20:52:12,774 AsyncAppender-Worker-Thread-5 [airpricingservice kafka.utils.Utils$] - [] fetching topic metadata for topics [Set(wikipedia)] from broker [ArrayBuffer(id:0,host:localhost,port:9092)] failed kafka.common.KafkaException: fetching topic metadata for topics [Set(wikipedia)] from broker [ArrayBuffer(id:0,host:localhost,port:9092)] failed at kafka.client.ClientUtils$.fetchTopicMetadata(Unknown Source) ~[AirPricingService-1.0.4-SNAPSHOT.jar:na] at kafka.producer.BrokerPartitionInfo.updateInfo(Unknown Source) ~[AirPricingService-1.0.4-SNAPSHOT.jar:na] at kafka.producer.async.DefaultEventHandler$$anonfun$handle$2.apply$mcV$sp(Unknown Source) ~[AirPricingService-1.0.4-SNAPSHOT.jar:na] at kafka.utils.Utils$.swallow(Unknown Source) [AirPricingService-1.0.4-SNAPSHOT.jar:na] at kafka.utils.Logging$class.swallowError(Unknown Source) [AirPricingService-1.0.4-SNAPSHOT.jar:na] at kafka.utils.Utils$.swallowError(Unknown Source) [AirPricingService-1.0.4-SNAPSHOT.jar:na] at kafka.producer.async.DefaultEventHandler.handle(Unknown Source) [AirPricingService-1.0.4-SNAPSHOT.jar:na] at kafka.producer.Producer.send(Unknown Source) [AirPricingService-1.0.4-SNAPSHOT.jar:na] at kafka.javaapi.producer.Producer.send(Unknown Source) [AirPricingService-1.0.4-SNAPSHOT.jar:na] at com.expedia.service.air.pricing.manager.logger.appender.KafkaAppender.append(KafkaAppender.java:69) [AirPricingService-1.0.4-SNAPSHOT.jar:na] at com.expedia.service.air.pricing.manager.logger.appender.KafkaAppender.append(KafkaAppender.java:12) [AirPricingService-1.0.4-SNAPSHOT.jar:na] at ch.qos.logback.core.AppenderBase.doAppend(AppenderBase.java:85) [AirPricingService-1.0.4-SNAPSHOT.jar:na] at ch.qos.logback.core.spi.AppenderAttachableImpl.appendLoopOnAppenders(AppenderAttachableImpl.java:48) [AirPricingService-1.0.4-SNAPSHOT.jar:na] at ch.qos.logback.core.AsyncAppenderBase$Worker.run(AsyncAppenderBase.java:226) [AirPricingService-1.0.4-SNAPSHOT.jar:na] Caused by: java.net.SocketTimeoutException: null at sun.nio.ch.SocketAdaptor$SocketInputStream.read(SocketAdaptor.java:229) ~[na:1.7.0_40] at sun.nio.ch.ChannelInputStream.read(ChannelInputStream.java:103) ~[na:1.7.0_40] at java.nio.channels.Channels$ReadableByteChannelImpl.read(Channels.java:385) ~[na:1.7.0_40] at kafka.utils.Utils$.read(Unknown Source) [AirPricingService-1.0.4-SNAPSHOT.jar:na] at kafka.network.BoundedByteBufferReceive.readFrom(Unknown Source) ~[AirPricingService-1.0.4-SNAPSHOT.jar:na] at kafka.network.Receive$class.readCompletely(Unknown Source) ~[AirPricingService-1.0.4-SNAPSHOT.jar:na] at kafka.network.BoundedByteBufferReceive.readCompletely(Unknown Source) ~[AirPricingService-1.0.4-SNAPSHOT.jar:na] at kafka.network.BlockingChannel.receive(Unknown Source) ~[AirPricingService-1.0.4-SNAPSHOT.jar:na] at kafka.producer.SyncProducer.liftedTree1$1(Unknown Source) ~[AirPricingService-1.0.4-SNAPSHOT.jar:na] at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(Unknown Source) ~[AirPricingService-1.0.4-SNAPSHOT.jar:na] at kafka.producer.SyncProducer.send(Unknown Source) ~[AirPricingService-1.0.4-SNAPSHOT.jar:na] ... 14 common frames omitted I am suspecting some config mismatch but my console producer is setting is same as my java producer. Thanks, Aayush
