[
https://issues.apache.org/jira/browse/KAFKA-129?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Neha Narkhede updated KAFKA-129:
--------------------------------
Attachment: KAFKA-129.patch
2. Good catch. Updated the patch to include this.
3. While I was making this change, I found a bug in the partitioning approach
of the producer when broker.list option is used. Previously, it choose a random
broker, and then created a produce request with -1 as the partition. This is
not, however, we intend to do partitioning. We let the default partitioner pick
the right broker partition from amongst all available. So we never end up with
a request with -1 as the partition. That test was also written with this buggy
logic. It is using the StaticPartitioner, so the broker partition is
deterministically selected. I fixed the bug as well as the tests to expose the
right behavior.
> 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, 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
