[ https://issues.apache.org/jira/browse/KAFKA-1282?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14137118#comment-14137118 ]
nicu marasoiu commented on KAFKA-1282: -------------------------------------- here is a time line: he -> produced he -> consumed [ wait beyond timeout here, connection got closed underneath by the other side] [2014-09-17 15:02:28,689] INFO Got user-level KeeperException when processing sessionid:0x148837ce1800001 type:setData cxid:0x24 zxid:0xec txntype:-1 reqpath:n/a Error Path:/consumers/console-consumer-87959/offsets/topi/0 Error:KeeperErrorCode = NoNode for /consumers/console-consumer-87959/offsets/topi/0 (org.apache.zookeeper.server.PrepRequestProcessor) [2014-09-17 15:02:28,691] INFO Got user-level KeeperException when processing sessionid:0x148837ce1800001 type:create cxid:0x25 zxid:0xed txntype:-1 reqpath:n/a Error Path:/consumers/console-consumer-87959/offsets Error:KeeperErrorCode = NoNode for /consumers/console-consumer-87959/offsets (org.apache.zookeeper.server.PrepRequestProcessor) dddddddddddddd --> produce attempt (never retried, or never reached the broker or at least never reached the consumer) [ many seconds wait, to see if the message is being retried, apparently not, even though the default retry is 3 times] wwwwwwwwwwwwwwwww --> new attempt (immediattely I see the message below with the stack trace, and reconnect + retry is instantly sucesfull) [2014-09-17 15:03:12,599] WARN Failed to send producer request with correlation id 9 to broker 0 with data for partitions [topi,0] (kafka.producer.async.DefaultEventHandler) java.io.IOException: Broken pipe at sun.nio.ch.FileDispatcherImpl.writev0(Native Method) at sun.nio.ch.SocketDispatcher.writev(SocketDispatcher.java:51) at sun.nio.ch.IOUtil.write(IOUtil.java:149) at sun.nio.ch.SocketChannelImpl.write(SocketChannelImpl.java:483) at java.nio.channels.SocketChannel.write(SocketChannel.java:493) at kafka.network.BoundedByteBufferSend.writeTo(BoundedByteBufferSend.scala:56) at kafka.network.Send$class.writeCompletely(Transmission.scala:75) at kafka.network.BoundedByteBufferSend.writeCompletely(BoundedByteBufferSend.scala:26) at kafka.network.BlockingChannel.send(BlockingChannel.scala:100) at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:72) at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:71) 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:255) at kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:106) at kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:100) at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772) at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98) at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98) at scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:226) at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:39) at scala.collection.mutable.HashMap.foreach(HashMap.scala:98) at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771) at kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:100) at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:72) at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:104) at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:87) at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:67) at scala.collection.immutable.Stream.foreach(Stream.scala:547) at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:66) at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:44) [2014-09-17 15:03:12,712] INFO Closing socket connection to /127.0.0.1. (kafka.network.Processor) wwwwwwwwwwwwwwwww > Disconnect idle socket connection in Selector > --------------------------------------------- > > Key: KAFKA-1282 > URL: https://issues.apache.org/jira/browse/KAFKA-1282 > Project: Kafka > Issue Type: Bug > Components: producer > Affects Versions: 0.8.2 > Reporter: Jun Rao > Assignee: nicu marasoiu > Labels: newbie++ > Fix For: 0.9.0 > > Attachments: 1282_brushed_up.patch, > KAFKA-1282_Disconnect_idle_socket_connection_in_Selector.patch > > > To reduce # socket connections, it would be useful for the new producer to > close socket connections that are idle. We can introduce a new producer > config for the idle time. -- This message was sent by Atlassian JIRA (v6.3.4#6332)