[jira] [Created] (KAFKA-2284) ConsumerRebalanceListener receives wrong type in partitionOwnership values

["E. Sammer (JIRA)" <ji...@apache.org>]

E. Sammer created KAFKA-2284:
--------------------------------

             Summary: ConsumerRebalanceListener receives wrong type in partitionOwnership values
                 Key: KAFKA-2284
                 URL: https://issues.apache.org/jira/browse/KAFKA-2284
             Project: Kafka
          Issue Type: Bug
          Components: consumer
    Affects Versions: 0.8.2.0
            Reporter: E. Sammer
            Assignee: Neha Narkhede
            Priority: Blocker


The ConsumerRebalanceListener's beforeReleasingPartitions() method is supposed to receive an arg of Map<String, Set<Integer>> (topic -> Set(partitions)). Even though the type of the map value is specified as a java.util.Set, a scala.collection.convert.Wrappers$JSetWrapper is passed instead which does not implement Set<T> causing a class cast exception as soon as one attempts to access any value of the map. It looks as if this method was never tested against the actual types specified by the interface.

Here's what happens if you call {{Set<T> foo = partitionOwnership.get(topic)}}:

{code}
2015-06-18 07:28:43,776 (search-consumer_esammer-mbp.local-1434637723383-12126c1b_watcher_executor) [WARN - com.rocana.search.consumer.ConsumerService$1.beforeReleasingPartitions(ConsumerService.java:246)] Exception while rebalancing!
java.lang.ClassCastException: scala.collection.convert.Wrappers$JSetWrapper cannot be cast to java.util.Set
	at com.rocana.search.consumer.IndexConsumerWorker.onRebalance(IndexConsumerWorker.java:80)
	at com.rocana.search.consumer.ConsumerService$1.beforeReleasingPartitions(ConsumerService.java:244)
	at kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.kafka$consumer$ZookeeperConsumerConnector$ZKRebalancerListener$$rebalance(ZookeeperConsumerConnector.scala:675)
	at kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1$$anonfun$apply$mcV$sp$1.apply$mcVI$sp(ZookeeperConsumerConnector.scala:625)
	at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:142)
	at kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply$mcV$sp(ZookeeperConsumerConnector.scala:619)
	at kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply(ZookeeperConsumerConnector.scala:616)
	at kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply(ZookeeperConsumerConnector.scala:616)
	at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
	at kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.syncedRebalance(ZookeeperConsumerConnector.scala:615)
	at kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anon$1.run(ZookeeperConsumerConnector.scala:568)
{code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)