You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@kafka.apache.org by tao xiao <xi...@gmail.com> on 2015/04/15 23:21:08 UTC
Got NPE during partition rebalancing in high level consumer
Hi team,
I discovered an issue that when a high level consumer with roundrobin
assignment strategy consumes a topic that hasn't been created on broker a
NPE exception is thrown during partition rebalancing phase. I use Kafka
0.8.2.1
Here is the step to reproduce:
1. create a high level consumer with roundrobin
2. use connector.createMessageStreamsByFilter to create a message stream in
the consumer to a topic that yet to be created on broker
Below is the exception.
2015-04-15 14:16:46 INFO kafka.utils.Logging$class:68 -
[test12345667fffff_localhost], Committing all offsets after clearing the
fetcher queues
2015-04-15 14:16:46 INFO kafka.utils.Logging$class:68 -
[test12345667fffff_localhost], Releasing partition ownership
2015-04-15 14:16:46 INFO kafka.utils.Logging$class:76 -
[test12345667fffff_localhost], exception during rebalance
java.lang.NullPointerException
at scala.collection.mutable.HashTable$$anon$1.next(HashTable.scala:210)
at scala.collection.mutable.HashTable$$anon$1.next(HashTable.scala:202)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
at scala.collection.IterableLike$class.head(IterableLike.scala:91)
at scala.collection.AbstractIterable.head(Iterable.scala:54)
at kafka.consumer.RoundRobinAssignor.assign(PartitionAssignor.scala:75)
at kafka.consumer.RoundRobinAssignor.assign(PartitionAssignor.scala:69)
at
kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.kafka$consumer$ZookeeperConsumerConnector$ZKRebalancerListener$$rebalance(ZookeeperConsumerConnector.scala:660)
at
kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1$$anonfun$apply$mcV$sp$1.apply$mcVI$sp(ZookeeperConsumerConnector.scala:608)
at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:141)
at
kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply$mcV$sp(ZookeeperConsumerConnector.scala:602)
at
kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply(ZookeeperConsumerConnector.scala:599)
at
kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply(ZookeeperConsumerConnector.scala:599)
at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
at
kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.syncedRebalance(ZookeeperConsumerConnector.scala:598)
at
kafka.consumer.ZookeeperConsumerConnector.kafka$consumer$ZookeeperConsumerConnector$$reinitializeConsumer(ZookeeperConsumerConnector.scala:905)
at
kafka.consumer.ZookeeperConsumerConnector$WildcardStreamsHandler.<init>(ZookeeperConsumerConnector.scala:939)
at
kafka.consumer.ZookeeperConsumerConnector.createMessageStreamsByFilter(ZookeeperConsumerConnector.scala:160)
at
kafka.javaapi.consumer.ZookeeperConsumerConnector.createMessageStreamsByFilter(ZookeeperConsumerConnector.scala:101)
at
kafka.javaapi.consumer.ZookeeperConsumerConnector.createMessageStreamsByFilter(ZookeeperConsumerConnector.scala:105)
at com.ebay.kafka.demo.Consumer.main(Consumer.java:61)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at com.intellij.rt.execution.application.AppMain.main(AppMain.java:134)
--
Regards,
Tao
Re: Got NPE during partition rebalancing in high level consumer
Posted by tao xiao <xi...@gmail.com>.
Guozhang,
No, I don't think the patch of KAFKA-2056 would fix this problem. The NPE
is thrown at the line that is called before the fix executes. But I do
notice that the code in trunk did fix the issue by ensuring the size of map
returned from ctx.consumersForTopic is > 0. So the code in trunk is safe.
On Wed, Apr 15, 2015 at 3:45 PM, Guozhang Wang <wa...@gmail.com> wrote:
> Hello Tao,
>
> Do you think the solution to KAFKA-2056 will resolve this issue? It will be
> included in 0.8.3 release.
>
> Guozhang
>
> On Wed, Apr 15, 2015 at 2:21 PM, tao xiao <xi...@gmail.com> wrote:
>
> > Hi team,
> >
> > I discovered an issue that when a high level consumer with roundrobin
> > assignment strategy consumes a topic that hasn't been created on broker a
> > NPE exception is thrown during partition rebalancing phase. I use Kafka
> > 0.8.2.1
> >
> > Here is the step to reproduce:
> >
> > 1. create a high level consumer with roundrobin
> > 2. use connector.createMessageStreamsByFilter to create a message stream
> in
> > the consumer to a topic that yet to be created on broker
> >
> > Below is the exception.
> >
> > 2015-04-15 14:16:46 INFO kafka.utils.Logging$class:68 -
> > [test12345667fffff_localhost], Committing all offsets after clearing the
> > fetcher queues
> > 2015-04-15 14:16:46 INFO kafka.utils.Logging$class:68 -
> > [test12345667fffff_localhost], Releasing partition ownership
> > 2015-04-15 14:16:46 INFO kafka.utils.Logging$class:76 -
> > [test12345667fffff_localhost], exception during rebalance
> > java.lang.NullPointerException
> > at scala.collection.mutable.HashTable$$anon$1.next(HashTable.scala:210)
> > at scala.collection.mutable.HashTable$$anon$1.next(HashTable.scala:202)
> > at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
> > at scala.collection.IterableLike$class.head(IterableLike.scala:91)
> > at scala.collection.AbstractIterable.head(Iterable.scala:54)
> > at kafka.consumer.RoundRobinAssignor.assign(PartitionAssignor.scala:75)
> > at kafka.consumer.RoundRobinAssignor.assign(PartitionAssignor.scala:69)
> > at
> >
> >
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.kafka$consumer$ZookeeperConsumerConnector$ZKRebalancerListener$$rebalance(ZookeeperConsumerConnector.scala:660)
> > at
> >
> >
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1$$anonfun$apply$mcV$sp$1.apply$mcVI$sp(ZookeeperConsumerConnector.scala:608)
> > at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:141)
> > at
> >
> >
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply$mcV$sp(ZookeeperConsumerConnector.scala:602)
> > at
> >
> >
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply(ZookeeperConsumerConnector.scala:599)
> > at
> >
> >
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply(ZookeeperConsumerConnector.scala:599)
> > at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> > at
> >
> >
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.syncedRebalance(ZookeeperConsumerConnector.scala:598)
> > at
> >
> >
> kafka.consumer.ZookeeperConsumerConnector.kafka$consumer$ZookeeperConsumerConnector$$reinitializeConsumer(ZookeeperConsumerConnector.scala:905)
> > at
> >
> >
> kafka.consumer.ZookeeperConsumerConnector$WildcardStreamsHandler.<init>(ZookeeperConsumerConnector.scala:939)
> > at
> >
> >
> kafka.consumer.ZookeeperConsumerConnector.createMessageStreamsByFilter(ZookeeperConsumerConnector.scala:160)
> > at
> >
> >
> kafka.javaapi.consumer.ZookeeperConsumerConnector.createMessageStreamsByFilter(ZookeeperConsumerConnector.scala:101)
> > at
> >
> >
> kafka.javaapi.consumer.ZookeeperConsumerConnector.createMessageStreamsByFilter(ZookeeperConsumerConnector.scala:105)
> > at com.ebay.kafka.demo.Consumer.main(Consumer.java:61)
> > at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> > at
> >
> >
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
> > at
> >
> >
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> > at java.lang.reflect.Method.invoke(Method.java:606)
> > at com.intellij.rt.execution.application.AppMain.main(AppMain.java:134)
> >
> > --
> > Regards,
> > Tao
> >
>
>
>
> --
> -- Guozhang
>
--
Regards,
Tao
Re: Got NPE during partition rebalancing in high level consumer
Posted by Guozhang Wang <wa...@gmail.com>.
Hello Tao,
Do you think the solution to KAFKA-2056 will resolve this issue? It will be
included in 0.8.3 release.
Guozhang
On Wed, Apr 15, 2015 at 2:21 PM, tao xiao <xi...@gmail.com> wrote:
> Hi team,
>
> I discovered an issue that when a high level consumer with roundrobin
> assignment strategy consumes a topic that hasn't been created on broker a
> NPE exception is thrown during partition rebalancing phase. I use Kafka
> 0.8.2.1
>
> Here is the step to reproduce:
>
> 1. create a high level consumer with roundrobin
> 2. use connector.createMessageStreamsByFilter to create a message stream in
> the consumer to a topic that yet to be created on broker
>
> Below is the exception.
>
> 2015-04-15 14:16:46 INFO kafka.utils.Logging$class:68 -
> [test12345667fffff_localhost], Committing all offsets after clearing the
> fetcher queues
> 2015-04-15 14:16:46 INFO kafka.utils.Logging$class:68 -
> [test12345667fffff_localhost], Releasing partition ownership
> 2015-04-15 14:16:46 INFO kafka.utils.Logging$class:76 -
> [test12345667fffff_localhost], exception during rebalance
> java.lang.NullPointerException
> at scala.collection.mutable.HashTable$$anon$1.next(HashTable.scala:210)
> at scala.collection.mutable.HashTable$$anon$1.next(HashTable.scala:202)
> at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
> at scala.collection.IterableLike$class.head(IterableLike.scala:91)
> at scala.collection.AbstractIterable.head(Iterable.scala:54)
> at kafka.consumer.RoundRobinAssignor.assign(PartitionAssignor.scala:75)
> at kafka.consumer.RoundRobinAssignor.assign(PartitionAssignor.scala:69)
> at
>
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.kafka$consumer$ZookeeperConsumerConnector$ZKRebalancerListener$$rebalance(ZookeeperConsumerConnector.scala:660)
> at
>
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1$$anonfun$apply$mcV$sp$1.apply$mcVI$sp(ZookeeperConsumerConnector.scala:608)
> at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:141)
> at
>
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply$mcV$sp(ZookeeperConsumerConnector.scala:602)
> at
>
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply(ZookeeperConsumerConnector.scala:599)
> at
>
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply(ZookeeperConsumerConnector.scala:599)
> at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> at
>
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.syncedRebalance(ZookeeperConsumerConnector.scala:598)
> at
>
> kafka.consumer.ZookeeperConsumerConnector.kafka$consumer$ZookeeperConsumerConnector$$reinitializeConsumer(ZookeeperConsumerConnector.scala:905)
> at
>
> kafka.consumer.ZookeeperConsumerConnector$WildcardStreamsHandler.<init>(ZookeeperConsumerConnector.scala:939)
> at
>
> kafka.consumer.ZookeeperConsumerConnector.createMessageStreamsByFilter(ZookeeperConsumerConnector.scala:160)
> at
>
> kafka.javaapi.consumer.ZookeeperConsumerConnector.createMessageStreamsByFilter(ZookeeperConsumerConnector.scala:101)
> at
>
> kafka.javaapi.consumer.ZookeeperConsumerConnector.createMessageStreamsByFilter(ZookeeperConsumerConnector.scala:105)
> at com.ebay.kafka.demo.Consumer.main(Consumer.java:61)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
>
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
> at
>
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:606)
> at com.intellij.rt.execution.application.AppMain.main(AppMain.java:134)
>
> --
> Regards,
> Tao
>
--
-- Guozhang