[jira] [Commented] (KAFKA-379) TopicCount.constructTopicCount isn't thread-safe

["Jun Rao (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-379?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13407248#comment-13407248 ] 

Jun Rao commented on KAFKA-379:
-------------------------------

Thanks for the patch. In 0.7, the only usage of JSON is in TopicCount and all usages are called from ZookeeperConsumerConnector.rebalance, which is synchronized. So, it seems that we just need to create a new JSONParser instance per ZookeeperConsumerConnctor instance. Creating 1 JSONParser instance each time a parser is needed may be too expensive.
                
> TopicCount.constructTopicCount isn't thread-safe
> ------------------------------------------------
>
>                 Key: KAFKA-379
>                 URL: https://issues.apache.org/jira/browse/KAFKA-379
>             Project: Kafka
>          Issue Type: Bug
>          Components: clients
>    Affects Versions: 0.7, 0.8
>            Reporter: Nick Howard
>         Attachments: TopicCount.scala.diff
>
>
> TopicCount uses scala.util.parsing.json.JSON, which isn't thread-safe https://issues.scala-lang.org/browse/SI-4929
> If you have multiple consumers within the same JVM, and they all rebalance at the same time, you can get errors like the following:
> [...] kafka.consumer.TopicCount$.constructTopicCount:39] ERROR: error parsing consumer json string [...]
> java.lang.NullPointerException
>         at scala.util.parsing.combinator.Parsers$NoSuccess.<init>(Parsers.scala:131)
>         at scala.util.parsing.combinator.Parsers$Failure.<init>(Parsers.scala:158)
>         at scala.util.parsing.combinator.Parsers$$anonfun$acceptIf$1.apply(Parsers.scala:489)
>         ...
>         at scala.util.parsing.combinator.Parsers$$anon$2.apply(Parsers.scala:742)
>         at scala.util.parsing.json.JSON$.parseRaw(JSON.scala:71)
>         at scala.util.parsing.json.JSON$.parseFull(JSON.scala:85)
>         at kafka.consumer.TopicCount$.constructTopicCount(TopicCount.scala:32)
>         at kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.kafka$consumer$ZookeeperConsumerConnector$ZKRebalancerListener$$getTopicCount(ZookeeperConsumerConnector.scala:422)
>         at kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.kafka$consumer$ZookeeperConsumerConnector$ZKRebalancerListener$$rebalance(ZookeeperConsumerConnector.scala:460)
>         at kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply$mcVI$sp(ZookeeperConsumerConnector.scala:437)
>         at scala.collection.immutable.Range$ByOne$class.foreach$mVc$sp(Range.scala:282)
>         at scala.collection.immutable.Range$$anon$2.foreach$mVc$sp(Range.scala:265)
>         at kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.syncedRebalance(ZookeeperConsumerConnector.scala:433)
>         at kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.handleChildChange(ZookeeperConsumerConnector.scala:375)
>         at org.I0Itec.zkclient.ZkClient$7.run(ZkClient.java:568)
>         at org.I0Itec.zkclient.ZkEventThread.run(ZkEventThread.java:71)
> I ran into this on 0.7.0, but the code in trunk appears to be vulnerable to the same issue.

--
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