[
https://issues.apache.org/jira/browse/KAFKA-129?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13120307#comment-13120307
]
Jun Rao commented on KAFKA-129:
-------------------------------
1. import collection.SortedSet is not used in Producer.
2. In ZKBrokerPatitionInfo, it seem that we need to synchronize on
zkWatcherLock in getBrokerInfo, to prevent seeing inconsistent info between
allBrokers and the syncProducer list in ProducerPool.
3. In ProducerTest.testSendSingleMessage, the comment says that we want to send
the request to a random partition, why is the partition number changed from -1
to 0?
> ZK-based producer can throw an unexpceted exception when sending a message
> --------------------------------------------------------------------------
>
> Key: KAFKA-129
> URL: https://issues.apache.org/jira/browse/KAFKA-129
> Project: Kafka
> Issue Type: Bug
> Affects Versions: 0.7
> Reporter: Jun Rao
> Attachments: KAFKA-129.patch
>
>
> Here is a log trace when that happens.
> 2011/08/26 11:25:20.104 FATAL [EmbeddedConsumer]
> [kafka-embedded-consumer-firehoseActivity-0] [kafka]
> java.util.NoSuchElementException: None.getjava.util.NoSuchElementException:
> None.get
> at scala.None$.get(Option.scala:185)
> at scala.None$.get(Option.scala:183)
> at kafka.producer.Producer$$anonfun$3.apply(Producer.scala:115)
> at kafka.producer.Producer$$anonfun$3.apply(Producer.scala:101)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:206)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:206)
> at
> scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:34)
> at
> scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:32)
> at
> scala.collection.TraversableLike$class.map(TraversableLike.scala:206)
> at scala.collection.mutable.WrappedArray.map(WrappedArray.scala:32)
> at kafka.producer.Producer.send(Producer.scala:101)
> at
> kafka.server.EmbeddedConsumer$$anonfun$startNewConsumerThreads$1$$anonfun$apply$1$$anon$1$$anonfun$run$1.apply(KafkaServerStartable.scala:136)
> at
> kafka.server.EmbeddedConsumer$$anonfun$startNewConsumerThreads$1$$anonfun$apply$1$$anon$1$$anonfun$run$1.apply(KafkaServerStartable.scala:134)
> at scala.collection.Iterator$class.foreach(Iterator.scala:631)
> at kafka.utils.IteratorTemplate.foreach(IteratorTemplate.scala:30)
> at scala.collection.IterableLike$class.foreach(IterableLike.scala:79)
> at
> kafka.consumer.KafkaMessageStream.foreach(KafkaMessageStream.scala:29)
> at
> kafka.server.EmbeddedConsumer$$anonfun$startNewConsumerThreads$1$$anonfun$apply$1$$anon$1.run(KafkaServerStartable.scala:134)
> at java.lang.Thread.run(Thread.java:619)
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators:
https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira
